[processorhelper] Add throughput metrics

processor/processorhelper/documentation.md

@@ -54,6 +54,14 @@ Number of spans that were dropped.
 | ---- | ----------- | ---------- | --------- |
 | {spans} | Sum | Int | true |
 
+### otelcol_processor_incoming_log_record_size
+
+Total size of log records passed to the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| By | Sum | Int | true |
+
 ### otelcol_processor_incoming_log_records
 
 Number of log records passed to the processor.
@@ -62,6 +70,14 @@ Number of log records passed to the processor.
 | ---- | ----------- | ---------- | --------- |
 | {records} | Sum | Int | true |
 
+### otelcol_processor_incoming_metric_point_size
+
+Total size of metric points passed to the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| By | Sum | Int | true |
+
 ### otelcol_processor_incoming_metric_points
 
 Number of metric points passed to the processor.
@@ -70,6 +86,14 @@ Number of metric points passed to the processor.
 | ---- | ----------- | ---------- | --------- |
 | {datapoints} | Sum | Int | true |
 
+### otelcol_processor_incoming_span_size
+
+Total size of spans passed to the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| By | Sum | Int | true |
+
 ### otelcol_processor_incoming_spans
 
 Number of spans passed to the processor.
@@ -102,6 +126,14 @@ Number of spans that were inserted.
 | ---- | ----------- | ---------- | --------- |
 | {spans} | Sum | Int | true |
 
+### otelcol_processor_outgoing_log_record_size
+
+Total size of log records emitted from the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| By | Sum | Int | true |
+
 ### otelcol_processor_outgoing_log_records
 
 Number of log records emitted from the processor.
@@ -110,6 +142,14 @@ Number of log records emitted from the processor.
 | ---- | ----------- | ---------- | --------- |
 | {records} | Sum | Int | true |
 
+### otelcol_processor_outgoing_metric_point_size
+
+Total size of metric points emitted from the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| By | Sum | Int | true |
+
 ### otelcol_processor_outgoing_metric_points
 
 Number of metric points emitted from the processor.
@@ -118,6 +158,14 @@ Number of metric points emitted from the processor.
 | ---- | ----------- | ---------- | --------- |
 | {datapoints} | Sum | Int | true |
 
+### otelcol_processor_outgoing_span_size
+
+Total size of spans emitted from the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| By | Sum | Int | true |
+
 ### otelcol_processor_outgoing_spans
 
 Number of spans emitted from the processor.

processor/processorhelper/logs.go

@@ -10,6 +10,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/processor"
@@ -49,10 +50,16 @@ func NewLogsProcessor(
 
 	eventOptions := spanAttributes(set.ID)
 	bs := fromOptions(options)
+	marshaler := new(plog.ProtoMarshaler)
 	logsConsumer, err := consumer.NewLogs(func(ctx context.Context, ld plog.Logs) error {
+		var recordsIn, recordsOut, bytesIn, bytesOut int
+
 		span := trace.SpanFromContext(ctx)
 		span.AddEvent("Start processing.", eventOptions)
-		recordsIn := ld.LogRecordCount()
+		recordsIn = ld.LogRecordCount()
+		if obs.metricLevel >= configtelemetry.LevelDetailed {
+			bytesIn = marshaler.LogsSize(ld)
+		}
 
 		ld, err = logsFunc(ctx, ld)
 		span.AddEvent("End processing.", eventOptions)
@@ -62,8 +69,12 @@ func NewLogsProcessor(
 			}
 			return err
 		}
-		recordsOut := ld.LogRecordCount()
+		recordsOut = ld.LogRecordCount()
 		obs.recordInOut(ctx, component.DataTypeLogs, recordsIn, recordsOut)
+		if obs.metricLevel >= configtelemetry.LevelDetailed {
+			bytesOut = marshaler.LogsSize(ld)
+			obs.recordInOutSize(ctx, component.DataTypeLogs, bytesIn, bytesOut)
+		}
 		return nextConsumer.ConsumeLogs(ctx, ld)
 	}, bs.consumerOptions...)
 	if err != nil {

processor/processorhelper/metadata.yaml

@@ -58,6 +58,60 @@ telemetry:
         value_type: int
         monotonic: true
 
+    processor_incoming_span_size:
+      enabled: true
+      level: detailed
+      description: Total size of spans passed to the processor.
+      unit: "By"
+      sum:
+        value_type: int
+        monotonic: true
+
+    processor_outgoing_span_size:
+      enabled: true
+      level: detailed
+      description: Total size of spans emitted from the processor.
+      unit: "By"
+      sum:
+        value_type: int
+        monotonic: true
+
+    processor_incoming_metric_point_size:
+      enabled: true
+      level: detailed
+      description: Total size of metric points passed to the processor.
+      unit: "By"
+      sum:
+        value_type: int
+        monotonic: true
+
+    processor_outgoing_metric_point_size:
+      enabled: true
+      level: detailed
+      description: Total size of metric points emitted from the processor.
+      unit: "By"
+      sum:
+        value_type: int
+        monotonic: true
+
+    processor_incoming_log_record_size:
+      enabled: true
+      level: detailed
+      description: Total size of log records passed to the processor.
+      unit: "By"
+      sum:
+        value_type: int
+        monotonic: true
+
+    processor_outgoing_log_record_size:
+      enabled: true
+      level: detailed
+      description: Total size of log records emitted from the processor.
+      unit: "By"
+      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

@@ -10,6 +10,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/processor"
@@ -49,11 +50,16 @@ func NewMetricsProcessor(
 
 	eventOptions := spanAttributes(set.ID)
 	bs := fromOptions(options)
+	marshaler := new(pmetric.ProtoMarshaler)
 	metricsConsumer, err := consumer.NewMetrics(func(ctx context.Context, md pmetric.Metrics) error {
+		var pointsIn, pointsOut, bytesIn, bytesOut int
+
 		span := trace.SpanFromContext(ctx)
 		span.AddEvent("Start processing.", eventOptions)
-		pointsIn := md.DataPointCount()
-
+		pointsIn = md.DataPointCount()
+		if obs.metricLevel >= configtelemetry.LevelDetailed {
+			bytesIn = marshaler.MetricsSize(md)
+		}
 		md, err = metricsFunc(ctx, md)
 		span.AddEvent("End processing.", eventOptions)
 		if err != nil {
@@ -62,8 +68,12 @@ func NewMetricsProcessor(
 			}
 			return err
 		}
-		pointsOut := md.DataPointCount()
+		pointsOut = md.DataPointCount()
 		obs.recordInOut(ctx, component.DataTypeMetrics, pointsIn, pointsOut)
+		if obs.metricLevel >= configtelemetry.LevelDetailed {
+			bytesOut = marshaler.MetricsSize(md)
+			obs.recordInOutSize(ctx, component.DataTypeTraces, bytesIn, bytesOut)
+		}
 		return nextConsumer.ConsumeMetrics(ctx, md)
 	}, bs.consumerOptions...)
 	if err != nil {