feat: add chip-ingress batch config keys

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 := int(cfg.ChipIngressMaxGRPCRequestSize)
+	if maxGRPCRequestSize == 0 {
+		maxGRPCRequestSize = int(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),
 	)
 	if err != nil {

pkg/beholder/batch_emitter_service_test.go

@@ -31,6 +31,7 @@ func newTestConfig() beholder.Config {
 		ChipIngressSendInterval:       50 * time.Millisecond,
 		ChipIngressSendTimeout:        5 * time.Second,
 		ChipIngressDrainTimeout:       5 * time.Second,
+		ChipIngressMaxGRPCRequestSize: 1024 * 1024,
 	}
 }
 
@@ -573,6 +574,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

@@ -51,12 +51,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)
+	ChipIngressBufferSize          uint          // Message buffer size (default 10000)
 	ChipIngressMaxBatchSize        uint          // Max events per PublishBatch call (default 500)
-	ChipIngressSendInterval        time.Duration // Flush interval (default 100ms)
+	ChipIngressSendInterval        time.Duration // Flush interval (default 250ms)
 	ChipIngressSendTimeout         time.Duration // Timeout per PublishBatch call (default 3s)
 	ChipIngressDrainTimeout        time.Duration // Max time to flush remaining events on shutdown (default 10s)
 	ChipIngressMaxConcurrentSends  int           // Max concurrent PublishBatch calls (default 10)
+	ChipIngressMaxGRPCRequestSize  uint          // Max serialized PublishBatch request size in bytes (default 10 MiB)
 	ChipIngressLogger              logger.Logger // Required when ChipIngressBatchEmitterEnabled is true
 
 	// OTel Log
@@ -152,12 +153,13 @@ func DefaultConfig() Config {
 		LogCompressor: "gzip",
 		// Chip Ingress Batch Emitter
 		ChipIngressBatchEmitterEnabled: false,
-		ChipIngressBufferSize:          1000,
+		ChipIngressBufferSize:          10000,
 		ChipIngressMaxBatchSize:        500,
-		ChipIngressSendInterval:        100 * time.Millisecond,
+		ChipIngressSendInterval:        250 * time.Millisecond,
 		ChipIngressSendTimeout:         3 * time.Second,
 		ChipIngressDrainTimeout:        10 * 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

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

pkg/chipingress/batch/client.go

@@ -539,6 +539,7 @@ func newBatchClientMetrics() (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
@@ -548,8 +549,8 @@ func newBatchClientMetrics() (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

@@ -91,6 +91,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
 )
@@ -105,6 +113,14 @@ type EnvConfig struct {
 	ChipIngressBatchEmitterEnabled   bool
 	ChipIngressDurableEmitterEnabled bool
 
+	ChipIngressBufferSize         uint
+	ChipIngressMaxBatchSize       uint
+	ChipIngressMaxConcurrentSends int
+	ChipIngressSendInterval       time.Duration
+	ChipIngressSendTimeout        time.Duration
+	ChipIngressDrainTimeout       time.Duration
+	ChipIngressMaxGRPCRequestSize uint
+
 	CRESettings        string
 	CRESettingsDefault string
 
@@ -269,6 +285,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.FormatUint(uint64(e.ChipIngressMaxGRPCRequestSize), 10))
 
 	if e.CRESettings != "" {
 		add(envCRESettings, e.CRESettings)
@@ -513,6 +536,34 @@ 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 = getUint(envChipIngressMaxGRPCRequestSize)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envChipIngressMaxGRPCRequestSize, err)
+		}
 	}
 
 	e.CRESettings = os.Getenv(envCRESettings)
@@ -605,6 +656,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, 64)
+	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

@@ -86,11 +86,19 @@ func TestEnvConfig_parse(t *testing.T) {
 				envTelemetryPrometheusBridgeEnabled:   "true",
 				envTelemetryPrometheusBridgePrefixes:  "foo,bar",
 
-				envChipIngressEndpoint:            "chip-ingress.example.com:50051",
-				envChipIngressInsecureConnection:  "true",
-				envChipIngressBatchEmitterEnabled: "false",
-
-				envCRESettings:        `{"global":{}}`,
+			envChipIngressEndpoint:            "chip-ingress.example.com:50051",
+			envChipIngressInsecureConnection:  "true",
+			envChipIngressBatchEmitterEnabled: "false",
+
+			envChipIngressBufferSize:         "1000",
+			envChipIngressMaxBatchSize:       "500",
+			envChipIngressMaxConcurrentSends: "10",
+			envChipIngressSendInterval:       "100ms",
+			envChipIngressSendTimeout:        "3s",
+			envChipIngressDrainTimeout:       "10s",
+			envChipIngressMaxGRPCRequestSize: "10485760",
+
+			envCRESettings:        `{"global":{}}`,
 				envCRESettingsDefault: `{"foo":"bar"}`,
 			},
 			expectError:  false,
@@ -205,6 +213,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"}`,
 }
@@ -269,6 +285,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

@@ -189,6 +189,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,
 		}