Instrument obsreport.receiver metrics with otel go

CHANGELOG.md

@@ -30,7 +30,8 @@
 - Add config marshaler (#5566)
 - Add semantic conventions for specification v1.10-v1.13 (#6213)
 - `receiver/otlp`: Make logs related to gRCPC and HTTP server startup clearer (#6174)
-- Add prometheus metric prefix and constant service attributes to Collector's own telemetry when using OpenTelemetry for internal telemetry (#6223)
+- Add prometheus metric prefix to Collector's own telemetry when using OpenTelemetry for internal telemetry (#6223)
+- Instrument `obsreport.Receiver` metrics with otel-go (#6222)
 
 ## v0.61.0 Beta
 

internal/obsreportconfig/obsreportconfig.go

@@ -20,9 +20,30 @@ import (
 	"go.opencensus.io/tag"
 
 	"go.opentelemetry.io/collector/config/configtelemetry"
+	"go.opentelemetry.io/collector/featuregate"
 	"go.opentelemetry.io/collector/internal/obsreportconfig/obsmetrics"
 )
 
+const (
+	// UseOtelForInternalMetricsfeatureGateID is the feature gate ID that controls whether the collector uses open
+	// telemetrySettings for internal metrics.
+	UseOtelForInternalMetricsfeatureGateID = "telemetry.useOtelForInternalMetrics"
+)
+
+func init() {
+	// register feature gate
+	RegisterInternalMetricFeatureGate(featuregate.GetRegistry())
+}
+
+// RegisterInternalMetricFeatureGate registers the Internal Metric feature gate to the passed in registry
+func RegisterInternalMetricFeatureGate(registry *featuregate.Registry) {
+	registry.MustRegister(featuregate.Gate{
+		ID:          UseOtelForInternalMetricsfeatureGateID,
+		Description: "controls whether the collector to uses OpenTelemetry for internal metrics",
+		Enabled:     false,
+	})
+}
+
 // ObsMetrics wraps OpenCensus View for Collector observability metrics
 type ObsMetrics struct {
 	Views []*view.View

obsreport/obsreport_receiver.go

@@ -20,11 +20,18 @@ import (
 	"go.opencensus.io/stats"
 	"go.opencensus.io/tag"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/metric/instrument"
+	"go.opentelemetry.io/otel/metric/instrument/syncint64"
+	"go.opentelemetry.io/otel/metric/unit"
 	"go.opentelemetry.io/otel/trace"
+	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/configtelemetry"
+	"go.opentelemetry.io/collector/featuregate"
+	"go.opentelemetry.io/collector/internal/obsreportconfig"
 	"go.opentelemetry.io/collector/internal/obsreportconfig/obsmetrics"
 )
 
@@ -36,6 +43,18 @@ type Receiver struct {
 	longLivedCtx   bool
 	mutators       []tag.Mutator
 	tracer         trace.Tracer
+	meter          metric.Meter
+	logger         *zap.Logger
+
+	useOtelForMetrics bool
+	otelAttrs         []attribute.KeyValue
+
+	acceptedSpansCounter        syncint64.Counter
+	refusedSpansCounter         syncint64.Counter
+	acceptedMetricPointsCounter syncint64.Counter
+	refusedMetricPointsCounter  syncint64.Counter
+	acceptedLogRecordsCounter   syncint64.Counter
+	refusedLogRecordsCounter    syncint64.Counter
 }
 
 // ReceiverSettings are settings for creating an Receiver.
@@ -53,7 +72,8 @@ type ReceiverSettings struct {
 
 // NewReceiver creates a new Receiver.
 func NewReceiver(cfg ReceiverSettings) *Receiver {
-	return &Receiver{
+
+	rec := &Receiver{
 		level:          cfg.ReceiverCreateSettings.TelemetrySettings.MetricsLevel,
 		spanNamePrefix: obsmetrics.ReceiverPrefix + cfg.ReceiverID.String(),
 		transport:      cfg.Transport,
@@ -63,7 +83,74 @@ func NewReceiver(cfg ReceiverSettings) *Receiver {
 			tag.Upsert(obsmetrics.TagKeyTransport, cfg.Transport, tag.WithTTL(tag.TTLNoPropagation)),
 		},
 		tracer: cfg.ReceiverCreateSettings.TracerProvider.Tracer(cfg.ReceiverID.String()),
+		meter:  cfg.ReceiverCreateSettings.MeterProvider.Meter(cfg.ReceiverID.String()),
+		logger: cfg.ReceiverCreateSettings.Logger,
+
+		useOtelForMetrics: featuregate.GetRegistry().IsEnabled(obsreportconfig.UseOtelForInternalMetricsfeatureGateID),
+		otelAttrs: []attribute.KeyValue{
+			attribute.String(obsmetrics.ReceiverKey, cfg.ReceiverID.String()),
+			attribute.String(obsmetrics.TransportKey, cfg.Transport),
+		},
+	}
+
+	rec.createOtelMetrics()
+
+	return rec
+}
+
+func (rec *Receiver) createOtelMetrics() {
+	if !rec.useOtelForMetrics {
+		return
 	}
+
+	var err error
+	handleError := func(metricName string, err error) {
+		if err != nil {
+			rec.logger.Warn("failed to create otel instrument", zap.Error(err), zap.String("metric", metricName))
+		}
+	}
+
+	rec.acceptedSpansCounter, err = rec.meter.SyncInt64().Counter(
+		obsmetrics.ReceiverPrefix+obsmetrics.AcceptedSpansKey,
+		instrument.WithDescription("Number of spans successfully pushed into the pipeline."),
+		instrument.WithUnit(unit.Dimensionless),
+	)
+	handleError(obsmetrics.ReceiverPrefix+obsmetrics.AcceptedSpansKey, err)
+
+	rec.refusedSpansCounter, err = rec.meter.SyncInt64().Counter(
+		obsmetrics.ReceiverPrefix+obsmetrics.RefusedSpansKey,
+		instrument.WithDescription("Number of spans that could not be pushed into the pipeline."),
+		instrument.WithUnit(unit.Dimensionless),
+	)
+	handleError(obsmetrics.ReceiverPrefix+obsmetrics.RefusedSpansKey, err)
+
+	rec.acceptedMetricPointsCounter, err = rec.meter.SyncInt64().Counter(
+		obsmetrics.ReceiverPrefix+obsmetrics.AcceptedMetricPointsKey,
+		instrument.WithDescription("Number of metric points successfully pushed into the pipeline."),
+		instrument.WithUnit(unit.Dimensionless),
+	)
+	handleError(obsmetrics.ReceiverPrefix+obsmetrics.AcceptedMetricPointsKey, err)
+
+	rec.refusedMetricPointsCounter, err = rec.meter.SyncInt64().Counter(
+		obsmetrics.ReceiverPrefix+obsmetrics.RefusedMetricPointsKey,
+		instrument.WithDescription("Number of metric points that could not be pushed into the pipeline."),
+		instrument.WithUnit(unit.Dimensionless),
+	)
+	handleError(obsmetrics.ReceiverPrefix+obsmetrics.RefusedMetricPointsKey, err)
+
+	rec.acceptedLogRecordsCounter, err = rec.meter.SyncInt64().Counter(
+		obsmetrics.ReceiverPrefix+obsmetrics.AcceptedLogRecordsKey,
+		instrument.WithDescription("Number of log records successfully pushed into the pipeline."),
+		instrument.WithUnit(unit.Dimensionless),
+	)
+	handleError(obsmetrics.ReceiverPrefix+obsmetrics.AcceptedLogRecordsKey, err)
+
+	rec.refusedLogRecordsCounter, err = rec.meter.SyncInt64().Counter(
+		obsmetrics.ReceiverPrefix+obsmetrics.RefusedLogRecordsKey,
+		instrument.WithDescription("Number of log records that could not be pushed into the pipeline."),
+		instrument.WithUnit(unit.Dimensionless),
+	)
+	handleError(obsmetrics.ReceiverPrefix+obsmetrics.RefusedLogRecordsKey, err)
 }
 
 // StartTracesOp is called when a request is received from a client.
@@ -163,23 +250,7 @@ func (rec *Receiver) endOp(
 	span := trace.SpanFromContext(receiverCtx)
 
 	if rec.level != configtelemetry.LevelNone {
-		var acceptedMeasure, refusedMeasure *stats.Int64Measure
-		switch dataType {
-		case config.TracesDataType:
-			acceptedMeasure = obsmetrics.ReceiverAcceptedSpans
-			refusedMeasure = obsmetrics.ReceiverRefusedSpans
-		case config.MetricsDataType:
-			acceptedMeasure = obsmetrics.ReceiverAcceptedMetricPoints
-			refusedMeasure = obsmetrics.ReceiverRefusedMetricPoints
-		case config.LogsDataType:
-			acceptedMeasure = obsmetrics.ReceiverAcceptedLogRecords
-			refusedMeasure = obsmetrics.ReceiverRefusedLogRecords
-		}
-
-		stats.Record(
-			receiverCtx,
-			acceptedMeasure.M(int64(numAccepted)),
-			refusedMeasure.M(int64(numRefused)))
+		rec.recordMetrics(receiverCtx, dataType, numAccepted, numRefused)
 	}
 
 	// end span according to errors
@@ -206,3 +277,49 @@ func (rec *Receiver) endOp(
 	}
 	span.End()
 }
+
+func (rec *Receiver) recordMetrics(receiverCtx context.Context, dataType config.DataType, numAccepted, numRefused int) {
+	if rec.useOtelForMetrics {
+		rec.recordWithOtel(receiverCtx, dataType, numAccepted, numRefused)
+	} else {
+		rec.recordWithOC(receiverCtx, dataType, numAccepted, numRefused)
+	}
+}
+
+func (rec *Receiver) recordWithOtel(receiverCtx context.Context, dataType config.DataType, numAccepted, numRefused int) {
+	var acceptedMeasure, refusedMeasure syncint64.Counter
+	switch dataType {
+	case config.TracesDataType:
+		acceptedMeasure = rec.acceptedSpansCounter
+		refusedMeasure = rec.refusedSpansCounter
+	case config.MetricsDataType:
+		acceptedMeasure = rec.acceptedMetricPointsCounter
+		refusedMeasure = rec.refusedMetricPointsCounter
+	case config.LogsDataType:
+		acceptedMeasure = rec.acceptedLogRecordsCounter
+		refusedMeasure = rec.refusedLogRecordsCounter
+	}
+
+	acceptedMeasure.Add(receiverCtx, int64(numAccepted), rec.otelAttrs...)
+	refusedMeasure.Add(receiverCtx, int64(numRefused), rec.otelAttrs...)
+}
+
+func (rec *Receiver) recordWithOC(receiverCtx context.Context, dataType config.DataType, numAccepted, numRefused int) {
+	var acceptedMeasure, refusedMeasure *stats.Int64Measure
+	switch dataType {
+	case config.TracesDataType:
+		acceptedMeasure = obsmetrics.ReceiverAcceptedSpans
+		refusedMeasure = obsmetrics.ReceiverRefusedSpans
+	case config.MetricsDataType:
+		acceptedMeasure = obsmetrics.ReceiverAcceptedMetricPoints
+		refusedMeasure = obsmetrics.ReceiverRefusedMetricPoints
+	case config.LogsDataType:
+		acceptedMeasure = obsmetrics.ReceiverAcceptedLogRecords
+		refusedMeasure = obsmetrics.ReceiverRefusedLogRecords
+	}
+
+	stats.Record(
+		receiverCtx,
+		acceptedMeasure.M(int64(numAccepted)),
+		refusedMeasure.M(int64(numRefused)))
+}

service/collector_test.go

@@ -38,6 +38,7 @@ import (
 	"go.opentelemetry.io/collector/confmap"
 	"go.opentelemetry.io/collector/extension/zpagesextension"
 	"go.opentelemetry.io/collector/featuregate"
+	"go.opentelemetry.io/collector/internal/obsreportconfig"
 	"go.opentelemetry.io/collector/internal/testutil"
 )
 
@@ -410,9 +411,9 @@ func TestCollectorStartWithOpenTelemetryMetrics(t *testing.T) {
 	for _, tc := range ownMetricsTestCases("test version") {
 		t.Run(tc.name, func(t *testing.T) {
 			registry := featuregate.NewRegistry()
-			registerInternalMetricFeatureGate(registry)
+			obsreportconfig.RegisterInternalMetricFeatureGate(registry)
 			colTel := newColTelemetry(registry)
-			require.NoError(t, colTel.registry.Apply(map[string]bool{useOtelForInternalMetricsfeatureGateID: true}))
+			require.NoError(t, colTel.registry.Apply(map[string]bool{obsreportconfig.UseOtelForInternalMetricsfeatureGateID: true}))
 			testCollectorStartHelper(t, colTel, tc)
 		})
 	}

service/telemetry.go

@@ -53,10 +53,6 @@ const (
 	zapKeyTelemetryAddress = "address"
 	zapKeyTelemetryLevel   = "level"
 
-	// useOtelForInternalMetricsfeatureGateID is the feature gate ID that controls whether the collector uses open
-	// telemetrySettings for internal metrics.
-	useOtelForInternalMetricsfeatureGateID = "telemetry.useOtelForInternalMetrics"
-
 	// supported trace propagators
 	traceContextPropagator = "tracecontext"
 	b3Propagator           = "b3"
@@ -77,20 +73,6 @@ type telemetryInitializer struct {
 	doInitOnce sync.Once
 }
 
-func init() {
-	// register feature gate
-	registerInternalMetricFeatureGate(featuregate.GetRegistry())
-}
-
-// registerInternalMetricFeatureGate registers the Internal Metric feature gate to the passed in registry
-func registerInternalMetricFeatureGate(registry *featuregate.Registry) {
-	registry.MustRegister(featuregate.Gate{
-		ID:          useOtelForInternalMetricsfeatureGateID,
-		Description: "controls whether the collector to uses OpenTelemetry for internal metrics",
-		Enabled:     false,
-	})
-}
-
 func newColTelemetry(registry *featuregate.Registry) *telemetryInitializer {
 	return &telemetryInitializer{
 		registry: registry,
@@ -131,7 +113,7 @@ func (tel *telemetryInitializer) initOnce(buildInfo component.BuildInfo, logger
 
 	var pe http.Handler
 	var err error
-	if tel.registry.IsEnabled(useOtelForInternalMetricsfeatureGateID) {
+	if tel.registry.IsEnabled(obsreportconfig.UseOtelForInternalMetricsfeatureGateID) {
 		pe, err = tel.initOpenTelemetry(telAttrs)
 	} else {
 		pe, err = tel.initOpenCensus(cfg, telAttrs)