Add batch size metric (#1241)

* Add batch size metric to otel collector

Signed-off-by: Pavol Loffay <[email protected]>

* Rename variable

Signed-off-by: Pavol Loffay <[email protected]>

Author: pavolloffay

processor/batchprocessor/batch_processor.go

@@ -171,7 +171,8 @@ func (bp *batchTraceProcessor) resetTimer() {
 func (bp *batchTraceProcessor) sendItems(measure *stats.Int64Measure) {
 	// Add that it came form the trace pipeline?
 	statsTags := []tag.Mutator{tag.Insert(processor.TagProcessorNameKey, bp.name)}
-	stats.RecordWithTags(context.Background(), statsTags, measure.M(1))
+	_ = stats.RecordWithTags(context.Background(), statsTags, measure.M(1))
+	_ = stats.RecordWithTags(context.Background(), statsTags, statBatchSendSize.M(int64(bp.batchTraces.getSpanCount())))
 
 	if err := bp.traceConsumer.ConsumeTraces(context.Background(), bp.batchTraces.getTraceData()); err != nil {
 		bp.logger.Warn("Sender failed", zap.Error(err))
@@ -232,6 +233,7 @@ func (bp *batchMetricProcessor) sendItems(measure *stats.Int64Measure) {
 	// Add that it came from the metrics pipeline
 	statsTags := []tag.Mutator{tag.Insert(processor.TagProcessorNameKey, bp.name)}
 	_ = stats.RecordWithTags(context.Background(), statsTags, measure.M(1))
+	_ = stats.RecordWithTags(context.Background(), statsTags, statBatchSendSize.M(int64(bp.batchMetrics.metricData.MetricCount())))
 
 	_ = bp.metricsConsumer.ConsumeMetrics(context.Background(), bp.batchMetrics.getData())
 	bp.batchMetrics.reset()

processor/batchprocessor/batch_processor_test.go

@@ -21,7 +21,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/bmizerany/assert"
 	"github.com/stretchr/testify/require"
+	"go.opencensus.io/stats/view"
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
@@ -73,6 +75,10 @@ func TestBatchProcessorSpansDelivered(t *testing.T) {
 }
 
 func TestBatchProcessorSentBySize(t *testing.T) {
+	views := MetricViews()
+	view.Register(views...)
+	defer view.Unregister(views...)
+
 	sender := newTestSender()
 	cfg := generateDefaultConfig()
 	sendBatchSize := 20
@@ -114,6 +120,15 @@ func TestBatchProcessorSentBySize(t *testing.T) {
 		}
 	}
 
+	data, err := view.RetrieveData(statBatchSendSize.Name())
+	require.NoError(t, err)
+	assert.Equal(t, 1, len(data))
+	distData := data[0].Data.(*view.DistributionData)
+	assert.Equal(t, int64(expectedBatchesNum), distData.Count)
+	assert.Equal(t, sender.spansReceived, int(distData.Sum()))
+	assert.Equal(t, sendBatchSize, int(distData.Min))
+	assert.Equal(t, sendBatchSize, int(distData.Max))
+
 	sender.mtx.RUnlock()
 }
 
@@ -384,6 +399,10 @@ func TestBatchMetricProcessor_ReceivingData(t *testing.T) {
 }
 
 func TestBatchMetricProcessor_BatchSize(t *testing.T) {
+	views := MetricViews()
+	view.Register(views...)
+	defer view.Unregister(views...)
+
 	// Instantiate the batch processor with low config values to test data
 	// gets sent through the processor.
 	cfg := Config{
@@ -427,6 +446,15 @@ func TestBatchMetricProcessor_BatchSize(t *testing.T) {
 			require.Equal(t, metricsPerRequest, md.ResourceMetrics().At(i).InstrumentationLibraryMetrics().At(0).Metrics().Len())
 		}
 	}
+
+	data, err := view.RetrieveData(statBatchSendSize.Name())
+	require.NoError(t, err)
+	assert.Equal(t, 1, len(data))
+	distData := data[0].Data.(*view.DistributionData)
+	assert.Equal(t, int64(expectedBatchesNum), distData.Count)
+	assert.Equal(t, tms.metricsReceived, int(distData.Sum()))
+	assert.Equal(t, cfg.SendBatchSize, uint32(distData.Min))
+	assert.Equal(t, cfg.SendBatchSize, uint32(distData.Max))
 	tms.mtx.RUnlock()
 }
 

processor/batchprocessor/metrics.go

@@ -26,6 +26,7 @@ import (
 var (
 	statBatchSizeTriggerSend = stats.Int64("batch_size_trigger_send", "Number of times the batch was sent due to a size trigger", stats.UnitDimensionless)
 	statTimeoutTriggerSend   = stats.Int64("timeout_trigger_send", "Number of times the batch was sent due to a timeout trigger", stats.UnitDimensionless)
+	statBatchSendSize        = stats.Int64("batch_send_size", "Number of units in the batch", stats.UnitDimensionless)
 )
 
 // MetricViews returns the metrics views related to batching
@@ -48,9 +49,18 @@ func MetricViews() []*view.View {
 		Aggregation: view.Sum(),
 	}
 
+	distributionBatchSendSizeView := &view.View{
+		Name:        statBatchSendSize.Name(),
+		Measure:     statBatchSendSize,
+		Description: statBatchSendSize.Description(),
+		TagKeys:     processorTagKeys,
+		Aggregation: view.Distribution(10, 25, 50, 75, 100, 250, 500, 750, 1000, 2000, 3000, 4000, 5000, 6000, 7000, 8000, 9000, 10000, 20000, 30000, 50000, 100000),
+	}
+
 	legacyViews := []*view.View{
 		countBatchSizeTriggerSendView,
 		countTimeoutTriggerSendView,
+		distributionBatchSendSizeView,
 	}
 
 	return obsreport.ProcessorMetricViews(typeStr, legacyViews)