batch: reserve 10KiB gRPC framing overhead in split and add batch_splits_total metric

Author: pkcll

pkg/chipingress/batch/client.go

@@ -57,6 +57,7 @@ type batchClientMetrics struct {
 	requestSizeBytes    otelmetric.Int64Histogram
 	requestLatencyMS    otelmetric.Float64Histogram
 	configInfo          otelmetric.Int64Gauge
+	batchSplitsTotal    otelmetric.Int64Counter
 	batchSizeAttr       otelmetric.MeasurementOption
 	maxGRPCReqSizeAttr  otelmetric.MeasurementOption
 	successStatusAttr   otelmetric.MeasurementOption
@@ -260,7 +261,11 @@ func (b *Client) sendBatch(ctx context.Context, messages []*messageWithCallback)
 	go func() {
 		defer func() { <-b.maxConcurrentSends }()
 
-		for _, batchMessages := range splitMessagesByRequestSize(messages, b.maxGRPCRequestSize) {
+		splitBatches := splitMessagesByRequestSize(messages, b.maxGRPCRequestSize)
+		if len(splitBatches) > 1 {
+			b.metrics.batchSplitsTotal.Add(ctx, 1)
+		}
+		for _, batchMessages := range splitBatches {
 			batchReq, batchBytes := newBatchRequest(batchMessages)
 			if b.maxGRPCRequestSize > 0 && batchBytes > b.maxGRPCRequestSize {
 				err := fmt.Errorf("publish batch serialized size %d exceeds max gRPC request size %d", batchBytes, b.maxGRPCRequestSize)
@@ -298,6 +303,15 @@ func (b *Client) completeBatchCallbacks(messages []*messageWithCallback, err err
 	})
 }
 
+// grpcFramingOverhead accounts for gRPC framing, HTTP/2 headers, auth tokens,
+// tracing metadata, and other per-request overhead not captured by proto.Size.
+const grpcFramingOverhead = 10 * 1024 // 10 KiB
+
+// minMaxGRPCRequestSize is the minimum allowed value for maxGRPCRequestSize.
+// Values below this threshold are clamped to ensure the framing overhead
+// reservation remains meaningful.
+const minMaxGRPCRequestSize = 1024 * 1024 // 1 MiB
+
 func splitMessagesByRequestSize(messages []*messageWithCallback, maxRequestSize int) [][]*messageWithCallback {
 	if len(messages) == 0 {
 		return nil
@@ -306,12 +320,18 @@ func splitMessagesByRequestSize(messages []*messageWithCallback, maxRequestSize
 		return [][]*messageWithCallback{messages}
 	}
 
+	effectiveMax := maxRequestSize - grpcFramingOverhead
+	if effectiveMax <= 0 {
+		// Should not happen given minMaxGRPCRequestSize, but guard defensively.
+		effectiveMax = maxRequestSize
+	}
+
 	var batches [][]*messageWithCallback
 	current := make([]*messageWithCallback, 0, len(messages))
 	for _, msg := range messages {
 		candidate := append(current, msg)
 		_, candidateBytes := newBatchRequest(candidate)
-		if len(current) > 0 && candidateBytes > maxRequestSize {
+		if len(current) > 0 && candidateBytes > effectiveMax {
 			batches = append(batches, current)
 			current = []*messageWithCallback{msg}
 			continue
@@ -340,9 +360,14 @@ func WithBatchSize(batchSize int) Opt {
 	}
 }
 
-// WithMaxGRPCRequestSize sets the max gRPC request size in bytes used for metric comparison attributes.
+// WithMaxGRPCRequestSize sets the max gRPC request size in bytes used for splitting batches.
+// Values below minMaxGRPCRequestSize (1 MiB) are clamped up to ensure the framing overhead
+// reservation remains meaningful.
 func WithMaxGRPCRequestSize(maxReqSize int) Opt {
 	return func(c *Client) {
+		if maxReqSize > 0 && maxReqSize < minMaxGRPCRequestSize {
+			maxReqSize = minMaxGRPCRequestSize
+		}
 		c.maxGRPCRequestSize = maxReqSize
 	}
 }
@@ -439,6 +464,14 @@ func newBatchClientMetrics() (batchClientMetrics, error) {
 	if err != nil {
 		return batchClientMetrics{}, err
 	}
+	batchSplitsTotal, err := meter.Int64Counter(
+		"chip_ingress.batch.batch_splits_total",
+		otelmetric.WithDescription("Total number of times a batch was split due to exceeding the effective gRPC request size limit (max request size minus reserved framing overhead)"),
+		otelmetric.WithUnit("{split}"),
+	)
+	if err != nil {
+		return batchClientMetrics{}, err
+	}
 
 	return batchClientMetrics{
 		sendRequestsTotal: sendRequestsTotal,
@@ -447,6 +480,7 @@ func newBatchClientMetrics() (batchClientMetrics, error) {
 		requestSizeBytes:    requestSizeBytes,
 		requestLatencyMS:    requestLatencyMS,
 		configInfo:          configInfo,
+		batchSplitsTotal:    batchSplitsTotal,
 		successStatusAttr: otelmetric.WithAttributeSet(attribute.NewSet(
 			attribute.String("status", "success"),
 		)),

pkg/chipingress/batch/client_test.go

@@ -79,9 +79,10 @@ func TestNewBatchClient(t *testing.T) {
 			WithBatchSize(1),
 			WithBatchInterval(time.Second),
 			WithMessageBuffer(10),
-			WithMaxGRPCRequestSize(maxGRPCSize),
 		)
 		require.NoError(t, err)
+		client.maxGRPCRequestSize = maxGRPCSize
+		require.NoError(t, err)
 		client.Start(t.Context())
 
 		err = client.QueueMessage(&chipingress.CloudEventPb{
@@ -303,8 +304,9 @@ func TestSendBatch(t *testing.T) {
 			}).
 			Times(3)
 
-		client, err := NewBatchClient(mockClient, WithMaxGRPCRequestSize(maxRequestSize))
+		client, err := NewBatchClient(mockClient)
 		require.NoError(t, err)
+		client.maxGRPCRequestSize = maxRequestSize
 
 		messages := make([]*messageWithCallback, 0, len(events))
 		for _, event := range events {
@@ -339,14 +341,69 @@ func TestSendBatch(t *testing.T) {
 		mockClient.AssertExpectations(t)
 	})
 
+	t.Run("records batch_splits_total metric when batch is split", func(t *testing.T) {
+		reader, restore := useTestMeterProvider(t)
+		defer restore()
+
+		events := []*chipingress.CloudEventPb{
+			largeTestEvent("split-metric-1"),
+			largeTestEvent("split-metric-2"),
+			largeTestEvent("split-metric-3"),
+		}
+		// Set maxRequestSize so that 2 events fit but 3 do not, forcing a split.
+		maxRequestSize := proto.Size(&chipingress.CloudEventBatch{Events: events[:2]})
+
+		mockClient := mocks.NewClient(t)
+		done := make(chan struct{})
+		var mu sync.Mutex
+		var publishCount int
+
+		mockClient.
+			On("PublishBatch", mock.Anything, mock.Anything).
+			Return(&chipingress.PublishResponse{}, nil).
+			Run(func(args mock.Arguments) {
+				mu.Lock()
+				publishCount++
+				if publishCount == 2 {
+					close(done)
+				}
+				mu.Unlock()
+			})
+
+		client, err := NewBatchClient(mockClient)
+		require.NoError(t, err)
+		client.maxGRPCRequestSize = maxRequestSize
+
+		messages := make([]*messageWithCallback, 0, len(events))
+		for _, event := range events {
+			messages = append(messages, &messageWithCallback{event: event})
+		}
+
+		client.sendBatch(t.Context(), messages)
+
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Fatal("timeout waiting for split batches to be sent")
+		}
+
+		rm := collectResourceMetrics(t, reader)
+		splitsMetric := mustMetric(t, rm, "chip_ingress.batch.batch_splits_total")
+		splitsSum, ok := splitsMetric.Data.(metricdata.Sum[int64])
+		require.True(t, ok)
+		require.Len(t, splitsSum.DataPoints, 1)
+		assert.Equal(t, int64(1), splitsSum.DataPoints[0].Value)
+	})
+
 	t.Run("doesn't publish a single event over max gRPC request size", func(t *testing.T) {
 		mockClient := mocks.NewClient(t)
 		callbackDone := make(chan error, 1)
 		event := largeTestEvent("oversized-id")
 		maxRequestSize := proto.Size(&chipingress.CloudEventBatch{Events: []*chipingress.CloudEventPb{event}}) - 1
 
-		client, err := NewBatchClient(mockClient, WithMaxGRPCRequestSize(maxRequestSize))
+		client, err := NewBatchClient(mockClient)
 		require.NoError(t, err)
+		client.maxGRPCRequestSize = maxRequestSize
 
 		client.sendBatch(t.Context(), []*messageWithCallback{
 			{
@@ -1350,7 +1407,7 @@ func TestBatchClient_Metrics(t *testing.T) {
 			WithBatchSize(1),
 			WithBatchInterval(time.Second),
 			WithMessageBuffer(10),
-			WithMaxGRPCRequestSize(2048),
+			WithMaxGRPCRequestSize(minMaxGRPCRequestSize),
 		)
 		require.NoError(t, err)
 		client.Start(t.Context())
@@ -1386,7 +1443,7 @@ func TestBatchClient_Metrics(t *testing.T) {
 		reqSize := mustMetric(t, rm, "chip_ingress.batch.request_size_bytes")
 		reqSizeHist, ok := reqSize.Data.(metricdata.Histogram[int64])
 		require.True(t, ok)
-		reqSizePoint := mustInt64HistogramPointWithIntAttr(t, reqSizeHist, "max_grpc_request_size_bytes", 2048)
+		reqSizePoint := mustInt64HistogramPointWithIntAttr(t, reqSizeHist, "max_grpc_request_size_bytes", minMaxGRPCRequestSize)
 		assert.GreaterOrEqual(t, reqSizePoint.Count, uint64(1))
 
 		latency := mustMetric(t, rm, "chip_ingress.batch.request_latency_ms")
@@ -1401,7 +1458,7 @@ func TestBatchClient_Metrics(t *testing.T) {
 		require.NotEmpty(t, configGauge.DataPoints)
 		assert.Equal(t, int64(1), configGauge.DataPoints[0].Value)
 		assert.True(t, hasIntAttr(configGauge.DataPoints[0].Attributes, "max_batch_size", 1))
-		assert.True(t, hasIntAttr(configGauge.DataPoints[0].Attributes, "max_grpc_request_size_bytes", 2048))
+		assert.True(t, hasIntAttr(configGauge.DataPoints[0].Attributes, "max_grpc_request_size_bytes", minMaxGRPCRequestSize))
 	})
 
 	t.Run("records failure counters and latency", func(t *testing.T) {
@@ -1540,11 +1597,11 @@ func BenchmarkSendBatch(b *testing.B) {
 			WithBatchSize(100),
 			WithMessageBuffer(b.N*100+10),
 			WithBatchInterval(time.Hour),
-			WithMaxGRPCRequestSize(512),
 		)
 		if err != nil {
 			b.Fatal(err)
 		}
+		client.maxGRPCRequestSize = 512
 		client.Start(b.Context())
 		defer client.Stop()