feat: add chip-ingress batch config keys and tune defaults

pkg/beholder/batch_emitter_service.go

@@ -63,6 +63,10 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg
 	if drainTimeout == 0 {
 		drainTimeout = defaults.ChipIngressDrainTimeout
 	}
+	maxGRPCRequestSize := cfg.ChipIngressMaxGRPCRequestSize
+	if maxGRPCRequestSize == 0 {
+		maxGRPCRequestSize = defaults.ChipIngressMaxGRPCRequestSize
+	}
 
 	meter := otel.Meter("beholder/chip_ingress_batch_emitter")
 	metrics, err := newBatchEmitterMetrics(meter)
@@ -77,6 +81,7 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg
 		batch.WithMaxPublishTimeout(sendTimeout),
 		batch.WithShutdownTimeout(drainTimeout),
 		batch.WithMaxConcurrentSends(maxConcurrentSends),
+		batch.WithMaxGRPCRequestSize(maxGRPCRequestSize),
 		batch.WithEventClone(false),
 		batch.WithClientName(batch.ClientNameBeholder),
 	)

pkg/beholder/batch_emitter_service_test.go

@@ -32,6 +32,7 @@ func newTestConfig() beholder.Config {
 		ChipIngressSendInterval:       50 * time.Millisecond,
 		ChipIngressSendTimeout:        5 * time.Second,
 		ChipIngressDrainTimeout:       5 * time.Second,
+		ChipIngressMaxGRPCRequestSize: 1024 * 1024,
 	}
 }
 
@@ -576,6 +577,7 @@ func BenchmarkChipIngressBatchEmitterService_Emit(b *testing.B) {
 		ChipIngressSendInterval:       time.Hour,
 		ChipIngressSendTimeout:        5 * time.Second,
 		ChipIngressDrainTimeout:       5 * time.Second,
+		ChipIngressMaxGRPCRequestSize: 1024 * 1024,
 	}
 	emitter, err := beholder.NewChipIngressBatchEmitterService(&chipingress.NoopClient{}, cfg, logger.Test(b))
 	if err != nil {

pkg/beholder/config.go

@@ -57,12 +57,13 @@ type Config struct {
 
 	// Chip Ingress Batch Emitter
 	ChipIngressBatchEmitterEnabled bool          // When true, use batch emitter; when false (default), use legacy per-event emitter
-	ChipIngressBufferSize          uint          // Message buffer size (default 1000)
-	ChipIngressMaxBatchSize        uint          // Max events per PublishBatch call (default 500)
-	ChipIngressSendInterval        time.Duration // Flush interval (default 100ms)
-	ChipIngressSendTimeout         time.Duration // Timeout per PublishBatch call (default 3s)
-	ChipIngressDrainTimeout        time.Duration // Max time to flush remaining events on shutdown (default 10s)
+	ChipIngressBufferSize          uint          // Message buffer size (default 10000)
+	ChipIngressMaxBatchSize        uint          // Max events per PublishBatch call (default 1000)
+	ChipIngressSendInterval        time.Duration // Flush interval (default 500ms)
+	ChipIngressSendTimeout         time.Duration // Timeout per PublishBatch call (default 10s)
+	ChipIngressDrainTimeout        time.Duration // Max time to flush remaining events on shutdown (default 30s)
 	ChipIngressMaxConcurrentSends  int           // Max concurrent PublishBatch calls (default 10)
+	ChipIngressMaxGRPCRequestSize  int           // Max serialized PublishBatch request size in bytes (default 10 MiB)
 	ChipIngressLogger              logger.Logger // Required when ChipIngressBatchEmitterEnabled is true
 
 	// OTel Log
@@ -162,12 +163,13 @@ func DefaultConfig() Config {
 		LogCompressor:         "gzip",
 		// Chip Ingress Batch Emitter
 		ChipIngressBatchEmitterEnabled: false,
-		ChipIngressBufferSize:          1000,
-		ChipIngressMaxBatchSize:        500,
-		ChipIngressSendInterval:        100 * time.Millisecond,
-		ChipIngressSendTimeout:         3 * time.Second,
-		ChipIngressDrainTimeout:        10 * time.Second,
+		ChipIngressBufferSize:          10000,
+		ChipIngressMaxBatchSize:        1000,
+		ChipIngressSendInterval:        500 * time.Millisecond,
+		ChipIngressSendTimeout:         10 * time.Second,
+		ChipIngressDrainTimeout:        30 * time.Second,
 		ChipIngressMaxConcurrentSends:  defaultMaxConcurrentSends,
+		ChipIngressMaxGRPCRequestSize:  10 * 1024 * 1024, // 10 MiB
 		// Auth (defaults to static auth mode with TTL=0)
 		AuthHeadersTTL: 0,
 	}

pkg/beholder/testdata/config-example.json

@@ -46,6 +46,7 @@
   "ChipIngressSendTimeout": 0,
   "ChipIngressDrainTimeout": 0,
   "ChipIngressMaxConcurrentSends": 0,
+  "ChipIngressMaxGRPCRequestSize": 0,
   "ChipIngressLogger": null,
   "LogExportTimeout": 1000000000,
   "LogExportInterval": 1000000000,

pkg/chipingress/batch/client.go

@@ -562,6 +562,7 @@ func newBatchClientMetrics(clientName string) (batchClientMetrics, error) {
 		"chip_ingress.batch.request_size_messages",
 		otelmetric.WithDescription("PublishBatch request size measured in number of events"),
 		otelmetric.WithUnit("{event}"),
+		otelmetric.WithExplicitBucketBoundaries(1, 2, 3, 5, 8, 13, 21, 34, 55, 100, 250, 500, 1000),
 	)
 	if err != nil {
 		return batchClientMetrics{}, err
@@ -571,8 +572,8 @@ func newBatchClientMetrics(clientName string) (batchClientMetrics, error) {
 		otelmetric.WithDescription("PublishBatch request size measured in bytes"),
 		otelmetric.WithUnit("By"),
 		otelmetric.WithExplicitBucketBoundaries(
-			// Buckets from 1 KiB to 10 MiB (default maxGRPCRequestSize).
-			1*1024, 4*1024, 16*1024, 64*1024, 256*1024,
+			// Buckets from 128 B to 10 MiB (default maxGRPCRequestSize).
+			128, 256, 512, 1*1024, 4*1024, 16*1024, 64*1024, 256*1024,
 			512*1024, 1*1024*1024, 2*1024*1024, 4*1024*1024,
 			8*1024*1024, 10*1024*1024,
 		),

pkg/loop/config.go

@@ -102,6 +102,14 @@ const (
 	envChipIngressBatchEmitterEnabled   = "CL_CHIP_INGRESS_BATCH_EMITTER_ENABLED"
 	envChipIngressDurableEmitterEnabled = "CL_CHIP_INGRESS_DURABLE_EMITTER_ENABLED"
 
+	envChipIngressBufferSize         = "CL_CHIP_INGRESS_BUFFER_SIZE"
+	envChipIngressMaxBatchSize       = "CL_CHIP_INGRESS_MAX_BATCH_SIZE"
+	envChipIngressMaxConcurrentSends = "CL_CHIP_INGRESS_MAX_CONCURRENT_SENDS"
+	envChipIngressSendInterval       = "CL_CHIP_INGRESS_SEND_INTERVAL"
+	envChipIngressSendTimeout        = "CL_CHIP_INGRESS_SEND_TIMEOUT"
+	envChipIngressDrainTimeout       = "CL_CHIP_INGRESS_DRAIN_TIMEOUT"
+	envChipIngressMaxGRPCRequestSize = "CL_CHIP_INGRESS_MAX_GRPC_REQUEST_SIZE"
+
 	envCRESettings        = cresettings.EnvNameSettings
 	envCRESettingsDefault = cresettings.EnvNameSettingsDefault
 )
@@ -116,6 +124,14 @@ type EnvConfig struct {
 	ChipIngressBatchEmitterEnabled   bool
 	ChipIngressDurableEmitterEnabled bool
 
+	ChipIngressBufferSize         uint
+	ChipIngressMaxBatchSize       uint
+	ChipIngressMaxConcurrentSends int
+	ChipIngressSendInterval       time.Duration
+	ChipIngressSendTimeout        time.Duration
+	ChipIngressDrainTimeout       time.Duration
+	ChipIngressMaxGRPCRequestSize int
+
 	CRESettings        string
 	CRESettingsDefault string
 
@@ -322,6 +338,13 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 	add(envChipIngressInsecureConnection, strconv.FormatBool(e.ChipIngressInsecureConnection))
 	add(envChipIngressBatchEmitterEnabled, strconv.FormatBool(e.ChipIngressBatchEmitterEnabled))
 	add(envChipIngressDurableEmitterEnabled, strconv.FormatBool(e.ChipIngressDurableEmitterEnabled))
+	add(envChipIngressBufferSize, strconv.FormatUint(uint64(e.ChipIngressBufferSize), 10))
+	add(envChipIngressMaxBatchSize, strconv.FormatUint(uint64(e.ChipIngressMaxBatchSize), 10))
+	add(envChipIngressMaxConcurrentSends, strconv.Itoa(e.ChipIngressMaxConcurrentSends))
+	add(envChipIngressSendInterval, e.ChipIngressSendInterval.String())
+	add(envChipIngressSendTimeout, e.ChipIngressSendTimeout.String())
+	add(envChipIngressDrainTimeout, e.ChipIngressDrainTimeout.String())
+	add(envChipIngressMaxGRPCRequestSize, strconv.Itoa(e.ChipIngressMaxGRPCRequestSize))
 
 	if e.CRESettings != "" {
 		add(envCRESettings, e.CRESettings)
@@ -582,6 +605,37 @@ func (e *EnvConfig) parse() error {
 		if err != nil {
 			return fmt.Errorf("failed to parse %s: %w", envChipIngressDurableEmitterEnabled, err)
 		}
+		e.ChipIngressBufferSize, err = getUint(envChipIngressBufferSize)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envChipIngressBufferSize, err)
+		}
+		e.ChipIngressMaxBatchSize, err = getUint(envChipIngressMaxBatchSize)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envChipIngressMaxBatchSize, err)
+		}
+		e.ChipIngressMaxConcurrentSends, err = getInt(envChipIngressMaxConcurrentSends)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envChipIngressMaxConcurrentSends, err)
+		}
+		e.ChipIngressSendInterval, err = getDuration(envChipIngressSendInterval)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envChipIngressSendInterval, err)
+		}
+		e.ChipIngressSendTimeout, err = getDuration(envChipIngressSendTimeout)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envChipIngressSendTimeout, err)
+		}
+		e.ChipIngressDrainTimeout, err = getDuration(envChipIngressDrainTimeout)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envChipIngressDrainTimeout, err)
+		}
+		e.ChipIngressMaxGRPCRequestSize, err = getInt(envChipIngressMaxGRPCRequestSize)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envChipIngressMaxGRPCRequestSize, err)
+		}
+		if e.ChipIngressMaxGRPCRequestSize < 0 {
+			return fmt.Errorf("failed to parse %s: value %d must not be negative", envChipIngressMaxGRPCRequestSize, e.ChipIngressMaxGRPCRequestSize)
+		}
 	}
 
 	e.CRESettings = os.Getenv(envCRESettings)
@@ -701,6 +755,18 @@ func getEnv[T any](key string, parse func(string) (T, error)) (t T, err error) {
 	return
 }
 
+func getUint(envKey string) (uint, error) {
+	s := os.Getenv(envKey)
+	if s == "" {
+		return 0, nil
+	}
+	u, err := strconv.ParseUint(s, 10, strconv.IntSize)
+	if err != nil {
+		return 0, err
+	}
+	return uint(u), nil
+}
+
 func getInt(envKey string) (int, error) {
 	s := os.Getenv(envKey)
 	if s == "" {

pkg/loop/config_test.go

@@ -100,6 +100,14 @@ func TestEnvConfig_parse(t *testing.T) {
 				envChipIngressInsecureConnection:  "true",
 				envChipIngressBatchEmitterEnabled: "false",
 
+				envChipIngressBufferSize:         "1000",
+				envChipIngressMaxBatchSize:       "500",
+				envChipIngressMaxConcurrentSends: "10",
+				envChipIngressSendInterval:       "100ms",
+				envChipIngressSendTimeout:        "3s",
+				envChipIngressDrainTimeout:       "10s",
+				envChipIngressMaxGRPCRequestSize: "10485760",
+
 				envCRESettings:        `{"global":{}}`,
 				envCRESettingsDefault: `{"foo":"bar"}`,
 			},
@@ -234,6 +242,14 @@ var envCfgFull = EnvConfig{
 	ChipIngressBatchEmitterEnabled:   false,
 	ChipIngressDurableEmitterEnabled: false,
 
+	ChipIngressBufferSize:         1000,
+	ChipIngressMaxBatchSize:       500,
+	ChipIngressMaxConcurrentSends: 10,
+	ChipIngressSendInterval:       100 * time.Millisecond,
+	ChipIngressSendTimeout:        3 * time.Second,
+	ChipIngressDrainTimeout:       10 * time.Second,
+	ChipIngressMaxGRPCRequestSize: 10485760,
+
 	CRESettings:        `{"global":{}}`,
 	CRESettingsDefault: `{"foo":"bar"}`,
 }
@@ -308,6 +324,13 @@ func TestEnvConfig_AsCmdEnv(t *testing.T) {
 	assert.Equal(t, "chip-ingress.example.com:50051", got[envChipIngressEndpoint])
 	assert.Equal(t, "true", got[envChipIngressInsecureConnection])
 	assert.Equal(t, "false", got[envChipIngressBatchEmitterEnabled])
+	assert.Equal(t, "1000", got[envChipIngressBufferSize])
+	assert.Equal(t, "500", got[envChipIngressMaxBatchSize])
+	assert.Equal(t, "10", got[envChipIngressMaxConcurrentSends])
+	assert.Equal(t, "100ms", got[envChipIngressSendInterval])
+	assert.Equal(t, "3s", got[envChipIngressSendTimeout])
+	assert.Equal(t, "10s", got[envChipIngressDrainTimeout])
+	assert.Equal(t, "10485760", got[envChipIngressMaxGRPCRequestSize])
 
 	assert.JSONEq(t, `{"global":{}}`, got[envCRESettings])
 	assert.JSONEq(t, `{"foo":"bar"}`, got[envCRESettingsDefault])

pkg/loop/server.go

@@ -190,6 +190,13 @@ func (s *Server) start(opts ...ServerOpt) error {
 			ChipIngressEmitterGRPCEndpoint: s.EnvConfig.ChipIngressEndpoint,
 			ChipIngressInsecureConnection:  s.EnvConfig.ChipIngressInsecureConnection,
 			ChipIngressBatchEmitterEnabled: s.EnvConfig.ChipIngressBatchEmitterEnabled,
+			ChipIngressBufferSize:          s.EnvConfig.ChipIngressBufferSize,
+			ChipIngressMaxBatchSize:        s.EnvConfig.ChipIngressMaxBatchSize,
+			ChipIngressMaxConcurrentSends:  s.EnvConfig.ChipIngressMaxConcurrentSends,
+			ChipIngressSendInterval:        s.EnvConfig.ChipIngressSendInterval,
+			ChipIngressSendTimeout:         s.EnvConfig.ChipIngressSendTimeout,
+			ChipIngressDrainTimeout:        s.EnvConfig.ChipIngressDrainTimeout,
+			ChipIngressMaxGRPCRequestSize:  s.EnvConfig.ChipIngressMaxGRPCRequestSize,
 			ChipIngressLogger:              s.Logger,
 			MetricCompressor:               s.EnvConfig.TelemetryMetricCompressor,
 			MetricCardinalityLimit:         *s.EnvConfig.TelemetryMetricCardinalityLimit,