Refactor BSP observability setup (#7264)

MrAlias · web-flow · commit be1e57fb3651 · 2025-08-29T08:05:29.000-07:00
Do not rely on side-effects from the configureSelfObservability method.
Instead, initialize with the new pure newBSPObs func.
diff --git a/sdk/trace/batch_span_processor.go b/sdk/trace/batch_span_processor.go
@@ -124,7 +124,20 @@ func NewBatchSpanProcessor(exporter SpanExporter, options ...BatchSpanProcessorO
 		stopCh: make(chan struct{}),
 	}
 
-	bsp.configureSelfObservability()
+	if x.SelfObservability.Enabled() {
+		bsp.selfObservabilityEnabled = true
+		bsp.componentNameAttr = componentName()
+
+		var err error
+		bsp.spansProcessedCounter, bsp.callbackRegistration, err = newBSPObs(
+			bsp.componentNameAttr,
+			func() int64 { return int64(len(bsp.queue)) },
+			int64(bsp.o.MaxQueueSize),
+		)
+		if err != nil {
+			otel.Handle(err)
+		}
+	}
 
 	bsp.stopWait.Add(1)
 	go func() {
@@ -144,45 +157,49 @@ func nextProcessorID() int64 {
 	return processorIDCounter.Add(1) - 1
 }
 
-// configureSelfObservability configures metrics for the batch span processor.
-func (bsp *batchSpanProcessor) configureSelfObservability() {
-	if !x.SelfObservability.Enabled() {
-		return
-	}
-	bsp.selfObservabilityEnabled = true
-	bsp.componentNameAttr = semconv.OTelComponentName(
-		fmt.Sprintf("%s/%d", otelconv.ComponentTypeBatchingSpanProcessor, nextProcessorID()))
+func componentName() attribute.KeyValue {
+	id := nextProcessorID()
+	name := fmt.Sprintf("%s/%d", otelconv.ComponentTypeBatchingSpanProcessor, id)
+	return semconv.OTelComponentName(name)
+}
+
+// newBSPObs creates and returns a new set of metrics instruments and a
+// registration for a BatchSpanProcessor. It is the caller's responsibility
+// to unregister the registration when it is no longer needed.
+func newBSPObs(
+	cmpnt attribute.KeyValue,
+	qLen func() int64,
+	qMax int64,
+) (otelconv.SDKProcessorSpanProcessed, metric.Registration, error) {
 	meter := otel.GetMeterProvider().Meter(
 		selfObsScopeName,
 		metric.WithInstrumentationVersion(sdk.Version()),
 		metric.WithSchemaURL(semconv.SchemaURL),
 	)
 
-	queueCapacityUpDownCounter, err := otelconv.NewSDKProcessorSpanQueueCapacity(meter)
-	if err != nil {
-		otel.Handle(err)
-	}
-	queueSizeUpDownCounter, err := otelconv.NewSDKProcessorSpanQueueSize(meter)
-	if err != nil {
-		otel.Handle(err)
-	}
-	bsp.spansProcessedCounter, err = otelconv.NewSDKProcessorSpanProcessed(meter)
-	if err != nil {
-		otel.Handle(err)
-	}
+	qCap, err := otelconv.NewSDKProcessorSpanQueueCapacity(meter)
+
+	qSize, e := otelconv.NewSDKProcessorSpanQueueSize(meter)
+	err = errors.Join(err, e)
+
+	spansProcessed, e := otelconv.NewSDKProcessorSpanProcessed(meter)
+	err = errors.Join(err, e)
 
-	callbackAttributesOpt := metric.WithAttributes(bsp.componentNameAttr,
-		semconv.OTelComponentTypeBatchingSpanProcessor)
-	bsp.callbackRegistration, err = meter.RegisterCallback(
+	cmpntT := semconv.OTelComponentTypeBatchingSpanProcessor
+	attrs := metric.WithAttributes(cmpnt, cmpntT)
+
+	reg, e := meter.RegisterCallback(
 		func(_ context.Context, o metric.Observer) error {
-			o.ObserveInt64(queueSizeUpDownCounter.Inst(), int64(len(bsp.queue)), callbackAttributesOpt)
-			o.ObserveInt64(queueCapacityUpDownCounter.Inst(), int64(bsp.o.MaxQueueSize), callbackAttributesOpt)
+			o.ObserveInt64(qSize.Inst(), qLen(), attrs)
+			o.ObserveInt64(qCap.Inst(), qMax, attrs)
 			return nil
 		},
-		queueSizeUpDownCounter.Inst(), queueCapacityUpDownCounter.Inst())
-	if err != nil {
-		otel.Handle(err)
-	}
+		qSize.Inst(),
+		qCap.Inst(),
+	)
+	err = errors.Join(err, e)
+
+	return spansProcessed, reg, err
 }
 
 // OnStart method does nothing.
