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

@@ -18,7 +18,8 @@ import (
 	"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 +31,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.
@@ -91,11 +84,9 @@ type batchSpanProcessor struct {
 	queue   chan ReadOnlySpan
 	dropped uint32
 
-	callbackRegistration      metric.Registration
-	spansProcessedCounter     metric.Int64Counter
-	successAttributes         metric.MeasurementOption
-	alreadyShutdownAttributes metric.MeasurementOption
-	queueFullAttributes       metric.MeasurementOption
+	callbackRegistration  metric.Registration
+	spansProcessedCounter otelconv.SDKProcessorSpanProcessed
+	componentNameAttr     attribute.KeyValue
 
 	batch      []ReadOnlySpan
 	batchMutex sync.Mutex
@@ -129,17 +120,19 @@ func NewBatchSpanProcessor(exporter SpanExporter, options ...BatchSpanProcessorO
 		ExportTimeout:      time.Duration(env.BatchSpanProcessorExportTimeout(DefaultExportTimeout)) * time.Millisecond,
 		MaxQueueSize:       maxQueueSize,
 		MaxExportBatchSize: maxExportBatchSize,
+		meterProvider:      otel.GetMeterProvider(),
 	}
 	for _, opt := range options {
 		opt(&o)
 	}
 	bsp := &batchSpanProcessor{
-		e:      exporter,
-		o:      o,
-		batch:  make([]ReadOnlySpan, 0, o.MaxExportBatchSize),
-		timer:  time.NewTimer(o.BatchTimeout),
-		queue:  make(chan ReadOnlySpan, o.MaxQueueSize),
-		stopCh: make(chan struct{}),
+		e:                 exporter,
+		o:                 o,
+		batch:             make([]ReadOnlySpan, 0, o.MaxExportBatchSize),
+		timer:             time.NewTimer(o.BatchTimeout),
+		queue:             make(chan ReadOnlySpan, o.MaxQueueSize),
+		stopCh:            make(chan struct{}),
+		componentNameAttr: semconv.OTelComponentName(fmt.Sprintf("%s/%d", otelconv.ComponentTypeBatchingSpanProcessor, nextProcessorID())),
 	}
 
 	bsp.configureSelfObservability()
@@ -154,17 +147,17 @@ 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()
+	mp := bsp.o.meterProvider
 	if !x.SelfObservability.Enabled() {
 		mp = metric.MeterProvider(noop.NewMeterProvider())
 	}
@@ -173,34 +166,28 @@ func (bsp *batchSpanProcessor) configureSelfObservability() {
 		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 +205,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
 	}
 
@@ -377,7 +362,10 @@ 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)
+		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 +474,10 @@ 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)
+		bsp.spansProcessedCounter.Add(ctx, 1,
+			bsp.componentNameAttr,
+			bsp.spansProcessedCounter.AttrComponentType(otelconv.ComponentTypeBatchingSpanProcessor),
+			bsp.spansProcessedCounter.AttrErrorType(queueFull))
 		return false
 	}
 }
@@ -501,7 +492,10 @@ func (bsp *batchSpanProcessor) enqueueDrop(ctx context.Context, sd ReadOnlySpan)
 		return true
 	default:
 		atomic.AddUint32(&bsp.dropped, 1)
-		bsp.spansProcessedCounter.Add(ctx, 1, bsp.queueFullAttributes)
+		bsp.spansProcessedCounter.Add(ctx, 1,
+			bsp.componentNameAttr,
+			bsp.spansProcessedCounter.AttrComponentType(otelconv.ComponentTypeBatchingSpanProcessor),
+			bsp.spansProcessedCounter.AttrErrorType(queueFull))
 	}
 	return false
 }