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,10 @@ 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
+
 func splitMessagesByRequestSize(messages []*messageWithCallback, maxRequestSize int) [][]*messageWithCallback {
 	if len(messages) == 0 {
 		return nil
@@ -306,12 +315,17 @@ func splitMessagesByRequestSize(messages []*messageWithCallback, maxRequestSize
 		return [][]*messageWithCallback{messages}
 	}
 
+	effectiveMax := maxRequestSize - grpcFramingOverhead
+	if effectiveMax <= 0 {
+		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
@@ -439,6 +453,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 max gRPC request size"),
+		otelmetric.WithUnit("{split}"),
+	)
+	if err != nil {
+		return batchClientMetrics{}, err
+	}
 
 	return batchClientMetrics{
 		sendRequestsTotal: sendRequestsTotal,
@@ -447,6 +469,7 @@ func newBatchClientMetrics() (batchClientMetrics, error) {
 		requestSizeBytes:    requestSizeBytes,
 		requestLatencyMS:    requestLatencyMS,
 		configInfo:          configInfo,
+		batchSplitsTotal:    batchSplitsTotal,
 		successStatusAttr: otelmetric.WithAttributeSet(attribute.NewSet(
 			attribute.String("status", "success"),
 		)),