diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index cd16a382fc..700d90bf13 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 := 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), ) diff --git a/pkg/beholder/batch_emitter_service_test.go b/pkg/beholder/batch_emitter_service_test.go index 8593006040..4120380e40 100644 --- a/pkg/beholder/batch_emitter_service_test.go +++ b/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 { diff --git a/pkg/beholder/config.go b/pkg/beholder/config.go index c69412c145..2b61573e1d 100644 --- a/pkg/beholder/config.go +++ b/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, } diff --git a/pkg/beholder/testdata/config-example.json b/pkg/beholder/testdata/config-example.json index ff76b530fd..ec445626f5 100644 --- a/pkg/beholder/testdata/config-example.json +++ b/pkg/beholder/testdata/config-example.json @@ -46,6 +46,7 @@ "ChipIngressSendTimeout": 0, "ChipIngressDrainTimeout": 0, "ChipIngressMaxConcurrentSends": 0, + "ChipIngressMaxGRPCRequestSize": 0, "ChipIngressLogger": null, "LogExportTimeout": 1000000000, "LogExportInterval": 1000000000, diff --git a/pkg/chipingress/batch/client.go b/pkg/chipingress/batch/client.go index 42e0a88cd6..ca66cb944e 100644 --- a/pkg/chipingress/batch/client.go +++ b/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, ), diff --git a/pkg/loop/config.go b/pkg/loop/config.go index bfb7adfd68..ab1531dcb6 100644 --- a/pkg/loop/config.go +++ b/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 == "" { diff --git a/pkg/loop/config_test.go b/pkg/loop/config_test.go index bf60e6e680..555b0ae5b3 100644 --- a/pkg/loop/config_test.go +++ b/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]) diff --git a/pkg/loop/server.go b/pkg/loop/server.go index 1d1a633887..741ae07c7e 100644 --- a/pkg/loop/server.go +++ b/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,