Add ingoing and outgoing counts to processorhelper (open-telemetry#10910

)

#### Description

Implements ingoing and outgoing counts as described in
open-telemetry#10708.

.chloggen/processor-helper-metrics.yaml

@@ -0,0 +1,33 @@
+# 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: processor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add incoming and outgoing counts for processors using processorhelper.
+
+# One or more tracking issues or pull requests related to the change
+issues: [10910]
+
+# (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: |
+  Any processor using the processorhelper package (this is most processors) will automatically report
+  incoming and outgoing item counts. The new metrics are:
+  - otelcol_processor_incoming_spans
+  - otelcol_processor_outgoing_spans
+  - otelcol_processor_incoming_metric_points
+  - otelcol_processor_outgoing_metric_points
+  - otelcol_processor_incoming_log_records
+  - otelcol_processor_outgoing_log_records
+
+# 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: []

processor/memorylimiterprocessor/memorylimiter_test.go

@@ -57,7 +57,8 @@ func TestNoDataLoss(t *testing.T) {
 	require.NoError(t, err)
 
 	processor, err := processorhelper.NewLogsProcessor(context.Background(), processor.Settings{
-		ID: component.MustNewID("nop"),
+		ID:                component.MustNewID("nop"),
+		TelemetrySettings: componenttest.NewNopTelemetrySettings(),
 	}, cfg, exporter,
 		limiter.processLogs,
 		processorhelper.WithStart(limiter.start),

processor/processorhelper/documentation.md

@@ -54,6 +54,30 @@ Number of spans that were dropped.
 | ---- | ----------- | ---------- | --------- |
 | {spans} | Sum | Int | true |
 
+### otelcol_processor_incoming_log_records
+
+Number of log records passed to the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {records} | Sum | Int | true |
+
+### otelcol_processor_incoming_metric_points
+
+Number of metric points passed to the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {datapoints} | Sum | Int | true |
+
+### otelcol_processor_incoming_spans
+
+Number of spans passed to the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {spans} | Sum | Int | true |
+
 ### otelcol_processor_inserted_log_records
 
 Number of log records that were inserted.
@@ -78,6 +102,30 @@ Number of spans that were inserted.
 | ---- | ----------- | ---------- | --------- |
 | {spans} | Sum | Int | true |
 
+### otelcol_processor_outgoing_log_records
+
+Number of log records emitted from the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {records} | Sum | Int | true |
+
+### otelcol_processor_outgoing_metric_points
+
+Number of metric points emitted from the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {datapoints} | Sum | Int | true |
+
+### otelcol_processor_outgoing_spans
+
+Number of spans emitted from the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {spans} | Sum | Int | true |
+
 ### otelcol_processor_refused_log_records
 
 Number of log records that were rejected by the next component in the pipeline.

processor/processorhelper/logs.go

@@ -39,12 +39,21 @@ func NewLogsProcessor(
 		return nil, errors.New("nil logsFunc")
 	}
 
+	obs, err := newObsReport(ObsReportSettings{
+		ProcessorID:             set.ID,
+		ProcessorCreateSettings: set,
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	eventOptions := spanAttributes(set.ID)
 	bs := fromOptions(options)
 	logsConsumer, err := consumer.NewLogs(func(ctx context.Context, ld plog.Logs) error {
 		span := trace.SpanFromContext(ctx)
 		span.AddEvent("Start processing.", eventOptions)
-		var err error
+		recordsIn := ld.LogRecordCount()
+
 		ld, err = logsFunc(ctx, ld)
 		span.AddEvent("End processing.", eventOptions)
 		if err != nil {
@@ -53,6 +62,8 @@ func NewLogsProcessor(
 			}
 			return err
 		}
+		recordsOut := ld.LogRecordCount()
+		obs.recordInOut(ctx, component.DataTypeLogs, recordsIn, recordsOut)
 		return nextConsumer.ConsumeLogs(ctx, ld)
 	}, bs.consumerOptions...)
 	if err != nil {

processor/processorhelper/logs_test.go

@@ -6,13 +6,20 @@ package processorhelper
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/pdata/plog"
@@ -67,3 +74,77 @@ func newTestLProcessor(retError error) ProcessLogsFunc {
 		return ld, retError
 	}
 }
+
+func TestLogsProcessor_RecordInOut(t *testing.T) {
+	// Regardless of how many logs are ingested, emit just one
+	mockAggregate := func(_ context.Context, _ plog.Logs) (plog.Logs, error) {
+		ld := plog.NewLogs()
+		ld.ResourceLogs().AppendEmpty().ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
+		return ld, nil
+	}
+
+	incomingLogs := plog.NewLogs()
+	incomingLogRecords := incomingLogs.ResourceLogs().AppendEmpty().ScopeLogs().AppendEmpty().LogRecords()
+
+	// Add 3 records to the incoming
+	incomingLogRecords.AppendEmpty()
+	incomingLogRecords.AppendEmpty()
+	incomingLogRecords.AppendEmpty()
+
+	metricReader := sdkmetric.NewManualReader()
+	set := processortest.NewNopSettings()
+	set.TelemetrySettings.MetricsLevel = configtelemetry.LevelBasic
+	set.TelemetrySettings.LeveledMeterProvider = func(level configtelemetry.Level) metric.MeterProvider {
+		if level >= configtelemetry.LevelBasic {
+			return sdkmetric.NewMeterProvider(sdkmetric.WithReader(metricReader))
+		}
+		return nil
+	}
+
+	lp, err := NewLogsProcessor(context.Background(), set, &testLogsCfg, consumertest.NewNop(), mockAggregate)
+	require.NoError(t, err)
+
+	assert.NoError(t, lp.Start(context.Background(), componenttest.NewNopHost()))
+	assert.NoError(t, lp.ConsumeLogs(context.Background(), incomingLogs))
+	assert.NoError(t, lp.Shutdown(context.Background()))
+
+	ownMetrics := new(metricdata.ResourceMetrics)
+	require.NoError(t, metricReader.Collect(context.Background(), ownMetrics))
+
+	require.Len(t, ownMetrics.ScopeMetrics, 1)
+	require.Len(t, ownMetrics.ScopeMetrics[0].Metrics, 2)
+
+	inMetric := ownMetrics.ScopeMetrics[0].Metrics[0]
+	outMetric := ownMetrics.ScopeMetrics[0].Metrics[1]
+	if strings.Contains(inMetric.Name, "outgoing") {
+		inMetric, outMetric = outMetric, inMetric
+	}
+
+	metricdatatest.AssertAggregationsEqual(t, metricdata.Sum[int64]{
+		Temporality: metricdata.CumulativeTemporality,
+		IsMonotonic: true,
+		DataPoints: []metricdata.DataPoint[int64]{
+			{
+				Attributes: attribute.NewSet(attribute.KeyValue{
+					Key:   attribute.Key("processor"),
+					Value: attribute.StringValue(set.ID.String()),
+				}),
+				Value: 3,
+			},
+		},
+	}, inMetric.Data, metricdatatest.IgnoreTimestamp())
+
+	metricdatatest.AssertAggregationsEqual(t, metricdata.Sum[int64]{
+		Temporality: metricdata.CumulativeTemporality,
+		IsMonotonic: true,
+		DataPoints: []metricdata.DataPoint[int64]{
+			{
+				Attributes: attribute.NewSet(attribute.KeyValue{
+					Key:   attribute.Key("processor"),
+					Value: attribute.StringValue(set.ID.String()),
+				}),
+				Value: 1,
+			},
+		},
+	}, outMetric.Data, metricdatatest.IgnoreTimestamp())
+}

processor/processorhelper/metadata.yaml

@@ -9,6 +9,55 @@ status:
 
 telemetry:
   metrics:
+
+    processor_incoming_spans:
+      enabled: true
+      description: Number of spans passed to the processor.
+      unit: "{spans}"
+      sum:
+        value_type: int
+        monotonic: true
+
+    processor_outgoing_spans:
+      enabled: true
+      description: Number of spans emitted from the processor.
+      unit: "{spans}"
+      sum:
+        value_type: int
+        monotonic: true
+
+    processor_incoming_metric_points:
+      enabled: true
+      description: Number of metric points passed to the processor.
+      unit: "{datapoints}"
+      sum:
+        value_type: int
+        monotonic: true
+
+    processor_outgoing_metric_points:
+      enabled: true
+      description: Number of metric points emitted from the processor.
+      unit: "{datapoints}"
+      sum:
+        value_type: int
+        monotonic: true
+
+    processor_incoming_log_records:
+      enabled: true
+      description: Number of log records passed to the processor.
+      unit: "{records}"
+      sum:
+        value_type: int
+        monotonic: true
+
+    processor_outgoing_log_records:
+      enabled: true
+      description: Number of log records emitted from the processor.
+      unit: "{records}"
+      sum:
+        value_type: int
+        monotonic: true
+
     processor_accepted_spans:
       enabled: true
       description: Number of spans successfully pushed into the next component in the pipeline.

processor/processorhelper/metrics.go

@@ -39,12 +39,21 @@ func NewMetricsProcessor(
 		return nil, errors.New("nil metricsFunc")
 	}
 
+	obs, err := newObsReport(ObsReportSettings{
+		ProcessorID:             set.ID,
+		ProcessorCreateSettings: set,
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	eventOptions := spanAttributes(set.ID)
 	bs := fromOptions(options)
 	metricsConsumer, err := consumer.NewMetrics(func(ctx context.Context, md pmetric.Metrics) error {
 		span := trace.SpanFromContext(ctx)
 		span.AddEvent("Start processing.", eventOptions)
-		var err error
+		pointsIn := md.DataPointCount()
+
 		md, err = metricsFunc(ctx, md)
 		span.AddEvent("End processing.", eventOptions)
 		if err != nil {
@@ -53,6 +62,8 @@ func NewMetricsProcessor(
 			}
 			return err
 		}
+		pointsOut := md.DataPointCount()
+		obs.recordInOut(ctx, component.DataTypeMetrics, pointsIn, pointsOut)
 		return nextConsumer.ConsumeMetrics(ctx, md)
 	}, bs.consumerOptions...)
 	if err != nil {