Expose metrics for batch sizes

Signed-off-by: Israel Blancas <[email protected]>

Author: iblancasa

.chloggen/12894.yaml

@@ -0,0 +1,28 @@
+# 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: Expose otelcol_exporter_batch_send_size_bytes from the default batcher.
+
+# One or more tracking issues or pull requests related to the change
+issues: [12894]
+
+# (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: |
+  Added the following metric to the exporter helper default batcher:
+  - otelcol_exporter_batch_send_size_bytes: Batches size in the queue (in bytes)
+
+
+# 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]

exporter/exporterhelper/documentation.md

@@ -6,6 +6,14 @@
 
 The following telemetry is emitted by this component.
 
+### otelcol_exporter_batch_send_size_bytes
+
+Batches size in the queue (in bytes). [alpha]
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {By} | Histogram | Int |
+
 ### otelcol_exporter_enqueue_failed_log_records
 
 Number of log records failed to be added to the sending queue. [alpha]
@@ -32,15 +40,15 @@ Number of spans failed to be added to the sending queue. [alpha]
 
 ### otelcol_exporter_queue_capacity
 
-Fixed capacity of the retry queue (in batches) [alpha]
+Fixed capacity of the retry queue (in batches). [alpha]
 
 | Unit | Metric Type | Value Type |
 | ---- | ----------- | ---------- |
 | {batches} | Gauge | Int |
 
 ### otelcol_exporter_queue_size
 
-Current size of the retry queue (in batches) [alpha]
+Current size of the retry queue (in batches). [alpha]
 
 | Unit | Metric Type | Value Type |
 | ---- | ----------- | ---------- |

exporter/exporterhelper/internal/base_exporter.go

@@ -14,6 +14,7 @@ import (
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadata"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
@@ -49,20 +50,27 @@ type BaseExporter struct {
 	queueBatchSettings QueueBatchSettings[request.Request]
 	queueCfg           queuebatch.Config
 	batcherCfg         BatcherConfig
+	tb                 *metadata.TelemetryBuilder
 }
 
 func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, pusher sender.SendFunc[request.Request], options ...Option) (*BaseExporter, error) {
+	var err error
 	be := &BaseExporter{
 		Set:        set,
 		timeoutCfg: NewDefaultTimeoutConfig(),
 	}
 
 	for _, op := range options {
-		if err := op(be); err != nil {
+		if err = op(be); err != nil {
 			return nil, err
 		}
 	}
 
+	be.tb, err = metadata.NewTelemetryBuilder(set.TelemetrySettings)
+	if err != nil {
+		return nil, err
+	}
+
 	// Consumer Sender is always initialized.
 	be.firstSender = sender.NewSender(pusher)
 
@@ -77,11 +85,7 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, pusher sende
 		be.firstSender = be.RetrySender
 	}
 
-	var err error
-	be.firstSender, err = newObsReportSender(set, signal, be.firstSender)
-	if err != nil {
-		return nil, err
-	}
+	be.firstSender = newObsReportSender(set, signal, be.firstSender, be.tb)
 
 	if be.batcherCfg.Enabled || be.queueCfg.Batch != nil {
 		// Batcher mutates the data.
@@ -90,12 +94,13 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, pusher sende
 
 	if be.queueCfg.Enabled || be.batcherCfg.Enabled {
 		qSet := queuebatch.Settings[request.Request]{
-			Signal:      signal,
-			ID:          set.ID,
-			Telemetry:   set.TelemetrySettings,
-			Encoding:    be.queueBatchSettings.Encoding,
-			Sizers:      be.queueBatchSettings.Sizers,
-			Partitioner: be.queueBatchSettings.Partitioner,
+			Signal:           signal,
+			ID:               set.ID,
+			Telemetry:        set.TelemetrySettings,
+			Encoding:         be.queueBatchSettings.Encoding,
+			Sizers:           be.queueBatchSettings.Sizers,
+			Partitioner:      be.queueBatchSettings.Partitioner,
+			TelemetryBuilder: be.tb,
 		}
 		be.QueueSender, err = NewQueueSender(qSet, be.queueCfg, be.batcherCfg, be.ExportFailureMessage, be.firstSender)
 		if err != nil {

exporter/exporterhelper/internal/base_exporter_test.go

@@ -10,6 +10,9 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/metric"
+	embeddedmetric "go.opentelemetry.io/otel/metric/embedded"
+	noopmetric "go.opentelemetry.io/otel/metric/noop"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest/observer"
 
@@ -143,6 +146,29 @@ func TestQueueRetryWithDisabledQueue(t *testing.T) {
 	}
 }
 
+type failingMeterProvider struct {
+	embeddedmetric.MeterProvider
+}
+
+func (m failingMeterProvider) Meter(_ string, _ ...metric.MeterOption) metric.Meter {
+	return failingMeter{}
+}
+
+type failingMeter struct {
+	noopmetric.Meter
+}
+
+func (m failingMeter) Int64Counter(_ string, _ ...metric.Int64CounterOption) (metric.Int64Counter, error) {
+	return nil, errors.New("failed to create counter")
+}
+
+func TestBaseExporterWithTelemetryError(t *testing.T) {
+	set := exportertest.NewNopSettings(exportertest.NopType)
+	set.MeterProvider = failingMeterProvider{}
+	_, err := NewBaseExporter(set, pipeline.SignalMetrics, noopExport)
+	require.Error(t, err)
+}
+
 func errExport(context.Context, request.Request) error {
 	return errors.New("my error")
 }

exporter/exporterhelper/internal/metadata/generated_telemetry.go

@@ -49,12 +49,7 @@ type obsReportSender[K request.Request] struct {
 	next            sender.Sender[K]
 }
 
-func newObsReportSender[K request.Request](set exporter.Settings, signal pipeline.Signal, next sender.Sender[K]) (sender.Sender[K], error) {
-	telemetryBuilder, err := metadata.NewTelemetryBuilder(set.TelemetrySettings)
-	if err != nil {
-		return nil, err
-	}
-
+func newObsReportSender[K request.Request](set exporter.Settings, signal pipeline.Signal, next sender.Sender[K], tb *metadata.TelemetryBuilder) sender.Sender[K] {
 	idStr := set.ID.String()
 	expAttr := attribute.String(ExporterKey, idStr)
 
@@ -68,26 +63,27 @@ func newObsReportSender[K request.Request](set exporter.Settings, signal pipelin
 
 	switch signal {
 	case pipeline.SignalTraces:
-		or.itemsSentInst = telemetryBuilder.ExporterSentSpans
-		or.itemsFailedInst = telemetryBuilder.ExporterSendFailedSpans
+		or.itemsSentInst = tb.ExporterSentSpans
+		or.itemsFailedInst = tb.ExporterSendFailedSpans
 
 	case pipeline.SignalMetrics:
-		or.itemsSentInst = telemetryBuilder.ExporterSentMetricPoints
-		or.itemsFailedInst = telemetryBuilder.ExporterSendFailedMetricPoints
+		or.itemsSentInst = tb.ExporterSentMetricPoints
+		or.itemsFailedInst = tb.ExporterSendFailedMetricPoints
 
 	case pipeline.SignalLogs:
-		or.itemsSentInst = telemetryBuilder.ExporterSentLogRecords
-		or.itemsFailedInst = telemetryBuilder.ExporterSendFailedLogRecords
+		or.itemsSentInst = tb.ExporterSentLogRecords
+		or.itemsFailedInst = tb.ExporterSendFailedLogRecords
 	}
 
-	return or, nil
+	return or
 }
 
 func (ors *obsReportSender[K]) Send(ctx context.Context, req K) error {
 	// Have to read the number of items before sending the request since the request can
 	// be modified by the downstream components like the batcher.
 	c := ors.startOp(ctx)
 	items := req.ItemsCount()
+
 	// Forward the data to the next consumer (this pusher is the next).
 	err := ors.next.Send(c, req)
 	ors.endOp(c, items, err)

exporter/exporterhelper/internal/metadatatest/generated_telemetrytest.go

@@ -18,6 +18,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadata"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadatatest"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
@@ -38,13 +39,17 @@ func TestExportTraceDataOp(t *testing.T) {
 	parentCtx, parentSpan := tt.NewTelemetrySettings().TracerProvider.Tracer("test").Start(context.Background(), t.Name())
 	defer parentSpan.End()
 
+	tb, err := metadata.NewTelemetryBuilder(tt.NewTelemetrySettings())
+	require.NoError(t, err)
+	t.Cleanup(func() { tb.Shutdown() })
+
 	var exporterErr error
-	obsrep, err := newObsReportSender(
+	obsrep := newObsReportSender(
 		exporter.Settings{ID: exporterID, TelemetrySettings: tt.NewTelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()},
 		pipeline.SignalTraces,
 		sender.NewSender(func(context.Context, request.Request) error { return exporterErr }),
+		tb,
 	)
-	require.NoError(t, err)
 
 	params := []testParams{
 		{items: 22, err: nil},
@@ -106,13 +111,17 @@ func TestExportMetricsOp(t *testing.T) {
 	parentCtx, parentSpan := tt.NewTelemetrySettings().TracerProvider.Tracer("test").Start(context.Background(), t.Name())
 	defer parentSpan.End()
 
+	tb, err := metadata.NewTelemetryBuilder(tt.NewTelemetrySettings())
+	require.NoError(t, err)
+	t.Cleanup(func() { tb.Shutdown() })
+
 	var exporterErr error
-	obsrep, err := newObsReportSender(
+	obsrep := newObsReportSender(
 		exporter.Settings{ID: exporterID, TelemetrySettings: tt.NewTelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()},
 		pipeline.SignalMetrics,
 		sender.NewSender(func(context.Context, request.Request) error { return exporterErr }),
+		tb,
 	)
-	require.NoError(t, err)
 
 	params := []testParams{
 		{items: 17, err: nil},
@@ -174,13 +183,17 @@ func TestExportLogsOp(t *testing.T) {
 	parentCtx, parentSpan := tt.NewTelemetrySettings().TracerProvider.Tracer("test").Start(context.Background(), t.Name())
 	defer parentSpan.End()
 
+	tb, err := metadata.NewTelemetryBuilder(tt.NewTelemetrySettings())
+	require.NoError(t, err)
+	t.Cleanup(func() { tb.Shutdown() })
+
 	var exporterErr error
-	obsrep, err := newObsReportSender(
+	obsrep := newObsReportSender(
 		exporter.Settings{ID: exporterID, TelemetrySettings: tt.NewTelemetrySettings(), BuildInfo: component.NewDefaultBuildInfo()},
 		pipeline.SignalLogs,
 		sender.NewSender(func(context.Context, request.Request) error { return exporterErr }),
+		tb,
 	)
-	require.NoError(t, err)
 
 	params := []testParams{
 		{items: 17, err: nil},