sdk/log: self observability: batch log processor metrics

CHANGELOG.md

@@ -35,6 +35,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Add experimental support for splitting metric data across multiple batches in `go.opentelemetry.io/otel/sdk/metric`.
   Set `OTEL_GO_X_METRIC_EXPORT_BATCH_SIZE=<max_size>` to enable for all periodic readers.
   See `go.opentelemetry.io/otel/sdk/metric/internal/x` for feature documentation. (#8071)
+- Add experimental observability metrics to the BatchProcessor in `go.opentelemetry.io/otel/sdk/log`. (#7124)
 - Add `WithDefaultAttributes` to `go.opentelemetry.io/otel/metric/x` to support setting default attributes on instruments. (#8135)
 - Add `Settable` to `go.opentelemetry.io/otel/metric/x` to allow reusing attribute options. (#8178)
 - Add experimental self-observability metrics in `go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp`. (#8194)

sdk/log/batch.go

@@ -11,7 +11,10 @@ import (
 	"sync/atomic"
 	"time"
 
+	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/internal/global"
+	"go.opentelemetry.io/otel/sdk/log/internal/counter"
+	"go.opentelemetry.io/otel/sdk/log/internal/observ"
 )
 
 const (
@@ -98,6 +101,9 @@ type BatchProcessor struct {
 	// stopped holds the stopped state of the BatchProcessor.
 	stopped atomic.Bool
 
+	// inst is the instrumentation for observability (nil when disabled).
+	inst *observ.BLP
+
 	noCmp [0]func() //nolint: unused  // This is indeed used.
 }
 
@@ -111,6 +117,31 @@ func NewBatchProcessor(exporter Exporter, opts ...BatchProcessorOption) *BatchPr
 		// Do not panic on nil export.
 		exporter = defaultNoopExporter
 	}
+
+	b := &BatchProcessor{
+		q:           newQueue(cfg.maxQSize.Value),
+		batchSize:   cfg.expMaxBatchSize.Value,
+		pollTrigger: make(chan struct{}, 1),
+		pollKill:    make(chan struct{}),
+	}
+
+	var err error
+	b.inst, err = observ.NewBLP(
+		counter.NextExporterID(),
+		func() int64 { return int64(b.q.Len()) },
+		int64(cfg.maxQSize.Value),
+	)
+	if err != nil {
+		otel.Handle(err)
+	}
+
+	// Wrap exporter with metrics recording if observability is enabled.
+	// This must be the innermost wrapper (closest to user exporter) to record
+	// metrics just before calling the actual exporter.
+	if b.inst != nil {
+		exporter = newMetricsExporter(exporter, b.inst)
+	}
+
 	// Order is important here. Wrap the timeoutExporter with the chunkExporter
 	// to ensure each export completes in timeout (instead of all chunked
 	// exports).
@@ -119,15 +150,9 @@ func NewBatchProcessor(exporter Exporter, opts ...BatchProcessorOption) *BatchPr
 	// appropriately on export.
 	exporter = newChunkExporter(exporter, cfg.expMaxBatchSize.Value)
 
-	b := &BatchProcessor{
-		exporter: newBufferExporter(exporter, cfg.expBufferSize.Value),
-
-		q:           newQueue(cfg.maxQSize.Value),
-		batchSize:   cfg.expMaxBatchSize.Value,
-		pollTrigger: make(chan struct{}, 1),
-		pollKill:    make(chan struct{}),
-	}
+	b.exporter = newBufferExporter(exporter, cfg.expBufferSize.Value)
 	b.pollDone = b.poll(cfg.expInterval.Value)
+
 	return b
 }
 
@@ -143,6 +168,8 @@ func (b *BatchProcessor) poll(interval time.Duration) (done chan struct{}) {
 		defer close(done)
 		defer ticker.Stop()
 
+		ctx := context.Background()
+
 		for {
 			select {
 			case <-ticker.C:
@@ -153,6 +180,9 @@ func (b *BatchProcessor) poll(interval time.Duration) (done chan struct{}) {
 			}
 
 			if d := b.q.Dropped(); d > 0 {
+				if b.inst != nil {
+					b.inst.ProcessedQueueFull(ctx, int64(d)) //nolint: gosec
+				}
 				global.Warn("dropped log records", "dropped", d)
 			}
 
@@ -225,6 +255,9 @@ func (b *BatchProcessor) Shutdown(ctx context.Context) error {
 
 	// Flush remaining queued before exporter shutdown.
 	err := b.exporter.Export(ctx, b.q.Flush())
+	if b.inst != nil {
+		err = errors.Join(err, b.inst.Shutdown())
+	}
 	return errors.Join(err, b.exporter.Shutdown(ctx))
 }
 

sdk/log/batch_test.go

@@ -20,8 +20,18 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/internal/global"
 	"go.opentelemetry.io/otel/log"
+	"go.opentelemetry.io/otel/sdk"
+	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/log/internal/counter"
+	"go.opentelemetry.io/otel/sdk/log/internal/observ"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
+	semconv "go.opentelemetry.io/otel/semconv/v1.40.0"
+	"go.opentelemetry.io/otel/semconv/v1.40.0/otelconv"
 )
 
 type concurrentBuffer struct {
@@ -673,3 +683,199 @@ func BenchmarkBatchProcessorOnEmit(b *testing.B) {
 		_ = err
 	})
 }
+
+const blpComponentID int64 = 0
+
+func TestBatchProcessorMetricsDisabled(t *testing.T) {
+	t.Setenv("OTEL_GO_X_OBSERVABILITY", "false")
+
+	counter.SetExporterID(blpComponentID)
+
+	orig := otel.GetMeterProvider()
+	t.Cleanup(func() { otel.SetMeterProvider(orig) })
+
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	otel.SetMeterProvider(mp)
+
+	e := newTestExporter(nil)
+	bp := NewBatchProcessor(
+		e,
+		WithMaxQueueSize(2),
+		WithExportMaxBatchSize(2),
+		WithExportInterval(time.Hour),
+		WithExportTimeout(time.Hour),
+	)
+	ctx := t.Context()
+
+	r := new(Record)
+	r.SetBody(log.BoolValue(true))
+	require.NoError(t, bp.OnEmit(ctx, r))
+	require.NoError(t, bp.ForceFlush(ctx))
+
+	var rm metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(ctx, &rm))
+	for _, sm := range rm.ScopeMetrics {
+		assert.NotEqual(t, observ.ScopeName, sm.Scope.Name,
+			"observ metrics should not be present when disabled")
+	}
+
+	e.Stop()
+	require.NoError(t, bp.Shutdown(ctx))
+}
+
+func TestBatchProcessorMetrics(t *testing.T) {
+	counter.SetExporterID(blpComponentID)
+
+	t.Setenv("OTEL_GO_X_OBSERVABILITY", "true")
+
+	origLogger := global.GetLogger()
+	t.Cleanup(func() { global.SetLogger(origLogger) })
+	buf := new(concurrentBuffer)
+	stdr.SetVerbosity(1)
+	global.SetLogger(stdr.New(stdlog.New(buf, "", 0)))
+
+	orig := otel.GetMeterProvider()
+	t.Cleanup(func() { otel.SetMeterProvider(orig) })
+
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	otel.SetMeterProvider(mp)
+
+	e := newTestExporter(nil)
+	e.ExportTrigger = make(chan struct{})
+	bp := NewBatchProcessor(
+		e,
+		WithMaxQueueSize(1),
+		WithExportMaxBatchSize(1),
+		WithExportInterval(time.Hour),
+		WithExportTimeout(time.Hour),
+	)
+	ctx := t.Context()
+
+	r := new(Record)
+	r.SetBody(log.BoolValue(true))
+	require.NoError(t, bp.OnEmit(ctx, r))
+	require.Eventually(t, func() bool {
+		return e.ExportN() > 0
+	}, 2*time.Second, time.Microsecond, "export not started")
+
+	assertBLPMetrics(t, reader,
+		blpQCap(1),
+		blpQSize(0),
+		blpProcessed(blpDPt(blpSet(), 1)),
+	)
+
+	require.NoError(t, bp.OnEmit(ctx, r))
+	require.Eventually(t, func() bool {
+		return len(bp.exporter.input) == cap(bp.exporter.input)
+	}, 2*time.Second, time.Microsecond, "buffer channel not filled")
+
+	require.NoError(t, bp.OnEmit(ctx, r))
+	require.NoError(t, bp.OnEmit(ctx, r))
+
+	wantMsg := `"level"=1 "msg"="dropped log records" "dropped"=1`
+	require.Eventually(t, func() bool {
+		return strings.Contains(buf.String(), wantMsg)
+	}, 2*time.Second, time.Microsecond, "drop not detected")
+
+	assertBLPMetrics(t, reader,
+		blpQCap(1),
+		blpQSize(1),
+		blpProcessed(
+			blpDPt(blpSet(), 1),
+			blpDPt(blpSet(observ.ErrQueueFull), 1),
+		),
+	)
+
+	close(e.ExportTrigger)
+	e.Stop()
+	require.NoError(t, bp.Shutdown(ctx))
+}
+
+func blpSet(attrs ...attribute.KeyValue) attribute.Set {
+	return attribute.NewSet(append([]attribute.KeyValue{
+		semconv.OTelComponentTypeBatchingLogProcessor,
+		observ.BLPComponentName(blpComponentID),
+	}, attrs...)...)
+}
+
+func blpDPt(set attribute.Set, value int64) metricdata.DataPoint[int64] {
+	return metricdata.DataPoint[int64]{Attributes: set, Value: value}
+}
+
+func blpQCap(v int64) metricdata.Metrics {
+	return metricdata.Metrics{
+		Name:        otelconv.SDKProcessorLogQueueCapacity{}.Name(),
+		Description: otelconv.SDKProcessorLogQueueCapacity{}.Description(),
+		Unit:        otelconv.SDKProcessorLogQueueCapacity{}.Unit(),
+		Data: metricdata.Sum[int64]{
+			Temporality: metricdata.CumulativeTemporality,
+			IsMonotonic: false,
+			DataPoints:  []metricdata.DataPoint[int64]{{Attributes: blpSet(), Value: v}},
+		},
+	}
+}
+
+func blpQSize(v int64) metricdata.Metrics {
+	return metricdata.Metrics{
+		Name:        otelconv.SDKProcessorLogQueueSize{}.Name(),
+		Description: otelconv.SDKProcessorLogQueueSize{}.Description(),
+		Unit:        otelconv.SDKProcessorLogQueueSize{}.Unit(),
+		Data: metricdata.Sum[int64]{
+			Temporality: metricdata.CumulativeTemporality,
+			IsMonotonic: false,
+			DataPoints:  []metricdata.DataPoint[int64]{{Attributes: blpSet(), Value: v}},
+		},
+	}
+}
+
+func blpProcessed(dPts ...metricdata.DataPoint[int64]) metricdata.Metrics {
+	return metricdata.Metrics{
+		Name:        otelconv.SDKProcessorLogProcessed{}.Name(),
+		Description: otelconv.SDKProcessorLogProcessed{}.Description(),
+		Unit:        otelconv.SDKProcessorLogProcessed{}.Unit(),
+		Data: metricdata.Sum[int64]{
+			Temporality: metricdata.CumulativeTemporality,
+			IsMonotonic: true,
+			DataPoints:  dPts,
+		},
+	}
+}
+
+func assertBLPMetrics(
+	t *testing.T,
+	reader sdkmetric.Reader,
+	wantMetrics ...metricdata.Metrics,
+) {
+	t.Helper()
+
+	var rm metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(t.Context(), &rm))
+
+	var found bool
+	var gotScope metricdata.ScopeMetrics
+	for _, sm := range rm.ScopeMetrics {
+		if sm.Scope.Name == observ.ScopeName {
+			gotScope = sm
+			found = true
+			break
+		}
+	}
+	require.True(t, found, "observ scope %q not found in collected metrics", observ.ScopeName)
+
+	metricdatatest.AssertEqual(
+		t,
+		metricdata.ScopeMetrics{
+			Scope: instrumentation.Scope{
+				Name:      observ.ScopeName,
+				Version:   sdk.Version(),
+				SchemaURL: observ.SchemaURL,
+			},
+			Metrics: wantMetrics,
+		},
+		gotScope,
+		metricdatatest.IgnoreTimestamp(),
+		metricdatatest.IgnoreExemplars(),
+	)
+}

sdk/log/exporter.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/sdk/log/internal/observ"
 )
 
 // Exporter handles the delivery of log records to external receivers.
@@ -324,3 +325,26 @@ func (e *bufferExporter) Shutdown(ctx context.Context) error {
 	}
 	return e.Exporter.Shutdown(ctx)
 }
+
+// metricsExporter wraps an Exporter to record log processing metrics
+// just before calling the wrapped exporter.
+type metricsExporter struct {
+	Exporter
+	inst *observ.BLP
+}
+
+// newMetricsExporter creates a metricsExporter that wraps the given exporter.
+func newMetricsExporter(exporter Exporter, inst *observ.BLP) Exporter {
+	return &metricsExporter{
+		Exporter: exporter,
+		inst:     inst,
+	}
+}
+
+// Export records the number of log records as a metric then forwards
+// them to the wrapped Exporter. Error returned from wrapped exporter
+// is not considered as per specification (to be measured by exporter).
+func (e *metricsExporter) Export(ctx context.Context, records []Record) error {
+	e.inst.Processed(ctx, int64(len(records)))
+	return e.Exporter.Export(ctx, records)
+}