Add processor duration metric.

Author: andborja

.chloggen/processor-duration.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. otlpreceiver)
+component: processorhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add processor duration metric.
+
+# One or more tracking issues or pull requests related to the change
+issues: [13231]
+
+# (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: [user]

docs/rfcs/component-universal-telemetry.md

@@ -80,11 +80,12 @@ package is internal, then the scope name should be that of the module which cont
 
 ### Auto-Instrumented Metrics
 
-There are two straightforward measurements that can be made on any pdata:
+There are a few measurements that can be made on any pdata:
 
-1. A count of "items" (spans, data points, or log records). These are low cost but broadly useful, so they should be enabled by default.
-2. A measure of size, based on [ProtoMarshaler.Sizer()](https://github.com/open-telemetry/opentelemetry-collector/blob/9907ba50df0d5853c34d2962cf21da42e15a560d/pdata/ptrace/pb.go#l11).
+1. Count of "items" (spans, data points, or log records). These are low cost but broadly useful, so they should be enabled by default.
+1. Measure of size, based on [ProtoMarshaler.Sizer()](https://github.com/open-telemetry/opentelemetry-collector/blob/9907ba50df0d5853c34d2962cf21da42e15a560d/pdata/ptrace/pb.go#l11).
   These may be high cost to compute, so by default they should be disabled (and not calculated). This default setting may change in the future if it is demonstrated that the cost is generally acceptable.
+1. Duration. It is the time it takes to process items in a component. Duration measurement is especially useful for tuning processors with high computational overhead.
 
 The location of these measurements can be described in terms of whether the data is "consumed" or "produced", from the perspective of the
 component to which the telemetry is attributed. Metrics which contain the term "produced" describe data which is emitted from the component,

processor/processorhelper/documentation.md

@@ -6,6 +6,14 @@
 
 The following telemetry is emitted by this component.
 
+### otelcol_processor_duration
+
+Time taken by the processor to complete. [alpha]
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| s | Histogram | Double |
+
 ### otelcol_processor_incoming_items
 
 Number of items passed to the processor. [alpha]

processor/processorhelper/generated_package_test.go

@@ -49,6 +49,10 @@ func NewLogs(
 	logsConsumer, err := consumer.NewLogs(func(ctx context.Context, ld plog.Logs) error {
 		span := trace.SpanFromContext(ctx)
 		span.AddEvent("Start processing.", eventOptions)
+
+		ps := newProcessorStart()
+		defer obs.recordDuration(ctx, ps)
+
 		recordsIn := ld.LogRecordCount()
 
 		var errFunc error

processor/processorhelper/internal/metadata/generated_telemetry.go

@@ -183,6 +183,40 @@ func TestLogs_RecordIn_ErrorOut(t *testing.T) {
 		}, metricdatatest.IgnoreTimestamp())
 }
 
+func TestLogs_ProcessDuration(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()
+
+	tel := componenttest.NewTelemetry()
+	lp, err := NewLogs(context.Background(), newSettings(tel), &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()))
+
+	metadatatest.AssertEqualProcessorDuration(t, tel,
+		[]metricdata.HistogramDataPoint[float64]{
+			{
+				Count:        1,
+				BucketCounts: []uint64{1},
+				Attributes:   attribute.NewSet(attribute.String("processor", "processorhelper"), attribute.String("otel.signal", "logs")),
+			},
+		}, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreValue())
+}
+
 func newSettings(tel *componenttest.Telemetry) processor.Settings {
 	set := processortest.NewNopSettings(component.MustNewType("processorhelper"))
 	set.TelemetrySettings = tel.NewTelemetrySettings()

processor/processorhelper/internal/metadatatest/generated_telemetrytest.go

@@ -28,3 +28,13 @@ telemetry:
       sum:
         value_type: int
         monotonic: true
+
+    processor_duration:
+      enabled: true
+      stability:
+        level: alpha
+      description: Time taken by the processor to complete.
+      unit: s
+      histogram:
+        async: false
+        value_type: double

processor/processorhelper/internal/metadatatest/generated_telemetrytest_test.go

@@ -49,6 +49,10 @@ func NewMetrics(
 	metricsConsumer, err := consumer.NewMetrics(func(ctx context.Context, md pmetric.Metrics) error {
 		span := trace.SpanFromContext(ctx)
 		span.AddEvent("Start processing.", eventOptions)
+
+		ps := newProcessorStart()
+		defer obs.recordDuration(ctx, ps)
+
 		pointsIn := md.DataPointCount()
 
 		var errFunc error

processor/processorhelper/logs.go

@@ -180,3 +180,38 @@ func TestMetrics_RecordIn_ErrorOut(t *testing.T) {
 			},
 		}, metricdatatest.IgnoreTimestamp())
 }
+
+func TestMetrics_ProcesDuration(t *testing.T) {
+	// Regardless of how many data points are ingested, emit 3
+	mockAggregate := func(_ context.Context, _ pmetric.Metrics) (pmetric.Metrics, error) {
+		md := pmetric.NewMetrics()
+		md.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty().SetEmptySum().DataPoints().AppendEmpty()
+		md.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty().SetEmptySum().DataPoints().AppendEmpty()
+		md.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty().SetEmptySum().DataPoints().AppendEmpty()
+		return md, nil
+	}
+
+	incomingMetrics := pmetric.NewMetrics()
+	dps := incomingMetrics.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty().SetEmptySum().DataPoints()
+
+	// Add 2 data points to the incoming
+	dps.AppendEmpty()
+	dps.AppendEmpty()
+
+	tel := componenttest.NewTelemetry()
+	mp, err := NewMetrics(context.Background(), newSettings(tel), &testMetricsCfg, consumertest.NewNop(), mockAggregate)
+	require.NoError(t, err)
+
+	assert.NoError(t, mp.Start(context.Background(), componenttest.NewNopHost()))
+	assert.NoError(t, mp.ConsumeMetrics(context.Background(), incomingMetrics))
+	assert.NoError(t, mp.Shutdown(context.Background()))
+
+	metadatatest.AssertEqualProcessorDuration(t, tel,
+		[]metricdata.HistogramDataPoint[float64]{
+			{
+				Count:        1,
+				BucketCounts: []uint64{1},
+				Attributes:   attribute.NewSet(attribute.String("processor", "processorhelper"), attribute.String("otel.signal", "metrics")),
+			},
+		}, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreValue())
+}

processor/processorhelper/logs_test.go

@@ -5,6 +5,7 @@ package processorhelper // import "go.opentelemetry.io/collector/processor/proce
 
 import (
 	"context"
+	"time"
 
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -40,3 +41,17 @@ func (or *obsReport) recordInOut(ctx context.Context, incoming, outgoing int) {
 	or.telemetryBuilder.ProcessorIncomingItems.Add(ctx, int64(incoming), or.otelAttrs)
 	or.telemetryBuilder.ProcessorOutgoingItems.Add(ctx, int64(outgoing), or.otelAttrs)
 }
+
+func (or *obsReport) recordDuration(ctx context.Context, ps processorStart) {
+	duration := time.Since(ps.Time)
+	durationSecs := float64(duration.Microseconds()) / 1e6 // Convert to seconds
+	or.telemetryBuilder.ProcessorDuration.Record(ctx, float64(durationSecs), or.otelAttrs)
+}
+
+type processorStart struct {
+	time.Time
+}
+
+func newProcessorStart() processorStart {
+	return processorStart{time.Now()}
+}

processor/processorhelper/metadata.yaml

@@ -49,6 +49,10 @@ func NewTraces(
 	traceConsumer, err := consumer.NewTraces(func(ctx context.Context, td ptrace.Traces) error {
 		span := trace.SpanFromContext(ctx)
 		span.AddEvent("Start processing.", eventOptions)
+
+		ps := newProcessorStart()
+		defer obs.recordDuration(ctx, ps)
+
 		spansIn := td.SpanCount()
 
 		var errFunc error

processor/processorhelper/metrics.go

@@ -184,3 +184,38 @@ func TestTraces_RecordIn_ErrorOut(t *testing.T) {
 			},
 		}, metricdatatest.IgnoreTimestamp())
 }
+
+func TestTraces_ProcessDuration(t *testing.T) {
+	// Regardless of how many spans are ingested, emit just one
+	mockAggregate := func(_ context.Context, _ ptrace.Traces) (ptrace.Traces, error) {
+		td := ptrace.NewTraces()
+		td.ResourceSpans().AppendEmpty().ScopeSpans().AppendEmpty().Spans().AppendEmpty()
+		return td, nil
+	}
+
+	incomingTraces := ptrace.NewTraces()
+	incomingSpans := incomingTraces.ResourceSpans().AppendEmpty().ScopeSpans().AppendEmpty().Spans()
+
+	// Add 4 records to the incoming
+	incomingSpans.AppendEmpty()
+	incomingSpans.AppendEmpty()
+	incomingSpans.AppendEmpty()
+	incomingSpans.AppendEmpty()
+
+	tel := componenttest.NewTelemetry()
+	tp, err := NewTraces(context.Background(), newSettings(tel), &testLogsCfg, consumertest.NewNop(), mockAggregate)
+	require.NoError(t, err)
+
+	assert.NoError(t, tp.Start(context.Background(), componenttest.NewNopHost()))
+	assert.NoError(t, tp.ConsumeTraces(context.Background(), incomingTraces))
+	assert.NoError(t, tp.Shutdown(context.Background()))
+
+	metadatatest.AssertEqualProcessorDuration(t, tel,
+		[]metricdata.HistogramDataPoint[float64]{
+			{
+				Count:        1,
+				BucketCounts: []uint64{1},
+				Attributes:   attribute.NewSet(attribute.String("processor", "processorhelper"), attribute.String("otel.signal", "traces")),
+			},
+		}, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreValue())
+}