Instrument the otelarrow OTLP receivers

.chloggen/otelarrow-otlp-inst.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: otelarrowreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Instrument the OTLP receiver path for consistency w/ OTel-Arrow path.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [36074]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: OTLP path adds in-flight bytes, items, and requests UpDownCounters to match the OTel-Arrow path.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/otelarrowreceiver/factory.go

@@ -75,8 +75,7 @@ func createTraces(
 		return nil, err
 	}
 
-	recv.Unwrap().(*otelArrowReceiver).registerTraceConsumer(nextConsumer)
-	return recv, nil
+	return recv, recv.Unwrap().(*otelArrowReceiver).registerTraceConsumer(nextConsumer)
 }
 
 // createMetrics creates a metrics receiver based on provided config.
@@ -96,8 +95,7 @@ func createMetrics(
 	if err != nil {
 		return nil, err
 	}
-	recv.Unwrap().(*otelArrowReceiver).registerMetricsConsumer(consumer)
-	return recv, nil
+	return recv, recv.Unwrap().(*otelArrowReceiver).registerMetricsConsumer(consumer)
 }
 
 // createLog creates a log receiver based on provided config.
@@ -118,8 +116,7 @@ func createLog(
 		return nil, err
 	}
 
-	recv.Unwrap().(*otelArrowReceiver).registerLogsConsumer(consumer)
-	return recv, nil
+	return recv, recv.Unwrap().(*otelArrowReceiver).registerLogsConsumer(consumer)
 }
 
 // This is the map of already created OTel-Arrow receivers for particular configurations.

receiver/otelarrowreceiver/internal/logs/otlp.go

@@ -6,36 +6,44 @@ package logs // import "github.com/open-telemetry/opentelemetry-collector-contri
 import (
 	"context"
 
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/plog/plogotlp"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/admission"
+	internalmetadata "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/otelarrowreceiver/internal/metadata"
 )
 
 const dataFormatProtobuf = "protobuf"
 
 // Receiver is the type used to handle logs from OpenTelemetry exporters.
 type Receiver struct {
 	plogotlp.UnimplementedGRPCServer
-	nextConsumer consumer.Logs
-	obsrecv      *receiverhelper.ObsReport
-	boundedQueue admission.Queue
-	sizer        *plog.ProtoMarshaler
-	logger       *zap.Logger
+	nextConsumer     consumer.Logs
+	obsrecv          *receiverhelper.ObsReport
+	boundedQueue     admission.Queue
+	sizer            *plog.ProtoMarshaler
+	logger           *zap.Logger
+	telemetryBuilder *internalmetadata.TelemetryBuilder
 }
 
 // New creates a new Receiver reference.
-func New(logger *zap.Logger, nextConsumer consumer.Logs, obsrecv *receiverhelper.ObsReport, bq admission.Queue) *Receiver {
-	return &Receiver{
-		nextConsumer: nextConsumer,
-		obsrecv:      obsrecv,
-		boundedQueue: bq,
-		sizer:        &plog.ProtoMarshaler{},
-		logger:       logger,
+func New(telset component.TelemetrySettings, nextConsumer consumer.Logs, obsrecv *receiverhelper.ObsReport, bq admission.Queue) (*Receiver, error) {
+	telemetryBuilder, err := internalmetadata.NewTelemetryBuilder(telset)
+	if err != nil {
+		return nil, err
 	}
+	return &Receiver{
+		nextConsumer:     nextConsumer,
+		obsrecv:          obsrecv,
+		boundedQueue:     bq,
+		sizer:            &plog.ProtoMarshaler{},
+		logger:           telset.Logger,
+		telemetryBuilder: telemetryBuilder,
+	}, nil
 }
 
 // Export implements the service Export logs func.
@@ -46,10 +54,15 @@ func (r *Receiver) Export(ctx context.Context, req plogotlp.ExportRequest) (plog
 		return plogotlp.NewExportResponse(), nil
 	}
 
+	sizeBytes := int64(r.sizer.LogsSize(req.Logs()))
+
+	r.telemetryBuilder.OtelArrowReceiverInFlightBytes.Add(ctx, sizeBytes)
+	r.telemetryBuilder.OtelArrowReceiverInFlightItems.Add(ctx, int64(numRecords))
+	r.telemetryBuilder.OtelArrowReceiverInFlightRequests.Add(ctx, 1)
+
 	ctx = r.obsrecv.StartLogsOp(ctx)
 
 	var err error
-	sizeBytes := int64(r.sizer.LogsSize(req.Logs()))
 	if acqErr := r.boundedQueue.Acquire(ctx, sizeBytes); acqErr == nil {
 		err = r.nextConsumer.ConsumeLogs(ctx, ld)
 		// Release() is not checked, see #36074.
@@ -60,6 +73,10 @@ func (r *Receiver) Export(ctx context.Context, req plogotlp.ExportRequest) (plog
 
 	r.obsrecv.EndLogsOp(ctx, dataFormatProtobuf, numRecords, err)
 
+	r.telemetryBuilder.OtelArrowReceiverInFlightBytes.Add(ctx, -sizeBytes)
+	r.telemetryBuilder.OtelArrowReceiverInFlightItems.Add(ctx, -int64(numRecords))
+	r.telemetryBuilder.OtelArrowReceiverInFlightRequests.Add(ctx, -1)
+
 	return plogotlp.NewExportResponse(), err
 }
 

receiver/otelarrowreceiver/internal/logs/otlp_test.go

@@ -25,7 +25,6 @@ import (
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
-	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
@@ -207,7 +206,8 @@ func otlpReceiverOnGRPCServer(t *testing.T, lc consumer.Logs) (net.Addr, *tracet
 	})
 	require.NoError(t, err)
 	bq := admission.NewBoundedQueue(telset, maxBytes, 0)
-	r := New(zap.NewNop(), lc, obsrecv, bq)
+	r, err := New(telset, lc, obsrecv, bq)
+	require.NoError(t, err)
 	// Now run it as a gRPC server
 	srv := grpc.NewServer()
 	plogotlp.RegisterGRPCServer(srv, r)

receiver/otelarrowreceiver/internal/metrics/otlp.go

@@ -6,36 +6,44 @@ package metrics // import "github.com/open-telemetry/opentelemetry-collector-con
 import (
 	"context"
 
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/admission"
+	internalmetadata "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/otelarrowreceiver/internal/metadata"
 )
 
 const dataFormatProtobuf = "protobuf"
 
 // Receiver is the type used to handle metrics from OpenTelemetry exporters.
 type Receiver struct {
 	pmetricotlp.UnimplementedGRPCServer
-	nextConsumer consumer.Metrics
-	obsrecv      *receiverhelper.ObsReport
-	boundedQueue admission.Queue
-	sizer        *pmetric.ProtoMarshaler
-	logger       *zap.Logger
+	nextConsumer     consumer.Metrics
+	obsrecv          *receiverhelper.ObsReport
+	boundedQueue     admission.Queue
+	sizer            *pmetric.ProtoMarshaler
+	logger           *zap.Logger
+	telemetryBuilder *internalmetadata.TelemetryBuilder
 }
 
 // New creates a new Receiver reference.
-func New(logger *zap.Logger, nextConsumer consumer.Metrics, obsrecv *receiverhelper.ObsReport, bq admission.Queue) *Receiver {
-	return &Receiver{
-		nextConsumer: nextConsumer,
-		obsrecv:      obsrecv,
-		boundedQueue: bq,
-		sizer:        &pmetric.ProtoMarshaler{},
-		logger:       logger,
+func New(telset component.TelemetrySettings, nextConsumer consumer.Metrics, obsrecv *receiverhelper.ObsReport, bq admission.Queue) (*Receiver, error) {
+	telemetryBuilder, err := internalmetadata.NewTelemetryBuilder(telset)
+	if err != nil {
+		return nil, err
 	}
+	return &Receiver{
+		nextConsumer:     nextConsumer,
+		obsrecv:          obsrecv,
+		boundedQueue:     bq,
+		sizer:            &pmetric.ProtoMarshaler{},
+		logger:           telset.Logger,
+		telemetryBuilder: telemetryBuilder,
+	}, nil
 }
 
 // Export implements the service Export metrics func.
@@ -46,10 +54,15 @@ func (r *Receiver) Export(ctx context.Context, req pmetricotlp.ExportRequest) (p
 		return pmetricotlp.NewExportResponse(), nil
 	}
 
+	sizeBytes := int64(r.sizer.MetricsSize(req.Metrics()))
+
+	r.telemetryBuilder.OtelArrowReceiverInFlightBytes.Add(ctx, sizeBytes)
+	r.telemetryBuilder.OtelArrowReceiverInFlightItems.Add(ctx, int64(dataPointCount))
+	r.telemetryBuilder.OtelArrowReceiverInFlightRequests.Add(ctx, 1)
+
 	ctx = r.obsrecv.StartMetricsOp(ctx)
 
 	var err error
-	sizeBytes := int64(r.sizer.MetricsSize(req.Metrics()))
 	if acqErr := r.boundedQueue.Acquire(ctx, sizeBytes); acqErr == nil {
 		err = r.nextConsumer.ConsumeMetrics(ctx, md)
 		// Release() is not checked, see #36074.
@@ -60,6 +73,10 @@ func (r *Receiver) Export(ctx context.Context, req pmetricotlp.ExportRequest) (p
 
 	r.obsrecv.EndMetricsOp(ctx, dataFormatProtobuf, dataPointCount, err)
 
+	r.telemetryBuilder.OtelArrowReceiverInFlightBytes.Add(ctx, -sizeBytes)
+	r.telemetryBuilder.OtelArrowReceiverInFlightItems.Add(ctx, -int64(dataPointCount))
+	r.telemetryBuilder.OtelArrowReceiverInFlightRequests.Add(ctx, -1)
+
 	return pmetricotlp.NewExportResponse(), err
 }
 

receiver/otelarrowreceiver/internal/metrics/otlp_test.go

@@ -25,7 +25,6 @@ import (
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
-	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
@@ -207,7 +206,9 @@ func otlpReceiverOnGRPCServer(t *testing.T, mc consumer.Metrics) (net.Addr, *tra
 	})
 	require.NoError(t, err)
 	bq := admission.NewBoundedQueue(telset, maxBytes, 0)
-	r := New(zap.NewNop(), mc, obsrecv, bq)
+	r, err := New(telset, mc, obsrecv, bq)
+	require.NoError(t, err)
+
 	// Now run it as a gRPC server
 	srv := grpc.NewServer()
 	pmetricotlp.RegisterGRPCServer(srv, r)

receiver/otelarrowreceiver/internal/trace/otlp.go

@@ -6,36 +6,44 @@ package trace // import "github.com/open-telemetry/opentelemetry-collector-contr
 import (
 	"context"
 
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/pdata/ptrace/ptraceotlp"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/admission"
+	internalmetadata "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/otelarrowreceiver/internal/metadata"
 )
 
 const dataFormatProtobuf = "protobuf"
 
 // Receiver is the type used to handle spans from OpenTelemetry exporters.
 type Receiver struct {
 	ptraceotlp.UnimplementedGRPCServer
-	nextConsumer consumer.Traces
-	obsrecv      *receiverhelper.ObsReport
-	boundedQueue admission.Queue
-	sizer        *ptrace.ProtoMarshaler
-	logger       *zap.Logger
+	nextConsumer     consumer.Traces
+	obsrecv          *receiverhelper.ObsReport
+	boundedQueue     admission.Queue
+	sizer            *ptrace.ProtoMarshaler
+	logger           *zap.Logger
+	telemetryBuilder *internalmetadata.TelemetryBuilder
 }
 
 // New creates a new Receiver reference.
-func New(logger *zap.Logger, nextConsumer consumer.Traces, obsrecv *receiverhelper.ObsReport, bq admission.Queue) *Receiver {
-	return &Receiver{
-		nextConsumer: nextConsumer,
-		obsrecv:      obsrecv,
-		boundedQueue: bq,
-		sizer:        &ptrace.ProtoMarshaler{},
-		logger:       logger,
+func New(telset component.TelemetrySettings, nextConsumer consumer.Traces, obsrecv *receiverhelper.ObsReport, bq admission.Queue) (*Receiver, error) {
+	telemetryBuilder, err := internalmetadata.NewTelemetryBuilder(telset)
+	if err != nil {
+		return nil, err
 	}
+	return &Receiver{
+		nextConsumer:     nextConsumer,
+		obsrecv:          obsrecv,
+		boundedQueue:     bq,
+		sizer:            &ptrace.ProtoMarshaler{},
+		logger:           telset.Logger,
+		telemetryBuilder: telemetryBuilder,
+	}, nil
 }
 
 // Export implements the service Export traces func.
@@ -46,10 +54,15 @@ func (r *Receiver) Export(ctx context.Context, req ptraceotlp.ExportRequest) (pt
 		return ptraceotlp.NewExportResponse(), nil
 	}
 
+	sizeBytes := int64(r.sizer.TracesSize(req.Traces()))
+
+	r.telemetryBuilder.OtelArrowReceiverInFlightBytes.Add(ctx, sizeBytes)
+	r.telemetryBuilder.OtelArrowReceiverInFlightItems.Add(ctx, int64(numSpans))
+	r.telemetryBuilder.OtelArrowReceiverInFlightRequests.Add(ctx, 1)
+
 	ctx = r.obsrecv.StartTracesOp(ctx)
 
 	var err error
-	sizeBytes := int64(r.sizer.TracesSize(req.Traces()))
 	if acqErr := r.boundedQueue.Acquire(ctx, sizeBytes); acqErr == nil {
 		err = r.nextConsumer.ConsumeTraces(ctx, td)
 		// Release() is not checked, see #36074.
@@ -58,6 +71,10 @@ func (r *Receiver) Export(ctx context.Context, req ptraceotlp.ExportRequest) (pt
 		err = acqErr
 	}
 
+	r.telemetryBuilder.OtelArrowReceiverInFlightBytes.Add(ctx, -sizeBytes)
+	r.telemetryBuilder.OtelArrowReceiverInFlightItems.Add(ctx, -int64(numSpans))
+	r.telemetryBuilder.OtelArrowReceiverInFlightRequests.Add(ctx, -1)
+
 	r.obsrecv.EndTracesOp(ctx, dataFormatProtobuf, numSpans, err)
 
 	return ptraceotlp.NewExportResponse(), err

receiver/otelarrowreceiver/internal/trace/otlp_test.go

@@ -25,7 +25,6 @@ import (
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
-	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
@@ -207,7 +206,8 @@ func otlpReceiverOnGRPCServer(t *testing.T, tc consumer.Traces) (net.Addr, *trac
 	})
 	require.NoError(t, err)
 	bq := admission.NewBoundedQueue(telset, maxBytes, 0)
-	r := New(zap.NewNop(), tc, obsrecv, bq)
+	r, err := New(telset, tc, obsrecv, bq)
+	require.NoError(t, err)
 	// Now run it as a gRPC server
 	srv := grpc.NewServer()
 	ptraceotlp.RegisterGRPCServer(srv, r)

receiver/otelarrowreceiver/otelarrow.go

@@ -192,16 +192,19 @@ func (r *otelArrowReceiver) Shutdown(_ context.Context) error {
 	return err
 }
 
-func (r *otelArrowReceiver) registerTraceConsumer(tc consumer.Traces) {
-	r.tracesReceiver = trace.New(r.settings.Logger, tc, r.obsrepGRPC, r.boundedQueue)
+func (r *otelArrowReceiver) registerTraceConsumer(tc consumer.Traces) (err error) {
+	r.tracesReceiver, err = trace.New(r.settings.TelemetrySettings, tc, r.obsrepGRPC, r.boundedQueue)
+	return
 }
 
-func (r *otelArrowReceiver) registerMetricsConsumer(mc consumer.Metrics) {
-	r.metricsReceiver = metrics.New(r.settings.Logger, mc, r.obsrepGRPC, r.boundedQueue)
+func (r *otelArrowReceiver) registerMetricsConsumer(mc consumer.Metrics) (err error) {
+	r.metricsReceiver, err = metrics.New(r.settings.TelemetrySettings, mc, r.obsrepGRPC, r.boundedQueue)
+	return
 }
 
-func (r *otelArrowReceiver) registerLogsConsumer(lc consumer.Logs) {
-	r.logsReceiver = logs.New(r.settings.Logger, lc, r.obsrepGRPC, r.boundedQueue)
+func (r *otelArrowReceiver) registerLogsConsumer(lc consumer.Logs) (err error) {
+	r.logsReceiver, err = logs.New(r.settings.TelemetrySettings, lc, r.obsrepGRPC, r.boundedQueue)
+	return
 }
 
 var _ arrow.Consumers = &otelArrowReceiver{}