Add accepted/refused/failed profile sample metrics to the receivers (#14226)

This adds obs metrics to the profiles receivers.

---------

Co-authored-by: Roger Coll <[email protected]>
Co-authored-by: Pablo Baeyens <[email protected]>

Author: 3 people

.chloggen/otlpreceiver-profile-metrics.yaml

@@ -0,0 +1,25 @@
+# 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. receiver/otlp)
+component: receiver/otlp
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add metrics tracking the number of receiver, refused and failed profile samples
+
+# One or more tracking issues or pull requests related to the change
+issues: [14226]
+
+# (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:
+
+# 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: [api]

.chloggen/receiverhelper-profile-metrics.yaml

@@ -0,0 +1,25 @@
+# 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. receiver/otlp)
+component: pkg/receiverhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for profile samples metrics
+
+# One or more tracking issues or pull requests related to the change
+issues: [14226]
+
+# (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:
+
+# 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: [api]

receiver/otlpreceiver/go.mod

@@ -76,6 +76,7 @@ require (
 	go.opentelemetry.io/collector/featuregate v1.49.0 // indirect
 	go.opentelemetry.io/collector/internal/componentalias v0.0.0-00010101000000-000000000000 // indirect
 	go.opentelemetry.io/collector/pipeline v1.49.0 // indirect
+	go.opentelemetry.io/collector/pipeline/xpipeline v0.0.0-00010101000000-000000000000 // 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.39.0 // indirect
@@ -169,4 +170,6 @@ replace go.opentelemetry.io/collector/extension/extensionmiddleware/extensionmid
 
 replace go.opentelemetry.io/collector/internal/testutil => ../../internal/testutil
 
+replace go.opentelemetry.io/collector/pipeline/xpipeline => ../../pipeline/xpipeline
+
 replace go.opentelemetry.io/collector/internal/componentalias => ../../internal/componentalias

receiver/otlpreceiver/internal/profiles/otlp.go

@@ -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_test.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/otlp.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/receiverhelper/documentation.md

@@ -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/go.mod

@@ -9,6 +9,7 @@ require (
 	go.opentelemetry.io/collector/consumer/consumererror v0.143.0
 	go.opentelemetry.io/collector/featuregate v1.49.0
 	go.opentelemetry.io/collector/pipeline v1.49.0
+	go.opentelemetry.io/collector/pipeline/xpipeline v0.0.0-00010101000000-000000000000
 	go.opentelemetry.io/collector/receiver v1.49.0
 	go.opentelemetry.io/otel v1.39.0
 	go.opentelemetry.io/otel/metric v1.39.0
@@ -69,4 +70,6 @@ replace go.opentelemetry.io/collector/consumer/consumererror => ../../consumer/c
 
 replace go.opentelemetry.io/collector/internal/testutil => ../../internal/testutil
 
+replace go.opentelemetry.io/collector/pipeline/xpipeline => ../../pipeline/xpipeline
+
 replace go.opentelemetry.io/collector/internal/componentalias => ../../internal/componentalias