add otel tracing support

appender.go

@@ -34,6 +34,7 @@ import (
 	"go.elastic.co/apm/v2"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 )
@@ -307,6 +308,7 @@ func (a *Appender) flush(ctx context.Context, bulkIndexer *BulkIndexer) error {
 	defer a.addCount(1, &a.bulkRequests, a.metrics.bulkRequests)
 
 	logger := a.config.Logger
+	var span trace.Span
 	if a.tracingEnabled() {
 		tx := a.config.Tracer.StartTransaction("docappender.flush", "output")
 		tx.Context.SetLabel("documents", n)
@@ -316,6 +318,19 @@ func (a *Appender) flush(ctx context.Context, bulkIndexer *BulkIndexer) error {
 		// Add trace IDs to logger, to associate any per-item errors
 		// below with the trace.
 		logger = logger.With(apmzap.TraceContext(ctx)...)
+	} else if a.otelTracingEnabled() {
+		// NOTE: this is missing transaction type information. How is this conveyed in otel?
+		ctx, span = a.config.OtelTracer.Start(ctx, "docappender.flush", trace.WithAttributes(
+			attribute.Int("documents", n),
+		))
+		defer span.End()
+
+		// Add trace IDs to logger, to associate any per-item errors
+		// below with the trace.
+		logger = logger.With(
+			zap.String("traceId", span.SpanContext().TraceID().String()),
+			zap.String("spanId", span.SpanContext().SpanID().String()),
+		)
 	}
 
 	var flushCtx context.Context
@@ -346,6 +361,8 @@ func (a *Appender) flush(ctx context.Context, bulkIndexer *BulkIndexer) error {
 		logger.Error("bulk indexing request failed", zap.Error(err))
 		if a.tracingEnabled() {
 			apm.CaptureError(ctx, err).Send()
+		} else if a.otelTracingEnabled() {
+			span.RecordError(err)
 		}
 
 		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
@@ -397,6 +414,9 @@ func (a *Appender) flush(ctx context.Context, bulkIndexer *BulkIndexer) error {
 		failedCount[info]++
 		if a.tracingEnabled() {
 			apm.CaptureError(ctx, errors.New(info.Error.Reason)).Send()
+		} else if a.otelTracingEnabled() {
+			// trace.SpanFromContext(ctx).RecordError(errors.New(info.Error.Reason))
+			span.RecordError(errors.New(info.Error.Reason))
 		}
 	}
 	for key, count := range failedCount {
@@ -682,9 +702,17 @@ func (a *Appender) indexFailureRate() float64 {
 
 // tracingEnabled checks whether we should be doing tracing
 func (a *Appender) tracingEnabled() bool {
+	// FIXME: remove
+	fmt.Println(a.config.Tracer != nil && a.config.Tracer.Recording())
 	return a.config.Tracer != nil && a.config.Tracer.Recording()
 }
 
+func (a *Appender) otelTracingEnabled() bool {
+	// FIXME: remove
+	fmt.Println(a.config.OtelTracer != nil)
+	return a.config.OtelTracer != nil
+}
+
 // activeLimit returns the value of GOMAXPROCS * cfg.ActiveRatio. Which limits
 // the maximum number of active indexers to a % of GOMAXPROCS.
 //

appender_test.go

@@ -42,6 +42,8 @@ import (
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zaptest/observer"
@@ -1647,6 +1649,51 @@ func TestAppenderTracing(t *testing.T) {
 	testAppenderTracing(t, 400, "failure")
 }
 
+func TestAppenderOtelTracing(t *testing.T) {
+	client := docappendertest.NewMockElasticsearchClient(t, func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(200)
+		_, result := docappendertest.DecodeBulkRequest(r)
+		json.NewEncoder(w).Encode(result)
+	})
+
+	core, observed := observer.New(zap.NewAtomicLevelAt(zapcore.DebugLevel))
+
+	exp := tracetest.NewInMemoryExporter()
+	tp := sdktrace.NewTracerProvider(
+		sdktrace.WithSyncer(exp),
+	)
+	defer tp.Shutdown(context.Background())
+
+	tr := tp.Tracer("test")
+	indexer, err := docappender.New(client, docappender.Config{
+		FlushInterval: time.Minute,
+		Logger:        zap.New(core),
+		// NOTE: Tracer must be nil to use otel tracing
+		Tracer:     nil,
+		OtelTracer: tr,
+	})
+	require.NoError(t, err)
+
+	const N = 100
+	for i := 0; i < N; i++ {
+		addMinimalDoc(t, indexer, "logs-foo-testing")
+	}
+
+	// Closing the indexer flushes enqueued documents.
+	require.NoError(t, indexer.Close(context.Background()))
+
+	require.NoError(t, tp.Shutdown(context.Background()))
+	err = exp.Shutdown(context.Background())
+	require.NoError(t, err)
+
+	spans := exp.GetSpans()
+	fmt.Println(len(spans))
+	assert.NotEmpty(t, spans)
+
+	correlatedLogs := observed.FilterFieldKey("transaction.id").All()
+	assert.NotEmpty(t, correlatedLogs)
+}
+
 func testAppenderTracing(t *testing.T, statusCode int, expectedOutcome string) {
 	client := docappendertest.NewMockElasticsearchClient(t, func(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(statusCode)

config.go

@@ -23,6 +23,7 @@ import (
 	"go.elastic.co/apm/v2"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 )
 
@@ -43,6 +44,8 @@ type Config struct {
 	// If Tracer is nil, requests will not be traced.
 	Tracer *apm.Tracer
 
+	OtelTracer trace.Tracer
+
 	// CompressionLevel holds the gzip compression level, from 0 (gzip.NoCompression)
 	// to 9 (gzip.BestCompression). Higher values provide greater compression, at a
 	// greater cost of CPU. The special value -1 (gzip.DefaultCompression) selects the