WIP

Author: dashpole

CHANGELOG.md

@@ -42,6 +42,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Add `ErrorType` attribute helper function to the `go.opentelmetry.io/otel/semconv/v1.34.0` package. (#6962)
 - Add `WithAllowKeyDuplication` in `go.opentelemetry.io/otel/sdk/log` which can be used to disable deduplication for log records. (#6968)
 - Add `Clone` method to `Record` in `go.opentelemetry.io/otel/log` that returns a copy of the record with no shared state. (#7001)
+- Add `OTEL_GO_X_SELF_OBSERVABILITY` environment variable to control whether self-observability metrics and traces are produced by SDKs. (#6393)
+- Add experimental `otel.sdk.processor.span.queue.size`, `otel.sdk.processor.span.queue.capacity`, and `otel.sdk.processor.span.processed.count` metrics to the trace batch span processor in `go.opentelemetry.io/otel/sdk/trace`. (#6393)
 
 ### Changed
 
@@ -163,11 +165,6 @@ The next release will require at least [Go 1.23].
 
 ### Added
 
-- Add `OTEL_GO_X_SELF_OBSERVABILITY` environment variable to control whether self-observability metrics and traces are produced by SDKs. (#TODO)
-- Add experimental `otel.sdk.processor.span.queue.size`, `otel.sdk.processor.span.queue.capacity`, and `otel.sdk.processor.span.processed.count` metrics to the trace batch span processor in `go.opentelemetry.io/otel/sdk/trace`. (#TODO)
-
-### Fixed
-
 - Add `ValueFromAttribute` and `KeyValueFromAttribute` in `go.opentelemetry.io/otel/log`. (#6180)
 - Add `EventName` and `SetEventName` to `Record` in `go.opentelemetry.io/otel/log`. (#6187)
 - Add `EventName` to `RecordFactory` in `go.opentelemetry.io/otel/log/logtest`. (#6187)

sdk/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/stretchr/testify v1.10.0
 	go.opentelemetry.io/otel v1.37.0
 	go.opentelemetry.io/otel/metric v1.37.0
+	go.opentelemetry.io/otel/sdk/metric v1.34.0
 	go.opentelemetry.io/otel/trace v1.37.0
 	go.uber.org/goleak v1.3.0
 	golang.org/x/sys v0.34.0

sdk/go.sum

@@ -21,6 +21,8 @@ github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOf
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
 go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
+go.opentelemetry.io/otel/sdk/metric v1.34.0 h1:5CeK9ujjbFVL5c1PhLuStg1wxA7vQv7ce1EK0Gyvahk=
+go.opentelemetry.io/otel/sdk/metric v1.34.0/go.mod h1:jQ/r8Ze28zRKoNRdkjCZxfs6YvBTG1+YIqyFVFYec5w=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 golang.org/x/sys v0.34.0 h1:H5Y5sJ2L2JRdyv7ROF1he/lPdvFsd0mJHFw2ThKHxLA=

sdk/trace/batch_span_processor.go

@@ -15,10 +15,10 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/internal/global"
 	"go.opentelemetry.io/otel/metric"
-	"go.opentelemetry.io/otel/metric/noop"
 	"go.opentelemetry.io/otel/sdk/internal/env"
 	"go.opentelemetry.io/otel/sdk/internal/x"
-	semconv "go.opentelemetry.io/otel/semconv/v1.30.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.34.0"
+	"go.opentelemetry.io/otel/semconv/v1.34.0/otelconv"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -30,19 +30,11 @@ const (
 	// DefaultExportTimeout is the duration after which an export is cancelled, in milliseconds.
 	DefaultExportTimeout      = 30000
 	DefaultMaxExportBatchSize = 512
-
-	queueSizeMetricName             = "otel.sdk.processor.span.queue.size"
-	queueSizeMetricDescription      = "The number of spans in the queue of a given instance of an SDK span processor"
-	queueCapacityMetricName         = "otel.sdk.processor.span.queue.capacity"
-	queueCapacityMetricDescription  = "The maximum number of spans the queue of a given instance of an SDK span processor can hold"
-	spansProcessedMetricName        = "otel.sdk.processor.span.processed.count"
-	spansProcessedMetricDescription = "The number of spans for which the processing has finished, either successful or failed"
-	spanCountUnit                   = "{span}"
 )
 
 var (
-	componentTypeKey = attribute.Key("otel.component.type")
-	componentNameKey = attribute.Key("otel.component.name")
+	noError   = otelconv.ErrorTypeAttr("")
+	queueFull = otelconv.ErrorTypeAttr("queue_full")
 )
 
 // BatchSpanProcessorOption configures a BatchSpanProcessor.
@@ -77,9 +69,6 @@ type BatchSpanProcessorOptions struct {
 	// Blocking option should be used carefully as it can severely affect the performance of an
 	// application.
 	BlockOnQueueFull bool
-
-	// meterProvider is the meterProvider used to record self-observability metrics.
-	meterProvider metric.MeterProvider
 }
 
 // batchSpanProcessor is a SpanProcessor that batches asynchronously-received
@@ -91,11 +80,10 @@ type batchSpanProcessor struct {
 	queue   chan ReadOnlySpan
 	dropped uint32
 
-	callbackRegistration      metric.Registration
-	spansProcessedCounter     metric.Int64Counter
-	successAttributes         metric.MeasurementOption
-	alreadyShutdownAttributes metric.MeasurementOption
-	queueFullAttributes       metric.MeasurementOption
+	selfObservabilityEnabled bool
+	callbackRegistration     metric.Registration
+	spansProcessedCounter    otelconv.SDKProcessorSpanProcessed
+	componentNameAttr        attribute.KeyValue
 
 	batch      []ReadOnlySpan
 	batchMutex sync.Mutex
@@ -154,53 +142,48 @@ func NewBatchSpanProcessor(exporter SpanExporter, options ...BatchSpanProcessorO
 	return bsp
 }
 
-var processorID atomic.Uint64
+var processorID atomic.Int64
 
 // nextProcessorID returns an identifier for this batch span processor,
 // starting with 0 and incrementing by 1 each time it is called.
 func nextProcessorID() int64 {
-	return int64(processorID.Add(1) - 1)
+	return processorID.Add(1) - 1
 }
 
 // configureSelfObservability configures metrics for the batch span processor.
 func (bsp *batchSpanProcessor) configureSelfObservability() {
-	mp := otel.GetMeterProvider()
 	if !x.SelfObservability.Enabled() {
-		mp = metric.MeterProvider(noop.NewMeterProvider())
+		return
 	}
-	meter := mp.Meter(
+	bsp.selfObservabilityEnabled = true
+	bsp.componentNameAttr = semconv.OTelComponentName(fmt.Sprintf("%s/%d", otelconv.ComponentTypeBatchingSpanProcessor, nextProcessorID()))
+	meter := otel.GetMeterProvider().Meter(
 		selfObsScopeName,
 		metric.WithInstrumentationVersion(version()),
 	)
 
-	queueCapacityUpDownCounter, err := meter.Int64ObservableUpDownCounter(queueCapacityMetricName,
-		metric.WithUnit(spanCountUnit),
-		metric.WithDescription(queueCapacityMetricDescription),
+	// TODO(#7029): switch to otelconv.NewSDKProcessorSpanQueueCapacity after asynchronous instruments are supported.
+	queueCapacityUpDownCounter, err := meter.Int64ObservableUpDownCounter(otelconv.SDKProcessorSpanQueueCapacity{}.Name(),
+		metric.WithUnit(otelconv.SDKProcessorSpanQueueCapacity{}.Unit()),
+		metric.WithDescription(otelconv.SDKProcessorSpanQueueCapacity{}.Description()),
 	)
 	if err != nil {
 		otel.Handle(err)
 	}
-	queueSizeUpDownCounter, err := meter.Int64ObservableUpDownCounter(queueSizeMetricName,
-		metric.WithUnit(spanCountUnit),
-		metric.WithDescription(queueSizeMetricDescription),
+	// TODO(#7029): switch to otelconv.NewSDKProcessorSpanQueueSize after asynchronous instruments are supported.
+	queueSizeUpDownCounter, err := meter.Int64ObservableUpDownCounter(otelconv.SDKProcessorSpanQueueSize{}.Name(),
+		metric.WithUnit(otelconv.SDKProcessorSpanQueueSize{}.Unit()),
+		metric.WithDescription(otelconv.SDKProcessorSpanQueueSize{}.Description()),
 	)
 	if err != nil {
 		otel.Handle(err)
 	}
-	bsp.spansProcessedCounter, err = meter.Int64Counter(spansProcessedMetricName,
-		metric.WithUnit(spanCountUnit),
-		metric.WithDescription(spansProcessedMetricDescription),
-	)
+	bsp.spansProcessedCounter, err = otelconv.NewSDKProcessorSpanProcessed(meter)
 	if err != nil {
 		otel.Handle(err)
 	}
 
-	componentTypeAttr := componentTypeKey.String("batching_span_processor")
-	componentNameAttr := componentNameKey.String(fmt.Sprintf("batching_span_processor/%d", nextProcessorID()))
-	bsp.successAttributes = metric.WithAttributes(componentNameAttr, componentTypeAttr, semconv.ErrorTypeKey.String(""))
-	bsp.alreadyShutdownAttributes = metric.WithAttributes(componentNameAttr, componentTypeAttr, semconv.ErrorTypeKey.String("already_shutdown"))
-	bsp.queueFullAttributes = metric.WithAttributes(componentNameAttr, componentTypeAttr, semconv.ErrorTypeKey.String("queue_full"))
-	callabckAttributesOpt := metric.WithAttributes(componentNameAttr, componentTypeAttr)
+	callabckAttributesOpt := metric.WithAttributes(bsp.componentNameAttr, semconv.OTelComponentTypeBatchingSpanProcessor)
 	bsp.callbackRegistration, err = meter.RegisterCallback(
 		func(ctx context.Context, o metric.Observer) error {
 			o.ObserveInt64(queueSizeUpDownCounter, int64(len(bsp.queue)), callabckAttributesOpt)
@@ -218,10 +201,8 @@ func (bsp *batchSpanProcessor) OnStart(parent context.Context, s ReadWriteSpan)
 
 // OnEnd method enqueues a ReadOnlySpan for later processing.
 func (bsp *batchSpanProcessor) OnEnd(s ReadOnlySpan) {
-	ctx := context.Background()
 	// Do not enqueue spans after Shutdown.
 	if bsp.stopped.Load() {
-		bsp.spansProcessedCounter.Add(ctx, 1, bsp.alreadyShutdownAttributes)
 		return
 	}
 
@@ -256,7 +237,10 @@ func (bsp *batchSpanProcessor) Shutdown(ctx context.Context) error {
 			err = ctx.Err()
 		}
 	})
-	return errors.Join(err, bsp.callbackRegistration.Unregister())
+	if bsp.selfObservabilityEnabled {
+		err = errors.Join(err, bsp.callbackRegistration.Unregister())
+	}
+	return err
 }
 
 type forceFlushSpan struct {
@@ -352,16 +336,6 @@ func WithBlocking() BatchSpanProcessorOption {
 	}
 }
 
-// withMeterProvider allows configuring the meterProvider used for recording
-// self-observability metrics during testing.
-func withMeterProvider(provider metric.MeterProvider) BatchSpanProcessorOption {
-	return func(o *BatchSpanProcessorOptions) {
-		if provider != nil {
-			o.meterProvider = provider
-		}
-	}
-}
-
 // exportSpans is a subroutine of processing and draining the queue.
 func (bsp *batchSpanProcessor) exportSpans(ctx context.Context) error {
 	bsp.timer.Reset(bsp.o.BatchTimeout)
@@ -377,7 +351,12 @@ func (bsp *batchSpanProcessor) exportSpans(ctx context.Context) error {
 
 	if l := len(bsp.batch); l > 0 {
 		global.Debug("exporting spans", "count", len(bsp.batch), "total_dropped", atomic.LoadUint32(&bsp.dropped))
-		bsp.spansProcessedCounter.Add(ctx, int64(len(bsp.batch)), bsp.successAttributes)
+		if bsp.selfObservabilityEnabled {
+			bsp.spansProcessedCounter.Add(ctx, int64(l),
+				bsp.componentNameAttr,
+				bsp.spansProcessedCounter.AttrComponentType(otelconv.ComponentTypeBatchingSpanProcessor),
+				bsp.spansProcessedCounter.AttrErrorType(noError))
+		}
 		err := bsp.e.ExportSpans(ctx, bsp.batch)
 
 		// A new batch is always created after exporting, even if the batch failed to be exported.
@@ -486,7 +465,12 @@ func (bsp *batchSpanProcessor) enqueueBlockOnQueueFull(ctx context.Context, sd R
 	case bsp.queue <- sd:
 		return true
 	case <-ctx.Done():
-		bsp.spansProcessedCounter.Add(ctx, 1, bsp.queueFullAttributes)
+		if bsp.selfObservabilityEnabled {
+			bsp.spansProcessedCounter.Add(ctx, 1,
+				bsp.componentNameAttr,
+				bsp.spansProcessedCounter.AttrComponentType(otelconv.ComponentTypeBatchingSpanProcessor),
+				bsp.spansProcessedCounter.AttrErrorType(queueFull))
+		}
 		return false
 	}
 }
@@ -501,7 +485,12 @@ func (bsp *batchSpanProcessor) enqueueDrop(ctx context.Context, sd ReadOnlySpan)
 		return true
 	default:
 		atomic.AddUint32(&bsp.dropped, 1)
-		bsp.spansProcessedCounter.Add(ctx, 1, bsp.queueFullAttributes)
+		if bsp.selfObservabilityEnabled {
+			bsp.spansProcessedCounter.Add(ctx, 1,
+				bsp.componentNameAttr,
+				bsp.spansProcessedCounter.AttrComponentType(otelconv.ComponentTypeBatchingSpanProcessor),
+				bsp.spansProcessedCounter.AttrErrorType(queueFull))
+		}
 	}
 	return false
 }