Update receiverhelper for requests that failed to be received (#12802)

#### Description

This is an ongoing work to add :
- In the endOp function of
[ObsReport](receiver/receiverhelper/obsreport.go), errors are classified
as either:
- Downstream errors (refused) - when internal.IsDownstreamError(err)
returns true
   - Internal errors - when the error is not a downstream error
- New metrics otelcol_receiver_internal_errors_spans,
otelcol_receiver_internal_errors_metric_points and
otelcol_receiver_internal_errors_log_records in case of inetrnal errors
returned above
- Support of otelcol_receiver_requests metric that will determines the
outcome of the request:
    -  "success" if there's no error
    -   "refused" if there's a downstream error
    -  "failure" if there's an internal error

#### Link to tracking issue
Fixes #
#12207

#### Testing

Only added the integration tests. 

#### Documentation

See updated
[receiver/receiverhelper/documentation.md](receiver/receiverhelper/documentation.md)

---------

Signed-off-by: Andreas Gkizas <[email protected]>
Co-authored-by: Jade Guiton <[email protected]>

Author: gizas

.chloggen/custom_receiverhelper.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. otlpreceiver)
+component: receiverhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: New feature flag to make receiverhelper distinguish internal vs. downstream errors using new `otelcol_receiver_failed_x` and `otelcol_receiver_requests` metrics
+
+# One or more tracking issues or pull requests related to the change
+issues: [12207, 12802]
+
+# (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: | 
+  This is a breaking change for the semantics of the otelcol_receiver_refused_metric_points,  otelcol_receiver_refused_log_records and otelcol_receiver_refused_spans metrics.
+  These new metrics and semantics are enabled through the `receiverhelper.newReceiverMetrics` feature gate.
+
+# 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: []

receiver/otlpreceiver/otlp_test.go

@@ -44,11 +44,13 @@ import (
 	"go.opentelemetry.io/collector/internal/testutil"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
 	"go.opentelemetry.io/collector/pdata/pprofile"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/pdata/ptrace/ptraceotlp"
 	"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"
 )
 
@@ -118,8 +120,8 @@ func TestJsonHttp(t *testing.T) {
 			name:               "Retryable GRPCError",
 			encoding:           "",
 			contentType:        "application/json",
-			err:                status.New(codes.Unavailable, "").Err(),
-			expectedStatus:     &spb.Status{Code: int32(codes.Unavailable), Message: ""},
+			err:                status.New(codes.Unavailable, "Service Unavailable").Err(),
+			expectedStatus:     &spb.Status{Code: int32(codes.Unavailable), Message: "Service Unavailable"},
 			expectedStatusCode: http.StatusServiceUnavailable,
 		},
 	}
@@ -145,7 +147,8 @@ func TestJsonHttp(t *testing.T) {
 					errStatus := &spb.Status{}
 					require.NoError(t, json.Unmarshal(respBytes, errStatus))
 					if s, ok := status.FromError(tt.err); ok {
-						assert.True(t, proto.Equal(errStatus, s.Proto()))
+						assert.Equal(t, s.Proto().Code, errStatus.Code)
+						assert.Equal(t, s.Proto().Message, errStatus.Message)
 					} else {
 						fmt.Println(errStatus)
 						assert.True(t, proto.Equal(errStatus, tt.expectedStatus))
@@ -365,15 +368,15 @@ func TestProtoHttp(t *testing.T) {
 		{
 			name:               "Permanent GRPCError",
 			encoding:           "",
-			err:                status.New(codes.InvalidArgument, "").Err(),
-			expectedStatus:     &spb.Status{Code: int32(codes.InvalidArgument), Message: ""},
+			err:                status.New(codes.InvalidArgument, "Bad Request").Err(),
+			expectedStatus:     &spb.Status{Code: int32(codes.InvalidArgument), Message: "Bad Request"},
 			expectedStatusCode: http.StatusBadRequest,
 		},
 		{
 			name:               "Retryable GRPCError",
 			encoding:           "",
-			err:                status.New(codes.Unavailable, "").Err(),
-			expectedStatus:     &spb.Status{Code: int32(codes.Unavailable), Message: ""},
+			err:                status.New(codes.Unavailable, "Service Unavailable").Err(),
+			expectedStatus:     &spb.Status{Code: int32(codes.Unavailable), Message: "Service Unavailable"},
 			expectedStatusCode: http.StatusServiceUnavailable,
 		},
 	}
@@ -553,6 +556,45 @@ func TestHTTPNewPortAlreadyUsed(t *testing.T) {
 	require.Error(t, r.Start(context.Background(), componenttest.NewNopHost()))
 }
 
+// TestOTLPReceiverGRPCMetricsIngestTest checks that the metrics receiver
+// is returning the proper response (return and metrics) when the next consumer
+// in the pipeline reports error.
+func TestOTLPReceiverGRPCMetricsIngestTest(t *testing.T) {
+	// Get a new available port
+	addr := testutil.GetAvailableLocalAddress(t)
+
+	// Create a sink
+	sink := &errOrSinkConsumer{MetricsSink: new(consumertest.MetricsSink)}
+
+	// Create a telemetry instance
+	tt := componenttest.NewTelemetry()
+	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
+	// Create telemetry settings
+	settings := tt.NewTelemetrySettings()
+
+	recv := newGRPCReceiver(t, settings, addr, sink)
+	require.NotNil(t, recv)
+	require.NoError(t, recv.Start(context.Background(), componenttest.NewNopHost()))
+	t.Cleanup(func() { require.NoError(t, recv.Shutdown(context.Background())) })
+
+	cc, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	defer func() {
+		assert.NoError(t, cc.Close())
+	}()
+	// Set up the error case
+	sink.SetConsumeError(errors.New("consumer error"))
+
+	md := testdata.GenerateMetrics(1)
+	_, err = pmetricotlp.NewGRPCClient(cc).Export(context.Background(), pmetricotlp.NewExportRequestFromMetrics(md))
+	errStatus, ok := status.FromError(err)
+	require.True(t, ok)
+	assert.Equal(t, codes.Unavailable, errStatus.Code())
+
+	// Assert receiver metrics including receiver_requests
+	assertReceiverMetrics(t, tt, otlpReceiverID, "grpc", 0, 2)
+}
+
 // TestOTLPReceiverGRPCTracesIngestTest checks that the gRPC trace receiver
 // is returning the proper response (return and metrics) when the next consumer
 // in the pipeline reports error. The test changes the responses returned by the
@@ -1262,8 +1304,41 @@ func (esc *errOrSinkConsumer) checkData(t *testing.T, data any, dataLen int) {
 	}
 }
 
-func assertReceiverTraces(t *testing.T, tt *componenttest.Telemetry, id component.ID, transport string, accepted, refused int64) {
-	got, err := tt.GetMetric("otelcol_receiver_accepted_spans")
+func assertReceiverTraces(t *testing.T, tt *componenttest.Telemetry, id component.ID, transport string, accepted, rejected int64) {
+	var refused, failed int64
+	var outcome string
+	gateEnabled := receiverhelper.NewReceiverMetricsGate.IsEnabled()
+	// The errors in the OTLP tests are not downstream, so they should be "failed" when the gate is enabled.
+	if gateEnabled {
+		failed = rejected
+		outcome = "failure"
+	} else {
+		// When the gate is disabled, all errors are "refused".
+		refused = rejected
+	}
+
+	got, err := tt.GetMetric("otelcol_receiver_failed_spans")
+	require.NoError(t, err)
+	metricdatatest.AssertEqual(t,
+		metricdata.Metrics{
+			Name:        "otelcol_receiver_failed_spans",
+			Description: "The number of spans that failed to be processed by the receiver due to internal errors. [alpha]",
+			Unit:        "{spans}",
+			Data: metricdata.Sum[int64]{
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(
+							attribute.String("receiver", id.String()),
+							attribute.String("transport", transport)),
+						Value: failed,
+					},
+				},
+			},
+		}, got, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+
+	got, err = tt.GetMetric("otelcol_receiver_accepted_spans")
 	require.NoError(t, err)
 	metricdatatest.AssertEqual(t,
 		metricdata.Metrics{
@@ -1304,4 +1379,165 @@ func assertReceiverTraces(t *testing.T, tt *componenttest.Telemetry, id componen
 				},
 			},
 		}, got, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+
+	// Assert receiver_requests metric
+	if gateEnabled {
+		got, err := tt.GetMetric("otelcol_receiver_requests")
+		require.NoError(t, err)
+
+		// Calculate expected requests based on accepted and refused counts
+		var expectedRequests []metricdata.DataPoint[int64]
+		if accepted > 0 {
+			expectedRequests = append(expectedRequests, metricdata.DataPoint[int64]{
+				Attributes: attribute.NewSet(
+					attribute.String("receiver", id.String()),
+					attribute.String("transport", transport),
+					attribute.String("outcome", "success")),
+				Value: accepted,
+			})
+		}
+		if rejected > 0 {
+			expectedRequests = append(expectedRequests, metricdata.DataPoint[int64]{
+				Attributes: attribute.NewSet(
+					attribute.String("receiver", id.String()),
+					attribute.String("transport", transport),
+					attribute.String("outcome", outcome)),
+				Value: rejected,
+			})
+		}
+
+		metricdatatest.AssertEqual(t,
+			metricdata.Metrics{
+				Name:        "otelcol_receiver_requests",
+				Description: "The number of requests performed.",
+				Unit:        "{requests}",
+				Data: metricdata.Sum[int64]{
+					Temporality: metricdata.CumulativeTemporality,
+					IsMonotonic: true,
+					DataPoints:  expectedRequests,
+				},
+			}, got, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+	} else {
+		_, err := tt.GetMetric("otelcol_receiver_requests")
+		require.Error(t, err)
+	}
+}
+
+func assertReceiverMetrics(t *testing.T, tt *componenttest.Telemetry, id component.ID, transport string, accepted, rejected int64) {
+	var refused, failed int64
+	var outcome string
+	gateEnabled := receiverhelper.NewReceiverMetricsGate.IsEnabled()
+	// The error used in the metrics test is not downstream.
+	if gateEnabled {
+		failed = rejected
+		outcome = "failure"
+	} else {
+		// When the gate is disabled, all errors are "refused".
+		refused = rejected
+	}
+
+	got, err := tt.GetMetric("otelcol_receiver_failed_metric_points")
+	require.NoError(t, err)
+	metricdatatest.AssertEqual(t,
+		metricdata.Metrics{
+			Name:        "otelcol_receiver_failed_metric_points",
+			Description: "The number of metric points that failed to be processed by the receiver due to internal errors. [alpha]",
+			Unit:        "{datapoints}",
+			Data: metricdata.Sum[int64]{
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(
+							attribute.String("receiver", id.String()),
+							attribute.String("transport", transport)),
+						Value: failed,
+					},
+				},
+			},
+		}, got, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+
+	got, err = tt.GetMetric("otelcol_receiver_accepted_metric_points")
+	require.NoError(t, err)
+	metricdatatest.AssertEqual(t,
+		metricdata.Metrics{
+			Name:        "otelcol_receiver_accepted_metric_points",
+			Description: "Number of metric points successfully pushed into the pipeline. [alpha]",
+			Unit:        "{datapoints}",
+			Data: metricdata.Sum[int64]{
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(
+							attribute.String("receiver", id.String()),
+							attribute.String("transport", transport)),
+						Value: accepted,
+					},
+				},
+			},
+		}, got, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+
+	got, err = tt.GetMetric("otelcol_receiver_refused_metric_points")
+	require.NoError(t, err)
+	metricdatatest.AssertEqual(t,
+		metricdata.Metrics{
+			Name:        "otelcol_receiver_refused_metric_points",
+			Description: "Number of metric points that could not be pushed into the pipeline. [alpha]",
+			Unit:        "{datapoints}",
+			Data: metricdata.Sum[int64]{
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(
+							attribute.String("receiver", id.String()),
+							attribute.String("transport", transport)),
+						Value: refused,
+					},
+				},
+			},
+		}, got, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+
+	// Assert receiver_requests metric
+	if gateEnabled {
+		got, err := tt.GetMetric("otelcol_receiver_requests")
+		require.NoError(t, err)
+
+		// Calculate expected requests based on accepted and refused counts
+		var expectedRequests []metricdata.DataPoint[int64]
+		if accepted > 0 {
+			expectedRequests = append(expectedRequests, metricdata.DataPoint[int64]{
+				Attributes: attribute.NewSet(
+					attribute.String("receiver", id.String()),
+					attribute.String("transport", transport),
+					attribute.String("outcome", "success")),
+				Value: accepted,
+			})
+		}
+		if rejected > 0 {
+			expectedRequests = append(expectedRequests, metricdata.DataPoint[int64]{
+				Attributes: attribute.NewSet(
+					attribute.String("receiver", id.String()),
+					attribute.String("transport", transport),
+					attribute.String("outcome", outcome)),
+				Value: 1, // One request failed
+			})
+		}
+
+		metricdatatest.AssertEqual(t,
+			metricdata.Metrics{
+				Name:        "otelcol_receiver_requests",
+				Description: "The number of requests performed.",
+				Unit:        "{requests}",
+				Data: metricdata.Sum[int64]{
+					Temporality: metricdata.CumulativeTemporality,
+					IsMonotonic: true,
+					DataPoints:  expectedRequests,
+				},
+			}, got, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+	} else {
+		_, err := tt.GetMetric("otelcol_receiver_requests")
+		require.Error(t, err)
+	}
 }

receiver/otlpreceiver/otlphttp_test.go

@@ -23,6 +23,7 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/internal/testutil"
 	"go.opentelemetry.io/collector/pdata/ptrace/ptraceotlp"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver/internal/errors"
 )
 
 func TestHttpRetryAfter(t *testing.T) {
@@ -126,7 +127,10 @@ func TestHttpRetryAfter(t *testing.T) {
 				} else {
 					errStatus := &spb.Status{}
 					require.NoError(t, proto.Unmarshal(respBytes, errStatus))
-					s, ok := status.FromError(tt.err)
+					// The HTTP receiver transforms errors through GetStatusFromError
+					// We need to get the expected transformed error, not the original
+					expectedErr := errors.GetStatusFromError(tt.err)
+					s, ok := status.FromError(expectedErr)
 					require.True(t, ok)
 					assert.True(t, proto.Equal(errStatus, s.Proto()))
 				}