Fix processor metrics not being reported initially with 0 values

.chloggen/fix-processor-metrics.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: processorhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix processor metrics not being reported initially with 0 values.
+
+# One or more tracking issues or pull requests related to the change
+issues: [10855]
+
+# 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]

processor/processorhelper/obsreport.go

@@ -32,8 +32,8 @@ func BuildCustomMetricName(configType, metric string) string {
 
 // ObsReport is a helper to add observability to a processor.
 type ObsReport struct {
-	otelAttrs  []attribute.KeyValue
-	telBuilder *metadata.TelemetryBuilder
+	otelAttrs        []attribute.KeyValue
+	telemetryBuilder *metadata.TelemetryBuilder
 }
 
 // ObsReportSettings are settings for creating an ObsReport.
@@ -48,74 +48,100 @@ func NewObsReport(cfg ObsReportSettings) (*ObsReport, error) {
 }
 
 func newObsReport(cfg ObsReportSettings) (*ObsReport, error) {
-	telBuilder, err := metadata.NewTelemetryBuilder(cfg.ProcessorCreateSettings.TelemetrySettings, metadata.WithLevel(cfg.ProcessorCreateSettings.MetricsLevel))
+	telemetryBuilder, err := metadata.NewTelemetryBuilder(cfg.ProcessorCreateSettings.TelemetrySettings, metadata.WithLevel(cfg.ProcessorCreateSettings.MetricsLevel))
 	if err != nil {
 		return nil, err
 	}
 	return &ObsReport{
 		otelAttrs: []attribute.KeyValue{
 			attribute.String(obsmetrics.ProcessorKey, cfg.ProcessorID.String()),
 		},
-		telBuilder: telBuilder,
+		telemetryBuilder: telemetryBuilder,
 	}, nil
 }
 
+func (or *ObsReport) recordData(ctx context.Context, dataType component.DataType, accepted, refused, dropped, inserted int64) {
+	var acceptedCount, refusedCount, droppedCount, insertedCount metric.Int64Counter
+	switch dataType {
+	case component.DataTypeTraces:
+		acceptedCount = or.telemetryBuilder.ProcessorAcceptedSpans
+		refusedCount = or.telemetryBuilder.ProcessorRefusedSpans
+		droppedCount = or.telemetryBuilder.ProcessorDroppedSpans
+		insertedCount = or.telemetryBuilder.ProcessorInsertedSpans
+	case component.DataTypeMetrics:
+		acceptedCount = or.telemetryBuilder.ProcessorAcceptedMetricPoints
+		refusedCount = or.telemetryBuilder.ProcessorRefusedMetricPoints
+		droppedCount = or.telemetryBuilder.ProcessorDroppedMetricPoints
+		insertedCount = or.telemetryBuilder.ProcessorInsertedMetricPoints
+	case component.DataTypeLogs:
+		acceptedCount = or.telemetryBuilder.ProcessorAcceptedLogRecords
+		refusedCount = or.telemetryBuilder.ProcessorRefusedLogRecords
+		droppedCount = or.telemetryBuilder.ProcessorDroppedLogRecords
+		insertedCount = or.telemetryBuilder.ProcessorInsertedLogRecords
+	}
+
+	acceptedCount.Add(ctx, accepted, metric.WithAttributes(or.otelAttrs...))
+	refusedCount.Add(ctx, refused, metric.WithAttributes(or.otelAttrs...))
+	droppedCount.Add(ctx, dropped, metric.WithAttributes(or.otelAttrs...))
+	insertedCount.Add(ctx, inserted, metric.WithAttributes(or.otelAttrs...))
+}
+
 // TracesAccepted reports that the trace data was accepted.
 func (or *ObsReport) TracesAccepted(ctx context.Context, numSpans int) {
-	or.telBuilder.ProcessorAcceptedSpans.Add(ctx, int64(numSpans), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeTraces, int64(numSpans), int64(0), int64(0), int64(0))
 }
 
 // TracesRefused reports that the trace data was refused.
 func (or *ObsReport) TracesRefused(ctx context.Context, numSpans int) {
-	or.telBuilder.ProcessorRefusedSpans.Add(ctx, int64(numSpans), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeTraces, int64(0), int64(numSpans), int64(0), int64(0))
 }
 
 // TracesDropped reports that the trace data was dropped.
 func (or *ObsReport) TracesDropped(ctx context.Context, numSpans int) {
-	or.telBuilder.ProcessorDroppedSpans.Add(ctx, int64(numSpans), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeTraces, int64(0), int64(0), int64(numSpans), int64(0))
 }
 
 // TracesInserted reports that the trace data was inserted.
 func (or *ObsReport) TracesInserted(ctx context.Context, numSpans int) {
-	or.telBuilder.ProcessorInsertedSpans.Add(ctx, int64(numSpans), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeTraces, int64(0), int64(0), int64(0), int64(numSpans))
 }
 
 // MetricsAccepted reports that the metrics were accepted.
 func (or *ObsReport) MetricsAccepted(ctx context.Context, numPoints int) {
-	or.telBuilder.ProcessorAcceptedMetricPoints.Add(ctx, int64(numPoints), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeMetrics, int64(numPoints), int64(0), int64(0), int64(0))
 }
 
 // MetricsRefused reports that the metrics were refused.
 func (or *ObsReport) MetricsRefused(ctx context.Context, numPoints int) {
-	or.telBuilder.ProcessorRefusedMetricPoints.Add(ctx, int64(numPoints), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeMetrics, int64(0), int64(numPoints), int64(0), int64(0))
 }
 
 // MetricsDropped reports that the metrics were dropped.
 func (or *ObsReport) MetricsDropped(ctx context.Context, numPoints int) {
-	or.telBuilder.ProcessorDroppedMetricPoints.Add(ctx, int64(numPoints), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeMetrics, int64(0), int64(0), int64(numPoints), int64(0))
 }
 
 // MetricsInserted reports that the metrics were inserted.
 func (or *ObsReport) MetricsInserted(ctx context.Context, numPoints int) {
-	or.telBuilder.ProcessorInsertedMetricPoints.Add(ctx, int64(numPoints), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeMetrics, int64(0), int64(0), int64(0), int64(numPoints))
 }
 
 // LogsAccepted reports that the logs were accepted.
 func (or *ObsReport) LogsAccepted(ctx context.Context, numRecords int) {
-	or.telBuilder.ProcessorAcceptedLogRecords.Add(ctx, int64(numRecords), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeLogs, int64(numRecords), int64(0), int64(0), int64(0))
 }
 
 // LogsRefused reports that the logs were refused.
 func (or *ObsReport) LogsRefused(ctx context.Context, numRecords int) {
-	or.telBuilder.ProcessorRefusedLogRecords.Add(ctx, int64(numRecords), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeLogs, int64(0), int64(numRecords), int64(0), int64(0))
 }
 
 // LogsDropped reports that the logs were dropped.
 func (or *ObsReport) LogsDropped(ctx context.Context, numRecords int) {
-	or.telBuilder.ProcessorDroppedLogRecords.Add(ctx, int64(numRecords), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeLogs, int64(0), int64(0), int64(numRecords), int64(0))
 }
 
 // LogsInserted reports that the logs were inserted.
 func (or *ObsReport) LogsInserted(ctx context.Context, numRecords int) {
-	or.telBuilder.ProcessorInsertedLogRecords.Add(ctx, int64(numRecords), metric.WithAttributes(or.otelAttrs...))
+	or.recordData(ctx, component.DataTypeLogs, int64(0), int64(0), int64(0), int64(numRecords))
 }