add OTEL_GO_X_SELF_OBSERVABILITY feature gate, and otel.sdk.batch_span_processor.queue_size metric

Author: dashpole

CHANGELOG.md

@@ -10,6 +10,11 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Added
 
+- Add OTEL_GO_X_SELF_OBSERVABILITY environment variable to control whether self-observability metrics and traces are produced by SDKs.
+- Add experimental otel.sdk.batch_span_processor.queue_size metric to the trace batch span processor.
+
+### 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/internal/x/README.md

@@ -22,6 +22,13 @@ All other values are ignored.
 [OpenTelemetry resource semantic conventions]: https://opentelemetry.io/docs/specs/semconv/resource/
 [resource detectors]: https://pkg.go.dev/go.opentelemetry.io/otel/sdk/resource#Detector
 
+### SDK Self-Observability
+
+To enable experimental metric and trace instrumentation in SDKs, set the `OTEL_GO_X_SELF_OBSERVABILITY` environment variable.
+If enabled, this instrumentation uses the global `TracerProvider` and `MeterProvider`.
+The value set must be the case-insensitive string of `"true"` to enable the feature.
+All other values are ignored.
+
 #### Examples
 
 Enable experimental resource semantic conventions.

sdk/internal/x/x.go

@@ -25,6 +25,19 @@ var Resource = newFeature("RESOURCE", func(v string) (string, bool) {
 	return "", false
 })
 
+// SelfObservability is an experimental feature flag that determines if SDK
+// self-observability metrics are enabled.
+//
+// To enable this feature set the OTEL_GO_X_SELF_OBSERVABILITY environment variable
+// to the case-insensitive string value of "true" (i.e. "True" and "TRUE"
+// will also enable this).
+var SelfObservability = newFeature("SELF_OBSERVABILITY", func(v string) (string, bool) {
+	if strings.ToLower(v) == "true" {
+		return v, true
+	}
+	return "", false
+})
+
 // Feature is an experimental feature control flag. It provides a uniform way
 // to interact with these feature flags and parse their values.
 type Feature[T any] struct {

sdk/trace/batch_span_processor.go

@@ -5,13 +5,19 @@ package trace // import "go.opentelemetry.io/otel/sdk/trace"
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"go.opentelemetry.io/otel"
+	"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"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -63,8 +69,9 @@ type batchSpanProcessor struct {
 	e SpanExporter
 	o BatchSpanProcessorOptions
 
-	queue   chan ReadOnlySpan
-	dropped uint32
+	queue                chan ReadOnlySpan
+	dropped              uint32
+	callbackRegistration metric.Registration
 
 	batch      []ReadOnlySpan
 	batchMutex sync.Mutex
@@ -111,6 +118,8 @@ func NewBatchSpanProcessor(exporter SpanExporter, options ...BatchSpanProcessorO
 		stopCh: make(chan struct{}),
 	}
 
+	bsp.configureSelfObservability()
+
 	bsp.stopWait.Add(1)
 	go func() {
 		defer bsp.stopWait.Done()
@@ -121,6 +130,38 @@ func NewBatchSpanProcessor(exporter SpanExporter, options ...BatchSpanProcessorO
 	return bsp
 }
 
+func (bsp *batchSpanProcessor) configureSelfObservability() {
+	mp := otel.GetMeterProvider()
+	if !x.SelfObservability.Enabled() {
+		mp = metric.MeterProvider(noop.NewMeterProvider())
+	}
+	meter := mp.Meter(
+		selfObsScopeName,
+		metric.WithInstrumentationVersion(version()),
+	)
+
+	queueSizeCounter, err := meter.Int64ObservableUpDownCounter("otel.sdk.span.processor.queue_size",
+		metric.WithUnit("{span}"),
+		metric.WithDescription("The number of spans in the queue of a given instance of an SDK span processor."),
+	)
+	if err != nil {
+		otel.Handle(err)
+	}
+
+	attrsOpt := metric.WithAttributes(
+		attribute.String("otel.sdk.component.name", fmt.Sprintf("batching_span_processor/%p", bsp)),
+	)
+	bsp.callbackRegistration, err = meter.RegisterCallback(
+		func(ctx context.Context, o metric.Observer) error {
+			o.ObserveInt64(queueSizeCounter, int64(len(bsp.queue)), attrsOpt)
+			return nil
+		},
+		queueSizeCounter)
+	if err != nil {
+		otel.Handle(err)
+	}
+}
+
 // OnStart method does nothing.
 func (bsp *batchSpanProcessor) OnStart(parent context.Context, s ReadWriteSpan) {}
 
@@ -162,7 +203,7 @@ func (bsp *batchSpanProcessor) Shutdown(ctx context.Context) error {
 			err = ctx.Err()
 		}
 	})
-	return err
+	return errors.Join(err, bsp.callbackRegistration.Unregister())
 }
 
 type forceFlushSpan struct {

sdk/trace/provider.go

@@ -20,6 +20,7 @@ import (
 
 const (
 	defaultTracerName = "go.opentelemetry.io/otel/sdk/tracer"
+	selfObsScopeName  = "go.opentelemetry.io/otel/sdk/trace"
 )
 
 // tracerProviderConfig.