feat: dimensional metrics for elasticsearch.events.processed metric

appender.go

@@ -141,7 +141,7 @@ func New(client *elasticsearch.Client, cfg Config) (*Appender, error) {
 		available:      available,
 		closed:         make(chan struct{}),
 		bulkItems:      make(chan bulkIndexerItem, cfg.DocumentBufferSize),
-		metrics:        ms,
+		metrics:        *ms,
 		telemetryAttrs: cfg.MetricAttributes,
 	}
 	indexer.addCount(int64(len(available)), &indexer.availableBulkRequests, ms.availableBulkRequests)
@@ -240,12 +240,13 @@ func (a *Appender) Add(ctx context.Context, index string, document io.Reader) er
 	return nil
 }
 
-func (a *Appender) addCount(delta int64, lm *int64, m metric.Int64Counter) {
+func (a *Appender) addCount(delta int64, lm *int64, m metric.Int64Counter, opts ...metric.AddOption) {
 	// legacy metric
 	atomic.AddInt64(lm, delta)
 
 	attrs := metric.WithAttributeSet(a.config.MetricAttributes)
-	m.Add(context.Background(), delta, attrs)
+	opts = append(opts, attrs)
+	m.Add(context.Background(), delta, opts...)
 }
 
 func (a *Appender) flush(ctx context.Context, bulkIndexer *bulkIndexer) error {
@@ -275,7 +276,7 @@ func (a *Appender) flush(ctx context.Context, bulkIndexer *bulkIndexer) error {
 		a.addCount(int64(flushed), &a.bytesTotal, a.metrics.bytesTotal)
 	}
 	if err != nil {
-		a.addCount(int64(n), &a.docsFailed, a.metrics.docsFailed)
+		atomic.AddInt64(&a.docsFailed, int64(n))
 		logger.Error("bulk indexing request failed", zap.Error(err))
 		if a.tracingEnabled() {
 			apm.CaptureError(ctx, err).Send()
@@ -284,7 +285,11 @@ func (a *Appender) flush(ctx context.Context, bulkIndexer *bulkIndexer) error {
 		var errTooMany errorTooManyRequests
 		// 429 may be returned as errors from the bulk indexer.
 		if errors.As(err, &errTooMany) {
-			a.addCount(int64(n), &a.tooManyRequests, a.metrics.tooManyRequests)
+			a.addCount(int64(n),
+				&a.tooManyRequests,
+				a.metrics.docsIndexed,
+				metric.WithAttributes(attribute.String("status", "TooMany")),
+			)
 		}
 		return err
 	}
@@ -319,19 +324,35 @@ func (a *Appender) flush(ctx context.Context, bulkIndexer *bulkIndexer) error {
 		}
 	}
 	if docsFailed > 0 {
-		a.addCount(docsFailed, &a.docsFailed, a.metrics.docsFailed)
+		atomic.AddInt64(&a.docsFailed, docsFailed)
 	}
 	if docsIndexed > 0 {
-		a.addCount(docsIndexed, &a.docsIndexed, a.metrics.docsIndexed)
+		a.addCount(docsIndexed,
+			&a.docsIndexed,
+			a.metrics.docsIndexed,
+			metric.WithAttributes(attribute.String("status", "Success")),
+		)
 	}
 	if tooManyRequests > 0 {
-		a.addCount(tooManyRequests, &a.tooManyRequests, a.metrics.tooManyRequests)
+		a.addCount(tooManyRequests,
+			&a.tooManyRequests,
+			a.metrics.docsIndexed,
+			metric.WithAttributes(attribute.String("status", "TooMany")),
+		)
 	}
 	if clientFailed > 0 {
-		a.addCount(clientFailed, &a.docsFailedClient, a.metrics.docsFailedClient)
+		a.addCount(clientFailed,
+			&a.docsFailedClient,
+			a.metrics.docsIndexed,
+			metric.WithAttributes(attribute.String("status", "FailedClient")),
+		)
 	}
 	if serverFailed > 0 {
-		a.addCount(serverFailed, &a.docsFailedServer, a.metrics.docsFailedServer)
+		a.addCount(serverFailed,
+			&a.docsFailedServer,
+			a.metrics.docsIndexed,
+			metric.WithAttributes(attribute.String("status", "FailedServer")),
+		)
 	}
 	logger.Debug(
 		"bulk request completed",

appender_test.go

@@ -36,6 +36,7 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zaptest/observer"
@@ -145,6 +146,33 @@ loop:
 			assert.Equal(t, attrs, dp.Attributes)
 		}
 	}
+
+	var processedAsserted int
+	assertProcessedCounter := func(metric metricdata.Metrics, count int64, attrs attribute.Set) {
+		asserted++
+		counter := metric.Data.(metricdata.Sum[int64])
+		for _, dp := range counter.DataPoints {
+			metricdatatest.AssertHasAttributes[metricdata.DataPoint[int64]](t, dp, attrs.ToSlice()...)
+			status, exist := dp.Attributes.Value(attribute.Key("status"))
+			assert.True(t, exist)
+			switch status.AsString() {
+			case "Success":
+				processedAsserted++
+				assert.Equal(t, stats.Indexed, dp.Value)
+			case "FailedClient":
+				processedAsserted++
+				assert.Equal(t, stats.FailedClient, dp.Value)
+			case "FailedServer":
+				processedAsserted++
+				assert.Equal(t, stats.FailedServer, dp.Value)
+			case "TooMany":
+				processedAsserted++
+				assert.Equal(t, stats.TooManyRequests, dp.Value)
+			default:
+				assert.FailNow(t, "Unexpected metric with status: "+status.AsString())
+			}
+		}
+	}
 	// check the set of names and then check the counter or histogram
 	unexpectedMetrics := []string{}
 	for _, metric := range rm.ScopeMetrics[0].Metrics {
@@ -155,16 +183,8 @@ loop:
 			assertCounter(metric, stats.Active, indexerAttrs)
 		case "elasticsearch.bulk_requests.count":
 			assertCounter(metric, stats.BulkRequests, indexerAttrs)
-		case "elasticsearch.failed.count":
-			assertCounter(metric, stats.Failed, indexerAttrs)
-		case "elasticsearch.failed.client.count":
-			assertCounter(metric, stats.FailedClient, indexerAttrs)
-		case "elasticsearch.failed.server.count":
-			assertCounter(metric, stats.FailedServer, indexerAttrs)
 		case "elasticsearch.events.processed":
-			assertCounter(metric, stats.Indexed, indexerAttrs)
-		case "elasticsearch.failed.too_many_reqs":
-			assertCounter(metric, stats.TooManyRequests, indexerAttrs)
+			assertProcessedCounter(metric, stats.Indexed, indexerAttrs)
 		case "elasticsearch.bulk_requests.available":
 			assertCounter(metric, stats.AvailableBulkRequests, indexerAttrs)
 		case "elasticsearch.flushed.bytes":
@@ -180,7 +200,8 @@ loop:
 		}
 	}
 	assert.Empty(t, unexpectedMetrics)
-	assert.Equal(t, 10, asserted)
+	assert.Equal(t, 6, asserted)
+	assert.Equal(t, 4, processedAsserted)
 }
 
 func TestAppenderAvailableAppenders(t *testing.T) {
@@ -390,7 +411,7 @@ func TestAppenderFlushMetric(t *testing.T) {
 			if !ignoreCount {
 				assert.Equal(t, count, int(dp.Count))
 			} else {
-				assert.Greater(t, int(dp.Count), count)
+				assert.GreaterOrEqual(t, int(dp.Count), count)
 			}
 			assert.Positive(t, dp.Sum)
 			assert.Equal(t, attrs, dp.Attributes)
@@ -607,7 +628,6 @@ func TestAppenderCloseInterruptAdd(t *testing.T) {
 	defer cancel()
 	go func() {
 		added <- indexer.Add(addContext, "logs-foo-testing", readerFunc(func(p []byte) (int, error) {
-			fmt.Println("hello?")
 			close(readInvoked)
 			return copy(p, "{}"), nil
 		}))

metric.go

@@ -31,11 +31,7 @@ type metrics struct {
 	bulkRequests          metric.Int64Counter
 	docsAdded             metric.Int64Counter
 	docsActive            metric.Int64Counter
-	docsFailed            metric.Int64Counter
-	docsFailedClient      metric.Int64Counter
-	docsFailedServer      metric.Int64Counter
 	docsIndexed           metric.Int64Counter
-	tooManyRequests       metric.Int64Counter
 	bytesTotal            metric.Int64Counter
 	availableBulkRequests metric.Int64Counter
 	activeCreated         metric.Int64Counter
@@ -56,7 +52,7 @@ type counterMetric struct {
 	p           *metric.Int64Counter
 }
 
-func newMetrics(cfg Config) (metrics, error) {
+func newMetrics(cfg Config) (*metrics, error) {
 	if cfg.MeterProvider == nil {
 		cfg.MeterProvider = otel.GetMeterProvider()
 	}
@@ -80,7 +76,7 @@ func newMetrics(cfg Config) (metrics, error) {
 	for _, m := range histograms {
 		err := newFloat64Histogram(meter, m)
 		if err != nil {
-			return ms, err
+			return &ms, err
 		}
 	}
 
@@ -92,39 +88,19 @@ func newMetrics(cfg Config) (metrics, error) {
 		},
 		{
 			name:        "elasticsearch.events.count",
-			description: "the total number of items added to the indexer.",
+			description: "Number of APM Events received for indexing",
 			p:           &ms.docsAdded,
 		},
 		{
 			name:        "elasticsearch.events.queued",
 			description: "the number of active items waiting in the indexer's queue.",
 			p:           &ms.docsActive,
 		},
-		{
-			name:        "elasticsearch.failed.count",
-			description: "The amount of time a document was buffered for, in seconds.",
-			p:           &ms.docsFailed,
-		},
-		{
-			name:        "elasticsearch.failed.client.count",
-			description: "The number of docs failed to get indexed with client error(status_code >= 400 < 500, but not 429).",
-			p:           &ms.docsFailedClient,
-		},
-		{
-			name:        "elasticsearch.failed.server.count",
-			description: "The number of docs failed to get indexed with server error(status_code >= 500).",
-			p:           &ms.docsFailedServer,
-		},
 		{
 			name:        "elasticsearch.events.processed",
-			description: "The number of docs indexed successfully.",
+			description: "Number of APM Events flushed to Elasticsearch. Dimensions are used to report the project ID, success or failures",
 			p:           &ms.docsIndexed,
 		},
-		{
-			name:        "elasticsearch.failed.too_many_reqs",
-			description: "The number of 429 errors returned from the bulk indexer due to too many requests.",
-			p:           &ms.tooManyRequests,
-		},
 		{
 			name:        "elasticsearch.flushed.bytes",
 			description: "The total number of bytes written to the request body",
@@ -150,11 +126,11 @@ func newMetrics(cfg Config) (metrics, error) {
 	for _, m := range counters {
 		err := newInt64Counter(meter, m)
 		if err != nil {
-			return ms, err
+			return &ms, err
 		}
 	}
 
-	return ms, nil
+	return &ms, nil
 }
 
 func newInt64Counter(meter metric.Meter, c counterMetric) error {