[exporterhelper] Add data_type attribute to internal queue metric

Add data_type attribute to the internal otelcol_exporter_queue_size metric to report the type of data being processed.

All other metrics have the data type reported as part of their names. We could've done the same for queue metrics, but that would introduce a significant breaking change. We want to avoid that until we have all the metrics standardized with OpenTelemetry semantic conventions.

.chloggen/exporterhelper-report-data-type-in-queue-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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: exporterhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add data_type attribute to `otelcol_exporter_queue_size` metric to report the type of data being processed.
+
+# One or more tracking issues or pull requests related to the change
+issues: [9943]
+
+# 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]

component/componenttest/obsreporttest.go

@@ -66,14 +66,26 @@ func (tts *TestTelemetry) CheckExporterEnqueueFailedLogs(enqueueFailed int64) er
 	return tts.prometheusChecker.checkExporterEnqueueFailed(tts.id, "log_records", enqueueFailed)
 }
 
+func (tts *TestTelemetry) CheckExporterQueueSizeMetric(size int64) error {
+	return tts.prometheusChecker.checkExporterQueueSize(tts.id, component.DataTypeMetrics, size)
+}
+
+func (tts *TestTelemetry) CheckExporterQueueSizeTraces(size int64) error {
+	return tts.prometheusChecker.checkExporterQueueSize(tts.id, component.DataTypeTraces, size)
+}
+
+func (tts *TestTelemetry) CheckExporterQueueSizeLogs(size int64) error {
+	return tts.prometheusChecker.checkExporterQueueSize(tts.id, component.DataTypeLogs, size)
+}
+
 // CheckExporterLogs checks that for the current exported values for logs exporter metrics match given values.
 // When this function is called it is required to also call SetupTelemetry as first thing.
 func (tts *TestTelemetry) CheckExporterLogs(sentLogRecords, sendFailedLogRecords int64) error {
 	return tts.prometheusChecker.checkExporterLogs(tts.id, sentLogRecords, sendFailedLogRecords)
 }
 
 func (tts *TestTelemetry) CheckExporterMetricGauge(metric string, val int64) error {
-	return tts.prometheusChecker.checkExporterMetricGauge(tts.id, metric, val)
+	return tts.prometheusChecker.checkGauge(metric, val, attributesForExporterMetrics(tts.id))
 }
 
 // CheckProcessorTraces checks that for the current exported values for trace exporter metrics match given values.

component/componenttest/otelprometheuschecker.go

@@ -92,6 +92,12 @@ func (pc *prometheusChecker) checkExporter(exporter component.ID, datatype strin
 	return errs
 }
 
+func (pc *prometheusChecker) checkExporterQueueSize(exporter component.ID, datatype component.DataType, size int64) error {
+	attrs := attributesForExporterMetrics(exporter)
+	attrs = append(attrs, attribute.String("data_type", datatype.String()))
+	return pc.checkGauge("otelcol_exporter_queue_size", size, attrs)
+}
+
 func (pc *prometheusChecker) checkExporterEnqueueFailed(exporter component.ID, datatype string, enqueueFailed int64) error {
 	if enqueueFailed == 0 {
 		return nil
@@ -100,10 +106,8 @@ func (pc *prometheusChecker) checkExporterEnqueueFailed(exporter component.ID, d
 	return pc.checkCounter(fmt.Sprintf("exporter_enqueue_failed_%s", datatype), enqueueFailed, exporterAttrs)
 }
 
-func (pc *prometheusChecker) checkExporterMetricGauge(exporter component.ID, metric string, val int64) error {
-	exporterAttrs := attributesForExporterMetrics(exporter)
-
-	ts, err := pc.getMetric(metric, io_prometheus_client.MetricType_GAUGE, exporterAttrs)
+func (pc *prometheusChecker) checkGauge(metric string, val int64, attrs []attribute.KeyValue) error {
+	ts, err := pc.getMetric(metric, io_prometheus_client.MetricType_GAUGE, attrs)
 	if err != nil {
 		return err
 	}

exporter/exporterhelper/common.go

@@ -110,7 +110,7 @@ func WithQueue(config QueueSettings) Option {
 			NumConsumers: config.NumConsumers,
 			QueueSize:    config.QueueSize,
 		})
-		o.queueSender = newQueueSender(q, o.set, config.NumConsumers, o.exportFailureMessage, o.obsrep.telemetryBuilder)
+		o.queueSender = newQueueSender(q, o.set, config.NumConsumers, o.exportFailureMessage, o.obsrep)
 		return nil
 	}
 }
@@ -132,7 +132,7 @@ func WithRequestQueue(cfg exporterqueue.Config, queueFactory exporterqueue.Facto
 			DataType:         o.signal,
 			ExporterSettings: o.set,
 		}
-		o.queueSender = newQueueSender(queueFactory(context.Background(), set, cfg), o.set, cfg.NumConsumers, o.exportFailureMessage, o.obsrep.telemetryBuilder)
+		o.queueSender = newQueueSender(queueFactory(context.Background(), set, cfg), o.set, cfg.NumConsumers, o.exportFailureMessage, o.obsrep)
 		return nil
 	}
 }
@@ -250,7 +250,7 @@ type baseExporter struct {
 }
 
 func newBaseExporter(set exporter.Settings, signal component.DataType, osf obsrepSenderFactory, options ...Option) (*baseExporter, error) {
-	obsReport, err := NewObsReport(ObsReportSettings{ExporterID: set.ID, ExporterCreateSettings: set})
+	obsReport, err := NewObsReport(ObsReportSettings{ExporterID: set.ID, ExporterCreateSettings: set, DataType: signal})
 	if err != nil {
 		return nil, err
 	}

exporter/exporterhelper/obsexporter.go

@@ -26,6 +26,7 @@ type ObsReport struct {
 	level          configtelemetry.Level
 	spanNamePrefix string
 	tracer         trace.Tracer
+	dataType       component.DataType
 
 	otelAttrs        []attribute.KeyValue
 	telemetryBuilder *metadata.TelemetryBuilder
@@ -38,6 +39,7 @@ type ObsReport struct {
 type ObsReportSettings struct {
 	ExporterID             component.ID
 	ExporterCreateSettings exporter.Settings
+	DataType               component.DataType
 }
 
 // NewObsReport creates a new Exporter.
@@ -58,7 +60,7 @@ func newExporter(cfg ObsReportSettings) (*ObsReport, error) {
 		level:          cfg.ExporterCreateSettings.TelemetrySettings.MetricsLevel,
 		spanNamePrefix: obsmetrics.ExporterPrefix + cfg.ExporterID.String(),
 		tracer:         cfg.ExporterCreateSettings.TracerProvider.Tracer(cfg.ExporterID.String()),
-
+		dataType:       cfg.DataType,
 		otelAttrs: []attribute.KeyValue{
 			attribute.String(obsmetrics.ExporterKey, cfg.ExporterID.String()),
 		},

exporter/exporterhelper/queue_sender.go

@@ -15,7 +15,6 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadata"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
 	"go.opentelemetry.io/collector/exporter/internal/queue"
 	"go.opentelemetry.io/collector/internal/obsreportconfig/obsmetrics"
@@ -74,18 +73,18 @@ type queueSender struct {
 	traceAttribute attribute.KeyValue
 	consumers      *queue.Consumers[Request]
 
-	telemetryBuilder *metadata.TelemetryBuilder
-	exporterID       component.ID
+	obsrep     *ObsReport
+	exporterID component.ID
 }
 
 func newQueueSender(q exporterqueue.Queue[Request], set exporter.Settings, numConsumers int,
-	exportFailureMessage string, telemetryBuilder *metadata.TelemetryBuilder) *queueSender {
+	exportFailureMessage string, obsrep *ObsReport) *queueSender {
 	qs := &queueSender{
-		queue:            q,
-		numConsumers:     numConsumers,
-		traceAttribute:   attribute.String(obsmetrics.ExporterKey, set.ID.String()),
-		telemetryBuilder: telemetryBuilder,
-		exporterID:       set.ID,
+		queue:          q,
+		numConsumers:   numConsumers,
+		traceAttribute: attribute.String(obsmetrics.ExporterKey, set.ID.String()),
+		obsrep:         obsrep,
+		exporterID:     set.ID,
 	}
 	consumeFunc := func(ctx context.Context, req Request) error {
 		err := qs.nextSender.send(ctx, req)
@@ -105,10 +104,12 @@ func (qs *queueSender) Start(ctx context.Context, host component.Host) error {
 		return err
 	}
 
-	opts := metric.WithAttributeSet(attribute.NewSet(attribute.String(obsmetrics.ExporterKey, qs.exporterID.String())))
+	dataTypeAttr := attribute.String(obsmetrics.DataTypeKey, qs.obsrep.dataType.String())
 	return multierr.Append(
-		qs.telemetryBuilder.InitExporterQueueSize(func() int64 { return int64(qs.queue.Size()) }, opts),
-		qs.telemetryBuilder.InitExporterQueueCapacity(func() int64 { return int64(qs.queue.Capacity()) }, opts),
+		qs.obsrep.telemetryBuilder.InitExporterQueueSize(func() int64 { return int64(qs.queue.Size()) },
+			metric.WithAttributeSet(attribute.NewSet(qs.traceAttribute, dataTypeAttr))),
+		qs.obsrep.telemetryBuilder.InitExporterQueueCapacity(func() int64 { return int64(qs.queue.Capacity()) },
+			metric.WithAttributeSet(attribute.NewSet(qs.traceAttribute))),
 	)
 }
 

exporter/exporterhelper/queue_sender_test.go

@@ -18,7 +18,6 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/exporter"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadata"
 	"go.opentelemetry.io/collector/exporter/exporterqueue"
 	"go.opentelemetry.io/collector/exporter/exportertest"
 	"go.opentelemetry.io/collector/exporter/internal/queue"
@@ -221,7 +220,7 @@ func TestQueuedRetry_QueueMetricsReported(t *testing.T) {
 	for i := 0; i < 7; i++ {
 		require.NoError(t, be.send(context.Background(), newErrorRequest()))
 	}
-	require.NoError(t, tt.CheckExporterMetricGauge("otelcol_exporter_queue_size", int64(7)))
+	require.NoError(t, tt.CheckExporterQueueSizeMetric(int64(7)))
 
 	assert.NoError(t, be.Shutdown(context.Background()))
 }
@@ -426,9 +425,12 @@ func TestQueuedRetryPersistentEnabled_NoDataLossOnShutdown(t *testing.T) {
 func TestQueueSenderNoStartShutdown(t *testing.T) {
 	queue := queue.NewBoundedMemoryQueue[Request](queue.MemoryQueueSettings[Request]{})
 	set := exportertest.NewNopSettings()
-	builder, err := metadata.NewTelemetryBuilder(set.TelemetrySettings)
+	obsrep, err := NewObsReport(ObsReportSettings{
+		ExporterID:             exporterID,
+		ExporterCreateSettings: exportertest.NewNopSettings(),
+	})
 	assert.NoError(t, err)
-	qs := newQueueSender(queue, set, 1, "", builder)
+	qs := newQueueSender(queue, set, 1, "", obsrep)
 	assert.NoError(t, qs.Shutdown(context.Background()))
 }
 

internal/obsreportconfig/obsmetrics/obs_exporter.go

@@ -7,6 +7,9 @@ const (
 	// ExporterKey used to identify exporters in metrics and traces.
 	ExporterKey = "exporter"
 
+	// DataTypeKey used to identify the data type in the queue size metric.
+	DataTypeKey = "data_type"
+
 	// SentSpansKey used to track spans sent by exporters.
 	SentSpansKey = "sent_spans"
 	// FailedToSendSpansKey used to track spans that failed to be sent by exporters.