From 1f54ac6a96209f16f9f33e74ffc607934b0e6fd7 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 29 Jun 2026 17:36:06 -0400 Subject: [PATCH 1/2] feat: add chip-ingress batch config keys Add 7 chip-ingress batch configuration keys to enable tunable batch emitter behavior: - BufferSize (default 1000): in-memory queue size for chip-ingress events - MaxBatchSize (default 500): max events per PublishBatch RPC - MaxConcurrentSends (default 10): limits parallel PublishBatch calls - SendInterval (default 100ms): max wait before flushing incomplete batch - SendTimeout (default 3s): per-RPC timeout for PublishBatch - DrainTimeout (default 10s): max shutdown wait to flush queued events - MaxGRPCRequestSize (default 10 MiB): max serialized request size before batch splitting (min 1 MiB enforced by batch client) Wired through beholder.Config, batch_emitter_service.go, and loop EnvConfig (env vars, serialization, parsing) for LOOP plugin support. --- pkg/beholder/batch_emitter_service.go | 5 ++ pkg/beholder/batch_emitter_service_test.go | 2 + pkg/beholder/config.go | 2 + pkg/beholder/testdata/config-example.json | 1 + pkg/loop/config.go | 63 ++++++++++++++++++++++ pkg/loop/config_test.go | 33 ++++++++++-- pkg/loop/server.go | 7 +++ 7 files changed, 108 insertions(+), 5 deletions(-) diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index a81fd82bc9..8879714e97 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/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 { diff --git a/pkg/beholder/batch_emitter_service_test.go b/pkg/beholder/batch_emitter_service_test.go index 4137d450d5..cb24f2f8ab 100644 --- a/pkg/beholder/batch_emitter_service_test.go +++ b/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 { diff --git a/pkg/beholder/config.go b/pkg/beholder/config.go index 09ef907ed7..aeb0d05ffe 100644 --- a/pkg/beholder/config.go +++ b/pkg/beholder/config.go @@ -57,6 +57,7 @@ type Config struct { 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 @@ -158,6 +159,7 @@ func DefaultConfig() Config { 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, } diff --git a/pkg/beholder/testdata/config-example.json b/pkg/beholder/testdata/config-example.json index c8aacc6baf..367f353c30 100644 --- a/pkg/beholder/testdata/config-example.json +++ b/pkg/beholder/testdata/config-example.json @@ -44,6 +44,7 @@ "ChipIngressSendTimeout": 0, "ChipIngressDrainTimeout": 0, "ChipIngressMaxConcurrentSends": 0, + "ChipIngressMaxGRPCRequestSize": 0, "ChipIngressLogger": null, "LogExportTimeout": 1000000000, "LogExportInterval": 1000000000, diff --git a/pkg/loop/config.go b/pkg/loop/config.go index 396c7d979d..5185c44d9d 100644 --- a/pkg/loop/config.go +++ b/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 == "" { diff --git a/pkg/loop/config_test.go b/pkg/loop/config_test.go index d592543fd4..bce09ab3b3 100644 --- a/pkg/loop/config_test.go +++ b/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]) diff --git a/pkg/loop/server.go b/pkg/loop/server.go index fca2fda761..77fa259f9e 100644 --- a/pkg/loop/server.go +++ b/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, } From 504389dcbb5cd499b6527e523860a65bda1f4890 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Mon, 13 Jul 2026 16:36:40 -0400 Subject: [PATCH 2/2] beholder: bump chip-ingress batch defaults, fix histogram buckets Bump ChipIngressBufferSize 1000->10000 and ChipIngressSendInterval 100ms->250ms in beholder.DefaultConfig() to reduce RPC volume and add buffer headroom, based on prod fleet evidence (avg 7.2 events/batch, 93% of batches <=5 events at ~3,190 PublishBatch req/s). Fix chip_ingress.batch.request_size_messages and request_size_bytes histogram buckets: add explicit low-cardinality boundaries for message counts and prepend sub-1KiB buckets for byte sizes, since the observed batch/request sizes are small and the default/coarse buckets produced unreliable quantiles across mixed bucket schemas. Co-authored-by: Cursor --- pkg/beholder/config.go | 8 ++++---- pkg/chipingress/batch/client.go | 5 +++-- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/pkg/beholder/config.go b/pkg/beholder/config.go index aeb0d05ffe..feed01022c 100644 --- a/pkg/beholder/config.go +++ b/pkg/beholder/config.go @@ -51,9 +51,9 @@ 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) @@ -153,9 +153,9 @@ 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, diff --git a/pkg/chipingress/batch/client.go b/pkg/chipingress/batch/client.go index 90c3bca980..dc70f1db39 100644 --- a/pkg/chipingress/batch/client.go +++ b/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, ),