add accepted/refused/failed profile sample metrics to the receivers

Author: dmathieu

receiver/otlpreceiver/go.mod

@@ -74,6 +74,7 @@ require (
 	go.opentelemetry.io/collector/extension/extensionmiddleware v0.140.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.46.0 // indirect
 	go.opentelemetry.io/collector/pipeline v1.46.0 // indirect
+	go.opentelemetry.io/collector/pipeline/xpipeline v0.140.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.63.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 // indirect
 	go.opentelemetry.io/otel/metric v1.38.0 // indirect

receiver/otlpreceiver/go.sum

@@ -9,31 +9,39 @@ import (
 	"go.opentelemetry.io/collector/consumer/xconsumer"
 	"go.opentelemetry.io/collector/pdata/pprofile/pprofileotlp"
 	"go.opentelemetry.io/collector/receiver/otlpreceiver/internal/errors"
+	"go.opentelemetry.io/collector/receiver/receiverhelper"
 )
 
+const dataFormatProtobuf = "protobuf"
+
 // Receiver is the type used to handle spans from OpenTelemetry exporters.
 type Receiver struct {
 	pprofileotlp.UnimplementedGRPCServer
 	nextConsumer xconsumer.Profiles
+	obsreport    *receiverhelper.ObsReport
 }
 
 // New creates a new Receiver reference.
-func New(nextConsumer xconsumer.Profiles) *Receiver {
+func New(nextConsumer xconsumer.Profiles, obsreport *receiverhelper.ObsReport) *Receiver {
 	return &Receiver{
 		nextConsumer: nextConsumer,
+		obsreport:    obsreport,
 	}
 }
 
 // Export implements the service Export profiles func.
 func (r *Receiver) Export(ctx context.Context, req pprofileotlp.ExportRequest) (pprofileotlp.ExportResponse, error) {
 	td := req.Profiles()
 	// We need to ensure that it propagates the receiver name as a tag
-	numProfiles := td.SampleCount()
-	if numProfiles == 0 {
+	numSamples := td.SampleCount()
+	if numSamples == 0 {
 		return pprofileotlp.NewExportResponse(), nil
 	}
 
+	ctx = r.obsreport.StartTracesOp(ctx)
 	err := r.nextConsumer.ConsumeProfiles(ctx, td)
+	r.obsreport.EndTracesOp(ctx, dataFormatProtobuf, numSamples, err)
+
 	// Use appropriate status codes for permanent/non-permanent errors
 	// If we return the error straightaway, then the grpc implementation will set status code to Unknown
 	// Refer: https://github.com/grpc/grpc-go/blob/v1.59.0/server.go#L1345

receiver/otlpreceiver/internal/profiles/otlp.go

@@ -16,11 +16,15 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/status"
 
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/consumer/xconsumer"
 	"go.opentelemetry.io/collector/pdata/pprofile/pprofileotlp"
 	"go.opentelemetry.io/collector/pdata/testdata"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver/internal/metadata"
+	"go.opentelemetry.io/collector/receiver/receiverhelper"
+	"go.opentelemetry.io/collector/receiver/receivertest"
 )
 
 func TestExport(t *testing.T) {
@@ -86,7 +90,16 @@ func otlpReceiverOnGRPCServer(t *testing.T, tc xconsumer.Profiles) net.Addr {
 		require.NoError(t, ln.Close())
 	})
 
-	r := New(tc)
+	set := receivertest.NewNopSettings(metadata.Type)
+	set.ID = component.MustNewIDWithName("otlp", "profiles")
+	obsreport, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{
+		ReceiverID:             set.ID,
+		Transport:              "grpc",
+		ReceiverCreateSettings: set,
+	})
+	require.NoError(t, err)
+
+	r := New(tc, obsreport)
 	// Now run it as a gRPC server
 	srv := grpc.NewServer()
 	pprofileotlp.RegisterGRPCServer(srv, r)

receiver/otlpreceiver/internal/profiles/otlp_test.go

@@ -110,7 +110,7 @@ func (r *otlpReceiver) startGRPCServer(ctx context.Context, host component.Host)
 	}
 
 	if r.nextProfiles != nil {
-		pprofileotlp.RegisterGRPCServer(r.serverGRPC, profiles.New(r.nextProfiles))
+		pprofileotlp.RegisterGRPCServer(r.serverGRPC, profiles.New(r.nextProfiles, r.obsrepGRPC))
 	}
 
 	var gln net.Listener
@@ -160,7 +160,7 @@ func (r *otlpReceiver) startHTTPServer(ctx context.Context, host component.Host)
 	}
 
 	if r.nextProfiles != nil {
-		httpProfilesReceiver := profiles.New(r.nextProfiles)
+		httpProfilesReceiver := profiles.New(r.nextProfiles, r.obsrepHTTP)
 		httpMux.HandleFunc(defaultProfilesURLPath, func(resp http.ResponseWriter, req *http.Request) {
 			handleProfiles(resp, req, httpProfilesReceiver)
 		})

receiver/otlpreceiver/otlp.go

@@ -22,6 +22,14 @@ Number of metric points successfully pushed into the pipeline. [Alpha]
 | ---- | ----------- | ---------- | --------- | --------- |
 | {datapoints} | Sum | Int | true | Alpha |
 
+### otelcol_receiver_accepted_profile_samples
+
+Number of profile samples successfully pushed into the pipeline. [Alpha]
+
+| Unit | Metric Type | Value Type | Monotonic | Stability |
+| ---- | ----------- | ---------- | --------- | --------- |
+| {samples} | Sum | Int | true | Alpha |
+
 ### otelcol_receiver_accepted_spans
 
 Number of spans successfully pushed into the pipeline. [Alpha]
@@ -46,6 +54,14 @@ The number of metric points that failed to be processed by the receiver due to i
 | ---- | ----------- | ---------- | --------- | --------- |
 | {datapoints} | Sum | Int | true | Alpha |
 
+### otelcol_receiver_failed_profile_samples
+
+The number of profile samples that failed to be processed by the receiver due to internal errors. [Alpha]
+
+| Unit | Metric Type | Value Type | Monotonic | Stability |
+| ---- | ----------- | ---------- | --------- | --------- |
+| {samples} | Sum | Int | true | Alpha |
+
 ### otelcol_receiver_failed_spans
 
 The number of spans that failed to be processed by the receiver due to internal errors. [Alpha]
@@ -70,6 +86,14 @@ Number of metric points that could not be pushed into the pipeline. [Alpha]
 | ---- | ----------- | ---------- | --------- | --------- |
 | {datapoints} | Sum | Int | true | Alpha |
 
+### otelcol_receiver_refused_profile_samples
+
+Number of profile samples that could not be pushed into the pipeline. [Alpha]
+
+| Unit | Metric Type | Value Type | Monotonic | Stability |
+| ---- | ----------- | ---------- | --------- | --------- |
+| {samples} | Sum | Int | true | Alpha |
+
 ### otelcol_receiver_refused_spans
 
 Number of spans that could not be pushed into the pipeline. [Alpha]

receiver/receiverhelper/documentation.md

@@ -9,6 +9,7 @@ require (
 	go.opentelemetry.io/collector/consumer/consumererror v0.140.0
 	go.opentelemetry.io/collector/featuregate v1.46.0
 	go.opentelemetry.io/collector/pipeline v1.46.0
+	go.opentelemetry.io/collector/pipeline/xpipeline v0.140.0
 	go.opentelemetry.io/collector/receiver v1.46.0
 	go.opentelemetry.io/otel v1.38.0
 	go.opentelemetry.io/otel/metric v1.38.0