[exporterhelper] record metric should log the number of log records before the data are sent to the consumers downstream

.chloggen/exporterhelper_metric_fix.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: exporterhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: There is no guarantee that after the exporterhelper sends the plog/pmetric/ptrace data downstream that the data won't be mutated in some way. (e.g by the batch_sender) This mutation could result in the proceeding call to req.ItemsCount() to provide inaccurate information to be logged.
+
+# One or more tracking issues or pull requests related to the change
+issues: [10033]
+
+# 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: []

exporter/exporterhelper/logs.go

@@ -155,7 +155,8 @@ func newLogsExporterWithObservability(obsrep *ObsReport) requestSender {
 
 func (lewo *logsExporterWithObservability) send(ctx context.Context, req Request) error {
 	c := lewo.obsrep.StartLogsOp(ctx)
+	numLogRecords := req.ItemsCount()
 	err := lewo.nextSender.send(c, req)
-	lewo.obsrep.EndLogsOp(c, req.ItemsCount(), err)
+	lewo.obsrep.EndLogsOp(c, numLogRecords, err)
 	return err
 }

exporter/exporterhelper/logs_test.go

@@ -191,6 +191,21 @@ func TestLogsExporter_WithRecordMetrics(t *testing.T) {
 	checkRecordedMetricsForLogsExporter(t, tt, le, nil)
 }
 
+func TestLogsExporter_pLogModifiedDownStream_WithRecordMetrics(t *testing.T) {
+	tt, err := componenttest.SetupTelemetry(fakeLogsExporterName)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
+
+	le, err := NewLogsExporter(context.Background(), exporter.Settings{ID: fakeLogsExporterName, TelemetrySettings: tt.TelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()}, &fakeLogsExporterConfig, newPushLogsDataModifiedDownstream(nil), WithCapabilities(consumer.Capabilities{MutatesData: true}))
+	assert.NotNil(t, le)
+	assert.NoError(t, err)
+	ld := testdata.GenerateLogs(2)
+
+	assert.NoError(t, le.ConsumeLogs(context.Background(), ld))
+	assert.Equal(t, 0, ld.LogRecordCount())
+	require.NoError(t, tt.CheckExporterLogs(int64(2), 0))
+}
+
 func TestLogsRequestExporter_WithRecordMetrics(t *testing.T) {
 	tt, err := componenttest.SetupTelemetry(fakeLogsExporterName)
 	require.NoError(t, err)
@@ -359,6 +374,13 @@ func TestLogsRequestExporter_WithShutdown_ReturnError(t *testing.T) {
 	assert.Equal(t, le.Shutdown(context.Background()), want)
 }
 
+func newPushLogsDataModifiedDownstream(retError error) consumer.ConsumeLogsFunc {
+	return func(_ context.Context, log plog.Logs) error {
+		log.ResourceLogs().MoveAndAppendTo(plog.NewResourceLogsSlice())
+		return retError
+	}
+}
+
 func newPushLogsData(retError error) consumer.ConsumeLogsFunc {
 	return func(_ context.Context, _ plog.Logs) error {
 		return retError

exporter/exporterhelper/metrics.go

@@ -155,7 +155,8 @@ func newMetricsSenderWithObservability(obsrep *ObsReport) requestSender {
 
 func (mewo *metricsSenderWithObservability) send(ctx context.Context, req Request) error {
 	c := mewo.obsrep.StartMetricsOp(ctx)
+	numMetricDataPoints := req.ItemsCount()
 	err := mewo.nextSender.send(c, req)
-	mewo.obsrep.EndMetricsOp(c, req.ItemsCount(), err)
+	mewo.obsrep.EndMetricsOp(c, numMetricDataPoints, err)
 	return err
 }

exporter/exporterhelper/metrics_test.go

@@ -192,6 +192,21 @@ func TestMetricsExporter_WithRecordMetrics(t *testing.T) {
 	checkRecordedMetricsForMetricsExporter(t, tt, me, nil)
 }
 
+func TestMetricsExporter_pMetricModifiedDownStream_WithRecordMetrics(t *testing.T) {
+	tt, err := componenttest.SetupTelemetry(fakeMetricsExporterName)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
+
+	me, err := NewMetricsExporter(context.Background(), exporter.Settings{ID: fakeMetricsExporterName, TelemetrySettings: tt.TelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()}, &fakeMetricsExporterConfig, newPushMetricsDataModifiedDownstream(nil), WithCapabilities(consumer.Capabilities{MutatesData: true}))
+	require.NoError(t, err)
+	require.NotNil(t, me)
+	md := testdata.GenerateMetrics(2)
+
+	assert.NoError(t, me.ConsumeMetrics(context.Background(), md))
+	assert.Equal(t, 0, md.MetricCount())
+	require.NoError(t, tt.CheckExporterMetrics(int64(4), 0))
+}
+
 func TestMetricsRequestExporter_WithRecordMetrics(t *testing.T) {
 	tt, err := componenttest.SetupTelemetry(fakeMetricsExporterName)
 	require.NoError(t, err)
@@ -370,6 +385,12 @@ func newPushMetricsData(retError error) consumer.ConsumeMetricsFunc {
 		return retError
 	}
 }
+func newPushMetricsDataModifiedDownstream(retError error) consumer.ConsumeMetricsFunc {
+	return func(_ context.Context, metric pmetric.Metrics) error {
+		metric.ResourceMetrics().MoveAndAppendTo(pmetric.NewResourceMetricsSlice())
+		return retError
+	}
+}
 
 func checkRecordedMetricsForMetricsExporter(t *testing.T, tt componenttest.TestTelemetry, me exporter.Metrics, wantError error) {
 	md := testdata.GenerateMetrics(2)

exporter/exporterhelper/traces.go

@@ -155,8 +155,9 @@ func newTracesExporterWithObservability(obsrep *ObsReport) requestSender {
 
 func (tewo *tracesExporterWithObservability) send(ctx context.Context, req Request) error {
 	c := tewo.obsrep.StartTracesOp(ctx)
+	numTraceSpans := req.ItemsCount()
 	// Forward the data to the next consumer (this pusher is the next).
 	err := tewo.nextSender.send(c, req)
-	tewo.obsrep.EndTracesOp(c, req.ItemsCount(), err)
+	tewo.obsrep.EndTracesOp(c, numTraceSpans, err)
 	return err
 }

exporter/exporterhelper/traces_test.go

@@ -189,6 +189,21 @@ func TestTracesExporter_WithRecordMetrics(t *testing.T) {
 	checkRecordedMetricsForTracesExporter(t, tt, te, nil)
 }
 
+func TestTracesExporter_pLogModifiedDownStream_WithRecordMetrics(t *testing.T) {
+	tt, err := componenttest.SetupTelemetry(fakeTracesExporterName)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
+
+	te, err := NewTracesExporter(context.Background(), exporter.Settings{ID: fakeTracesExporterName, TelemetrySettings: tt.TelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()}, &fakeTracesExporterConfig, newTraceDataPusherModifiedDownstream(nil), WithCapabilities(consumer.Capabilities{MutatesData: true}))
+	assert.NotNil(t, te)
+	assert.NoError(t, err)
+	td := testdata.GenerateTraces(2)
+
+	assert.NoError(t, te.ConsumeTraces(context.Background(), td))
+	assert.Equal(t, 0, td.SpanCount())
+	require.NoError(t, tt.CheckExporterTraces(int64(2), 0))
+}
+
 func TestTracesRequestExporter_WithRecordMetrics(t *testing.T) {
 	tt, err := componenttest.SetupTelemetry(fakeTracesExporterName)
 	require.NoError(t, err)
@@ -372,6 +387,13 @@ func newTraceDataPusher(retError error) consumer.ConsumeTracesFunc {
 	}
 }
 
+func newTraceDataPusherModifiedDownstream(retError error) consumer.ConsumeTracesFunc {
+	return func(_ context.Context, trace ptrace.Traces) error {
+		trace.ResourceSpans().MoveAndAppendTo(ptrace.NewResourceSpansSlice())
+		return retError
+	}
+}
+
 func checkRecordedMetricsForTracesExporter(t *testing.T, tt componenttest.TestTelemetry, te exporter.Traces, wantError error) {
 	td := testdata.GenerateTraces(2)
 	const numBatches = 7