[batchprocessor] ensure attributes are set on metadata metric

Fixing this bug required a change in mdatagen to ensure attributes could be passed in to the telemetry builder.

Fixes #9674

Signed-off-by: Alex Boten <223565+codeboten@users.noreply.github.com>

.chloggen/codeboten_fix-9674-2.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: mdatagen
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: support setting an AttributeSet for async instruments
+
+# One or more tracking issues or pull requests related to the change
+issues: [9674]
+
+# (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: []

.chloggen/codeboten_fix-9674.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: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: batchprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: ensure attributes are set on cardinality metadata metric
+
+# One or more tracking issues or pull requests related to the change
+issues: [9674]
+
+# (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: []

cmd/mdatagen/main.go

@@ -189,6 +189,14 @@ func templatize(tmplFile string, md metadata) *template.Template {
 					}
 					return result
 				},
+				"hasAsync": func(t telemetry) bool {
+					for _, m := range t.Metrics {
+						if m.Data().IsAsync() {
+							return true
+						}
+					}
+					return false
+				},
 				"inc": func(i int) int { return i + 1 },
 				"distroURL": func(name string) string {
 					return distros[name]

cmd/mdatagen/templates/telemetry.go.tmpl

@@ -8,6 +8,7 @@ import (
     "errors"
     {{- end }}
 
+    {{ if hasAsync .Telemetry }}"go.opentelemetry.io/otel/attribute"{{- end }}
 	"go.opentelemetry.io/otel/metric"
     "go.opentelemetry.io/otel/metric/noop"
 	"go.opentelemetry.io/otel/trace"
@@ -35,6 +36,7 @@ type TelemetryBuilder struct {
     {{- end }}
 	{{- end }}
     level configtelemetry.Level
+    {{ if hasAsync .Telemetry }}attributeSet attribute.Set{{- end }}
 }
 
 // telemetryBuilderOption applies changes to default builder.
@@ -47,6 +49,15 @@ func WithLevel(lvl configtelemetry.Level) telemetryBuilderOption {
     }
 }
 
+{{- if hasAsync .Telemetry }}
+// WithAttributeSet applies a set of attributes for asynchronous instruments.
+func WithAttributeSet(set attribute.Set) telemetryBuilderOption {
+	return func(builder *TelemetryBuilder) {
+		builder.attributeSet = set
+	}
+}
+{{- end }}
+
 {{- range $name, $metric := .Telemetry.Metrics }}
     {{ if $metric.Data.Async -}}
 // With{{ $name.Render }}Callback sets callback for observable {{ $name.Render }} metric.
@@ -58,6 +69,7 @@ func With{{ $name.Render }}Callback(cb func() {{ $metric.Data.BasicType }}) tele
     {{- end }}
 {{- end }}
 
+
 // NewTelemetryBuilder provides a struct with methods to update all internal telemetry
 // for a component
 func NewTelemetryBuilder(settings component.TelemetrySettings, options ...telemetryBuilderOption) (*TelemetryBuilder, error) {
@@ -85,7 +97,7 @@ func NewTelemetryBuilder(settings component.TelemetrySettings, options ...teleme
         {{- end }}
         {{ if $metric.Data.Async -}}
         metric.With{{ casesTitle $metric.Data.BasicType }}Callback(func(_ context.Context, o metric.{{ casesTitle $metric.Data.BasicType }}Observer) error {
-            o.Observe(builder.observe{{ $name.Render }}())
+            o.Observe(builder.observe{{ $name.Render }}(), metric.WithAttributeSet(builder.attributeSet))
             return nil
         }),
         {{- end }}

processor/batchprocessor/batch_processor_test.go

@@ -275,7 +275,8 @@ func TestBatchProcessorSentBySize(t *testing.T) {
 				IsMonotonic: false,
 				DataPoints: []metricdata.DataPoint[int64]{
 					{
-						Value: 1,
+						Value:      1,
+						Attributes: attribute.NewSet(attribute.String("processor", "batch")),
 					},
 				},
 			},

processor/batchprocessor/metrics.go

@@ -28,38 +28,41 @@ type batchProcessorTelemetry struct {
 
 	exportCtx context.Context
 
-	processorAttr    []attribute.KeyValue
+	processorAttr    attribute.Set
 	telemetryBuilder *metadata.TelemetryBuilder
 }
 
 func newBatchProcessorTelemetry(set processor.CreateSettings, currentMetadataCardinality func() int) (*batchProcessorTelemetry, error) {
+	attrs := attribute.NewSet(attribute.String(obsmetrics.ProcessorKey, set.ID.String()))
+
 	telemetryBuilder, err := metadata.NewTelemetryBuilder(set.TelemetrySettings,
 		metadata.WithLevel(set.MetricsLevel),
 		metadata.WithProcessorBatchMetadataCardinalityCallback(func() int64 { return int64(currentMetadataCardinality()) }),
+		metadata.WithAttributeSet(attrs),
 	)
 
 	if err != nil {
 		return nil, err
 	}
 
 	return &batchProcessorTelemetry{
-		processorAttr:    []attribute.KeyValue{attribute.String(obsmetrics.ProcessorKey, set.ID.String())},
 		exportCtx:        context.Background(),
 		detailed:         set.MetricsLevel == configtelemetry.LevelDetailed,
 		telemetryBuilder: telemetryBuilder,
+		processorAttr:    attrs,
 	}, nil
 }
 
 func (bpt *batchProcessorTelemetry) record(trigger trigger, sent, bytes int64) {
 	switch trigger {
 	case triggerBatchSize:
-		bpt.telemetryBuilder.ProcessorBatchBatchSizeTriggerSend.Add(bpt.exportCtx, 1, metric.WithAttributes(bpt.processorAttr...))
+		bpt.telemetryBuilder.ProcessorBatchBatchSizeTriggerSend.Add(bpt.exportCtx, 1, metric.WithAttributeSet(bpt.processorAttr))
 	case triggerTimeout:
-		bpt.telemetryBuilder.ProcessorBatchTimeoutTriggerSend.Add(bpt.exportCtx, 1, metric.WithAttributes(bpt.processorAttr...))
+		bpt.telemetryBuilder.ProcessorBatchTimeoutTriggerSend.Add(bpt.exportCtx, 1, metric.WithAttributeSet(bpt.processorAttr))
 	}
 
-	bpt.telemetryBuilder.ProcessorBatchBatchSendSize.Record(bpt.exportCtx, sent, metric.WithAttributes(bpt.processorAttr...))
+	bpt.telemetryBuilder.ProcessorBatchBatchSendSize.Record(bpt.exportCtx, sent, metric.WithAttributeSet(bpt.processorAttr))
 	if bpt.detailed {
-		bpt.telemetryBuilder.ProcessorBatchBatchSendSizeBytes.Record(bpt.exportCtx, bytes, metric.WithAttributes(bpt.processorAttr...))
+		bpt.telemetryBuilder.ProcessorBatchBatchSendSizeBytes.Record(bpt.exportCtx, bytes, metric.WithAttributeSet(bpt.processorAttr))
 	}
 }