From 0dfeae75ba2d1d48912bf46be81f49d8f64ebc96 Mon Sep 17 00:00:00 2001 From: Pavel <177363085+pkcll@users.noreply.github.com> Date: Fri, 10 Jul 2026 21:06:17 -0400 Subject: [PATCH] Cap chip-ingress batch client buffer at 1 GiB live bytes. Track buffered message bytes alongside the message-count channel cap, expose buffer_live_bytes metrics, and wire MaxMessageBufferBytes through beholder, durableemitter, and LOOP env config. --- go.mod | 2 + go.sum | 2 - pkg/beholder/batch_emitter_service.go | 33 ++++- pkg/beholder/config.go | 9 +- pkg/beholder/testdata/config-example.json | 1 + pkg/chipingress/batch/batcher.go | 10 ++ pkg/chipingress/batch/client.go | 72 +++++++++- pkg/chipingress/batch/client_test.go | 162 ++++++++++++++++++++++ pkg/durableemitter/setup.go | 11 +- pkg/loop/config.go | 12 ++ pkg/loop/server.go | 8 +- 11 files changed, 310 insertions(+), 12 deletions(-) diff --git a/go.mod b/go.mod index 0e9436885c..4bc8577d28 100644 --- a/go.mod +++ b/go.mod @@ -163,3 +163,5 @@ require ( retract v1.3.0 // accidentally published retract v1.3.1 // exists only to retract v1.3.0 + +replace github.com/smartcontractkit/chainlink-common/pkg/chipingress => ./pkg/chipingress diff --git a/go.sum b/go.sum index cfba65084a..9076712f97 100644 --- a/go.sum +++ b/go.sum @@ -258,8 +258,6 @@ github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= github.com/smartcontractkit/chain-selectors v1.0.100 h1:wpiSpmI/eFjY+wx/nPr5VuNF4hki0prIBMKEaQWn3g4= github.com/smartcontractkit/chain-selectors v1.0.100/go.mod h1:qy7whtgG5g+7z0jt0nRyii9bLND9m15NZTzuQPkMZ5w= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 h1:GCzrxDWn3b7jFfEA+WiYRi8CKoegsayiDoJBCjYkneE= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA= github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b h1:VDgJWDipihV9f7M5+d21d1RzSsg5rEv+iI12oN1VQbo= diff --git a/pkg/beholder/batch_emitter_service.go b/pkg/beholder/batch_emitter_service.go index a81fd82bc9..610799eae8 100644 --- a/pkg/beholder/batch_emitter_service.go +++ b/pkg/beholder/batch_emitter_service.go @@ -2,6 +2,7 @@ package beholder import ( "context" + "errors" "fmt" "sync" @@ -63,6 +64,10 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg if drainTimeout == 0 { drainTimeout = defaults.ChipIngressDrainTimeout } + maxMessageBufferBytes := int(cfg.ChipIngressMaxMessageBufferBytes) + if maxMessageBufferBytes == 0 { + maxMessageBufferBytes = int(defaults.ChipIngressMaxMessageBufferBytes) + } meter := otel.Meter("beholder/chip_ingress_batch_emitter") metrics, err := newBatchEmitterMetrics(meter) @@ -73,6 +78,7 @@ func NewChipIngressBatchEmitterService(client chipingress.Client, cfg Config, lg batchClient, err := batch.NewBatchClient(client, batch.WithBatchSize(maxBatchSize), batch.WithMessageBuffer(bufferSize), + batch.WithMaxMessageBufferBytes(maxMessageBufferBytes), batch.WithBatchInterval(sendInterval), batch.WithMaxPublishTimeout(sendTimeout), batch.WithShutdownTimeout(drainTimeout), @@ -167,7 +173,7 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body } }) if queueErr != nil { - e.metrics.eventsDropped.Add(ctx, 1, metricAttrs) + e.metrics.eventsDropped.Add(ctx, 1, e.dropMetricAttrs(domain, entity, queueFailureType(queueErr))) e.eng.Errorw("failed to queue message for chip ingress", "error", queueErr, "domain", domain, "entity", entity) if callback != nil { callback(queueErr) @@ -178,6 +184,31 @@ func (e *ChipIngressBatchEmitterService) emitInternal(ctx context.Context, body }) } +func queueFailureType(err error) string { + switch { + case errors.Is(err, batch.ErrMessageBufferBytesExceeded): + return "buffer_bytes_exceeded" + case errors.Is(err, batch.ErrMessageBufferFull): + return "buffer_full" + default: + return "queue_error" + } +} + +func (e *ChipIngressBatchEmitterService) dropMetricAttrs(domain, entity, failureType string) otelmetric.MeasurementOption { + key := domain + "\x00" + entity + "\x00" + failureType + if v, ok := e.metricAttrsCache.Load(key); ok { + return v.(otelmetric.MeasurementOption) + } + attrs := otelmetric.WithAttributeSet(attribute.NewSet( + attribute.String("domain", domain), + attribute.String("entity", entity), + attribute.String("failure_type", failureType), + )) + v, _ := e.metricAttrsCache.LoadOrStore(key, attrs) + return v.(otelmetric.MeasurementOption) +} + func (e *ChipIngressBatchEmitterService) metricAttrsFor(domain, entity string) otelmetric.MeasurementOption { key := domain + "\x00" + entity if v, ok := e.metricAttrsCache.Load(key); ok { diff --git a/pkg/beholder/config.go b/pkg/beholder/config.go index 09ef907ed7..fea6703cc3 100644 --- a/pkg/beholder/config.go +++ b/pkg/beholder/config.go @@ -8,6 +8,7 @@ import ( "go.opentelemetry.io/otel/sdk/trace" "go.uber.org/zap/zapcore" + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/batch" "github.com/smartcontractkit/chainlink-common/pkg/logger" ) @@ -56,8 +57,9 @@ type Config struct { 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) - ChipIngressMaxConcurrentSends int // Max concurrent PublishBatch calls (default 10) - ChipIngressLogger logger.Logger // Required when ChipIngressBatchEmitterEnabled is true + ChipIngressMaxConcurrentSends int // Max concurrent PublishBatch calls (default 10) + ChipIngressMaxMessageBufferBytes uint // Max live byte size of buffered messages (default 1 GiB) + ChipIngressLogger logger.Logger // Required when ChipIngressBatchEmitterEnabled is true // OTel Log LogExportTimeout time.Duration @@ -157,7 +159,8 @@ func DefaultConfig() Config { ChipIngressSendInterval: 100 * time.Millisecond, ChipIngressSendTimeout: 3 * time.Second, ChipIngressDrainTimeout: 10 * time.Second, - ChipIngressMaxConcurrentSends: defaultMaxConcurrentSends, + ChipIngressMaxConcurrentSends: defaultMaxConcurrentSends, + ChipIngressMaxMessageBufferBytes: uint(batch.DefaultMaxMessageBufferBytes), // 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..eff91d235b 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, + "ChipIngressMaxMessageBufferBytes": 0, "ChipIngressLogger": null, "LogExportTimeout": 1000000000, "LogExportInterval": 1000000000, diff --git a/pkg/chipingress/batch/batcher.go b/pkg/chipingress/batch/batcher.go index c2c50c9308..55bcfbc625 100644 --- a/pkg/chipingress/batch/batcher.go +++ b/pkg/chipingress/batch/batcher.go @@ -7,13 +7,19 @@ import ( // batchWithInterval reads from a channel and calls processFn with batches based on size or time interval. // When the context is cancelled, it automatically flushes any remaining items in the batch. +// onDequeue, when non-nil, is invoked immediately after each message is read from input. func batchWithInterval[T any]( ctx context.Context, input <-chan T, batchSize int, interval time.Duration, processFn func([]T), + onDequeue ...func(T), ) { + var dequeue func(T) + if len(onDequeue) > 0 { + dequeue = onDequeue[0] + } var batch []T timer := time.NewTimer(interval) timer.Stop() @@ -38,6 +44,10 @@ func batchWithInterval[T any]( return } + if dequeue != nil { + dequeue(msg) + } + // Start timer on first message in batch if len(batch) == 0 { timer.Reset(interval) diff --git a/pkg/chipingress/batch/client.go b/pkg/chipingress/batch/client.go index 90c3bca980..31a1356138 100644 --- a/pkg/chipingress/batch/client.go +++ b/pkg/chipingress/batch/client.go @@ -19,9 +19,23 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/chipingress" ) +// DefaultMaxMessageBufferBytes is the default live-byte cap for the message buffer channel. +const DefaultMaxMessageBufferBytes = 1 << 30 // 1 GiB + +// defaultMaxMessageBufferBytes is an alias kept for internal use. +const defaultMaxMessageBufferBytes = DefaultMaxMessageBufferBytes + +var ( + // ErrMessageBufferFull is returned when the message-count buffer is full. + ErrMessageBufferFull = errors.New("message buffer is full") + // ErrMessageBufferBytesExceeded is returned when enqueueing would exceed the byte cap. + ErrMessageBufferBytesExceeded = errors.New("message buffer bytes exceeded") +) + type messageWithCallback struct { event *chipingress.CloudEventPb callback func(error) + byteSize int } type seqnumKey struct { @@ -62,6 +76,9 @@ type Client struct { batchInterval time.Duration maxPublishTimeout time.Duration messageBuffer chan *messageWithCallback + maxMessageBufferBytes int + bufferedBytes atomic.Int64 + bufferMu sync.Mutex stopCh stopCh log *zap.SugaredLogger callbackWg sync.WaitGroup @@ -82,6 +99,7 @@ type batchClientMetrics struct { requestSizeBytes otelmetric.Int64Histogram requestLatencyMS otelmetric.Float64Histogram configInfo otelmetric.Int64Gauge + bufferLiveBytes otelmetric.Int64ObservableGauge batchSplitsTotal otelmetric.Int64Counter resultsMismatchTotal otelmetric.Int64Counter batchSizeAttr otelmetric.MeasurementOption @@ -105,6 +123,7 @@ func NewBatchClient(client chipingress.Client, opts ...Opt) (*Client, error) { transactionEnabled: false, maxConcurrentSends: make(chan struct{}, 1), messageBuffer: make(chan *messageWithCallback, 200), + maxMessageBufferBytes: defaultMaxMessageBufferBytes, batchInterval: 100 * time.Millisecond, maxPublishTimeout: 5 * time.Second, stopCh: make(chan struct{}), @@ -132,6 +151,7 @@ func NewBatchClient(client chipingress.Client, opts ...Opt) (*Client, error) { // context that is cancelled when Stop is called. func (b *Client) Start(ctx context.Context) { b.metrics.recordConfig(ctx, b) + b.metrics.registerBufferLiveBytesCallback(b) b.started = true // Detach from the caller's cancellation but keep its values (trace IDs, etc.). @@ -158,6 +178,11 @@ func (b *Client) Start(ctx context.Context) { // sendBatch still enforces maxPublishTimeout for each publish call. b.sendBatch(context.WithoutCancel(batcherCtx), batch) }, + func(msg *messageWithCallback) { + if msg.byteSize > 0 { + b.bufferedBytes.Add(-int64(msg.byteSize)) + } + }, ) }() } @@ -265,16 +290,30 @@ func (b *Client) QueueMessage(event *chipingress.CloudEventPb, callback func(err }, } + msgSize := proto.Size(eventToQueue) msg := &messageWithCallback{ event: eventToQueue, callback: callback, + byteSize: msgSize, + } + + b.bufferMu.Lock() + if b.maxMessageBufferBytes > 0 { + nextBytes := b.bufferedBytes.Load() + int64(msgSize) + if nextBytes > int64(b.maxMessageBufferBytes) { + b.bufferMu.Unlock() + return ErrMessageBufferBytesExceeded + } } select { case b.messageBuffer <- msg: + b.bufferedBytes.Add(int64(msgSize)) + b.bufferMu.Unlock() return nil default: - return errors.New("message buffer is full") + b.bufferMu.Unlock() + return ErrMessageBufferFull } } @@ -497,6 +536,14 @@ func WithMessageBuffer(messageBufferSize int) Opt { } } +// WithMaxMessageBufferBytes sets the maximum live byte size of messages buffered in the +// message channel. Zero disables the byte cap (intended for tests). +func WithMaxMessageBufferBytes(maxBytes int) Opt { + return func(c *Client) { + c.maxMessageBufferBytes = maxBytes + } +} + // WithMaxPublishTimeout sets the maximum time to wait for a batch publish operation func WithMaxPublishTimeout(maxPublishTimeout time.Duration) Opt { return func(c *Client) { @@ -573,6 +620,14 @@ func newBatchClientMetrics() (batchClientMetrics, error) { if err != nil { return batchClientMetrics{}, err } + bufferLiveBytes, err := meter.Int64ObservableGauge( + "chip_ingress.batch.buffer_live_bytes", + otelmetric.WithDescription("Live byte size of messages currently buffered in the batch client message channel"), + otelmetric.WithUnit("By"), + ) + 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 the effective gRPC request size limit (max request size minus reserved framing overhead)"), @@ -596,6 +651,7 @@ func newBatchClientMetrics() (batchClientMetrics, error) { requestSizeBytes: requestSizeBytes, requestLatencyMS: requestLatencyMS, configInfo: configInfo, + bufferLiveBytes: bufferLiveBytes, batchSplitsTotal: batchSplitsTotal, resultsMismatchTotal: resultsMismatchTotal, successStatusAttr: otelmetric.WithAttributeSet(attribute.NewSet( @@ -607,6 +663,19 @@ func newBatchClientMetrics() (batchClientMetrics, error) { }, nil } +func (m *batchClientMetrics) registerBufferLiveBytesCallback(c *Client) { + if m.bufferLiveBytes == nil { + return + } + _, _ = otel.Meter("chipingress/batch_client").RegisterCallback( + func(_ context.Context, o otelmetric.Observer) error { + o.ObserveInt64(m.bufferLiveBytes, c.bufferedBytes.Load()) + return nil + }, + m.bufferLiveBytes, + ) +} + func (m *batchClientMetrics) recordConfig(ctx context.Context, c *Client) { m.batchSizeAttr = otelmetric.WithAttributeSet(attribute.NewSet( attribute.Int("max_batch_size", c.batchSize), @@ -624,6 +693,7 @@ func (m *batchClientMetrics) recordConfig(ctx context.Context, c *Client) { attribute.Bool("clone_event", c.cloneEvent), attribute.Bool("transaction_enabled", c.transactionEnabled), attribute.Int("max_grpc_request_size_bytes", c.maxGRPCRequestSize), + attribute.Int("max_message_buffer_bytes", c.maxMessageBufferBytes), )) } diff --git a/pkg/chipingress/batch/client_test.go b/pkg/chipingress/batch/client_test.go index ce29dd0c0e..a08e927a2d 100644 --- a/pkg/chipingress/batch/client_test.go +++ b/pkg/chipingress/batch/client_test.go @@ -7,6 +7,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "testing" "time" @@ -65,6 +66,16 @@ func TestNewBatchClient(t *testing.T) { assert.Equal(t, 1000, cap(client.messageBuffer)) }) + t.Run("WithMaxMessageBufferBytes", func(t *testing.T) { + client, err := NewBatchClient(mocks.NewClient(t), WithMaxMessageBufferBytes(4096)) + require.NoError(t, err) + assert.Equal(t, 4096, client.maxMessageBufferBytes) + + defaultClient, err := NewBatchClient(mocks.NewClient(t)) + require.NoError(t, err) + assert.Equal(t, defaultMaxMessageBufferBytes, defaultClient.maxMessageBufferBytes) + }) + t.Run("WithMaxGRPCRequestSize", func(t *testing.T) { t.Run("applies value at or above minimum", func(t *testing.T) { client, err := NewBatchClient(mocks.NewClient(t), WithMaxGRPCRequestSize(4*1024*1024)) @@ -203,6 +214,157 @@ func TestQueueMessage(t *testing.T) { }) } +func TestQueueMessageByteCap(t *testing.T) { + t.Run("drops when byte cap exceeded before count cap", func(t *testing.T) { + client, err := NewBatchClient( + mocks.NewClient(t), + WithMessageBuffer(100), + WithMaxMessageBufferBytes(512), + ) + require.NoError(t, err) + + event := eventWithMinProtoSize(t, 400) + require.NoError(t, client.QueueMessage(event, nil)) + assert.Greater(t, client.bufferedBytes.Load(), int64(0)) + + err = client.QueueMessage(eventWithMinProtoSize(t, 400), nil) + require.ErrorIs(t, err, ErrMessageBufferBytesExceeded) + assert.Len(t, client.messageBuffer, 1) + }) + + t.Run("drops when count cap hit before byte cap", func(t *testing.T) { + client, err := NewBatchClient( + mocks.NewClient(t), + WithMessageBuffer(1), + WithMaxMessageBufferBytes(0), + ) + require.NoError(t, err) + + event := &chipingress.CloudEventPb{Id: "a", Source: "src", Type: "typ"} + require.NoError(t, client.QueueMessage(event, nil)) + err = client.QueueMessage(event, nil) + require.ErrorIs(t, err, ErrMessageBufferFull) + }) + + t.Run("rejects single message larger than byte cap", func(t *testing.T) { + client, err := NewBatchClient( + mocks.NewClient(t), + WithMessageBuffer(10), + WithMaxMessageBufferBytes(128), + ) + require.NoError(t, err) + + err = client.QueueMessage(eventWithMinProtoSize(t, 256), nil) + require.ErrorIs(t, err, ErrMessageBufferBytesExceeded) + assert.Empty(t, client.messageBuffer) + }) + + t.Run("bufferedBytes drains to zero after batcher consumes messages", func(t *testing.T) { + mockClient := mocks.NewClient(t) + mockClient.EXPECT().Close().Return(nil).Maybe() + done := make(chan struct{}) + mockClient. + On("PublishBatch", mock.Anything, mock.Anything). + Return(&chipingress.PublishResponse{}, nil). + Run(func(_ mock.Arguments) { close(done) }). + Once() + + client, err := NewBatchClient( + mockClient, + WithBatchSize(2), + WithBatchInterval(time.Hour), + WithMessageBuffer(10), + WithMaxMessageBufferBytes(1<<20), + ) + require.NoError(t, err) + client.Start(t.Context()) + defer client.Stop() + + event := eventWithMinProtoSize(t, 256) + require.NoError(t, client.QueueMessage(event, nil)) + require.NoError(t, client.QueueMessage(event, nil)) + assert.Greater(t, client.bufferedBytes.Load(), int64(0)) + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timeout waiting for batch publish") + } + + assert.Eventually(t, func() bool { + return client.bufferedBytes.Load() == 0 + }, time.Second, 10*time.Millisecond) + }) + + t.Run("records max_message_buffer_bytes in config metric", func(t *testing.T) { + reader, restore := useTestMeterProvider(t) + defer restore() + + const capBytes = 8192 + mockClient := mocks.NewClient(t) + mockClient.EXPECT().Close().Return(nil).Maybe() + client, err := NewBatchClient( + mockClient, + WithMaxMessageBufferBytes(capBytes), + ) + require.NoError(t, err) + client.Start(t.Context()) + client.Stop() + + rm := collectResourceMetrics(t, reader) + config := mustMetric(t, rm, "chip_ingress.batch.config.info") + configGauge, ok := config.Data.(metricdata.Gauge[int64]) + require.True(t, ok) + require.NotEmpty(t, configGauge.DataPoints) + assert.True(t, hasIntAttr(configGauge.DataPoints[0].Attributes, "max_message_buffer_bytes", capBytes)) + }) + + t.Run("concurrent enqueue respects byte cap", func(t *testing.T) { + client, err := NewBatchClient( + mocks.NewClient(t), + WithMessageBuffer(1000), + WithMaxMessageBufferBytes(4096), + WithBatchInterval(time.Hour), + ) + require.NoError(t, err) + + event := eventWithMinProtoSize(t, 512) + var wg sync.WaitGroup + var dropped atomic.Int64 + for range 20 { + wg.Add(1) + go func() { + defer wg.Done() + if err := client.QueueMessage(event, nil); err != nil { + dropped.Add(1) + } + }() + } + wg.Wait() + + assert.Greater(t, dropped.Load(), int64(0)) + assert.LessOrEqual(t, client.bufferedBytes.Load(), int64(4096)) + }) +} + +func eventWithMinProtoSize(t *testing.T, minSize int) *chipingress.CloudEventPb { + t.Helper() + data := []byte{} + event := &chipingress.CloudEventPb{ + Id: "payload-event", + Source: "test-source", + Type: "test.event.type", + Data: &cepb.CloudEvent_BinaryData{ + BinaryData: data, + }, + } + for proto.Size(event) < minSize { + data = append(data, 'x') + event.Data = &cepb.CloudEvent_BinaryData{BinaryData: data} + } + return event +} + func TestSendBatch(t *testing.T) { t.Run("successfully sends a batch", func(t *testing.T) { mockClient := mocks.NewClient(t) diff --git a/pkg/durableemitter/setup.go b/pkg/durableemitter/setup.go index 6fa0c58f10..676afb7715 100644 --- a/pkg/durableemitter/setup.go +++ b/pkg/durableemitter/setup.go @@ -70,6 +70,8 @@ type SetupConfig struct { // channel. QueueMessage drops events (non-blocking send) when this is full, // which happens when emit throughput outpaces the batcher. MessageBufferSize int // default: 10000 + // MaxMessageBufferBytes caps live buffered message bytes. Zero uses batch client default (1 GiB). + MaxMessageBufferBytes int // EmitterConfig overrides DefaultConfig when non-nil. EmitterConfig *Config @@ -107,14 +109,19 @@ func Setup( return nil, fmt.Errorf("failed to create batch chip ingress client: %w", err) } - batchClient, err := chipingressbatch.NewBatchClient(batchChipClient, + batchOpts := []chipingressbatch.Opt{ chipingressbatch.WithBatchSize(defaultInt(cfg.BatchSize, 50)), chipingressbatch.WithBatchInterval(defaultDuration(cfg.BatchInterval, 50*time.Millisecond)), chipingressbatch.WithMaxConcurrentSends(defaultInt(cfg.MaxConcurrentSends, 4)), chipingressbatch.WithMessageBuffer(defaultInt(cfg.MessageBufferSize, 10_000)), chipingressbatch.WithMaxPublishTimeout(defaultDuration(cfg.MaxPublishTimeout, 5*time.Second)), chipingressbatch.WithShutdownTimeout(defaultDuration(cfg.ShutdownTimeout, 30*time.Second)), - ) + } + if cfg.MaxMessageBufferBytes > 0 { + batchOpts = append(batchOpts, chipingressbatch.WithMaxMessageBufferBytes(cfg.MaxMessageBufferBytes)) + } + + batchClient, err := chipingressbatch.NewBatchClient(batchChipClient, batchOpts...) if err != nil { _ = batchChipClient.Close() return nil, fmt.Errorf("failed to create batch client: %w", err) diff --git a/pkg/loop/config.go b/pkg/loop/config.go index 58d956bbfc..edf001ac07 100644 --- a/pkg/loop/config.go +++ b/pkg/loop/config.go @@ -98,6 +98,7 @@ const ( envChipIngressInsecureConnection = "CL_CHIP_INGRESS_INSECURE_CONNECTION" envChipIngressBatchEmitterEnabled = "CL_CHIP_INGRESS_BATCH_EMITTER_ENABLED" envChipIngressDurableEmitterEnabled = "CL_CHIP_INGRESS_DURABLE_EMITTER_ENABLED" + envChipIngressMaxMessageBufferBytes = "CL_CHIP_INGRESS_MAX_MESSAGE_BUFFER_BYTES" envCRESettings = cresettings.EnvNameSettings envCRESettingsDefault = cresettings.EnvNameSettingsDefault @@ -112,6 +113,7 @@ type EnvConfig struct { ChipIngressInsecureConnection bool ChipIngressBatchEmitterEnabled bool ChipIngressDurableEmitterEnabled bool + ChipIngressMaxMessageBufferBytes uint CRESettings string CRESettingsDefault string @@ -306,6 +308,9 @@ func (e *EnvConfig) AsCmdEnv() (env []string) { add(envChipIngressInsecureConnection, strconv.FormatBool(e.ChipIngressInsecureConnection)) add(envChipIngressBatchEmitterEnabled, strconv.FormatBool(e.ChipIngressBatchEmitterEnabled)) add(envChipIngressDurableEmitterEnabled, strconv.FormatBool(e.ChipIngressDurableEmitterEnabled)) + if e.ChipIngressMaxMessageBufferBytes > 0 { + add(envChipIngressMaxMessageBufferBytes, strconv.FormatUint(uint64(e.ChipIngressMaxMessageBufferBytes), 10)) + } if e.CRESettings != "" { add(envCRESettings, e.CRESettings) @@ -550,6 +555,13 @@ func (e *EnvConfig) parse() error { if err != nil { return fmt.Errorf("failed to parse %s: %w", envChipIngressDurableEmitterEnabled, err) } + if v := os.Getenv(envChipIngressMaxMessageBufferBytes); v != "" { + n, parseErr := strconv.ParseUint(v, 10, 64) + if parseErr != nil { + return fmt.Errorf("failed to parse %s: %w", envChipIngressMaxMessageBufferBytes, parseErr) + } + e.ChipIngressMaxMessageBufferBytes = uint(n) + } } e.CRESettings = os.Getenv(envCRESettings) diff --git a/pkg/loop/server.go b/pkg/loop/server.go index fca2fda761..48ebbb1230 100644 --- a/pkg/loop/server.go +++ b/pkg/loop/server.go @@ -188,7 +188,8 @@ func (s *Server) start(opts ...ServerOpt) error { ChipIngressEmitterEnabled: s.EnvConfig.ChipIngressEndpoint != "", ChipIngressEmitterGRPCEndpoint: s.EnvConfig.ChipIngressEndpoint, ChipIngressInsecureConnection: s.EnvConfig.ChipIngressInsecureConnection, - ChipIngressBatchEmitterEnabled: s.EnvConfig.ChipIngressBatchEmitterEnabled, + ChipIngressBatchEmitterEnabled: s.EnvConfig.ChipIngressBatchEmitterEnabled, + ChipIngressMaxMessageBufferBytes: s.EnvConfig.ChipIngressMaxMessageBufferBytes, ChipIngressLogger: s.Logger, MetricCompressor: s.EnvConfig.TelemetryMetricCompressor, } @@ -356,8 +357,9 @@ func (s *Server) start(opts ...ServerOpt) error { emitterCfg := durableemitter.DefaultConfig() emitterCfg.Metrics = &durableemitter.DurableEmitterMetricsConfig{} durableCfg := durableemitter.SetupConfig{ - Endpoint: s.EnvConfig.ChipIngressEndpoint, - InsecureConnection: s.EnvConfig.ChipIngressInsecureConnection, + Endpoint: s.EnvConfig.ChipIngressEndpoint, + InsecureConnection: s.EnvConfig.ChipIngressInsecureConnection, + MaxMessageBufferBytes: int(s.EnvConfig.ChipIngressMaxMessageBufferBytes), Auth: durableemitter.AuthConfig{ AuthHeaders: s.EnvConfig.TelemetryAuthHeaders, AuthHeadersTTL: s.EnvConfig.TelemetryAuthHeadersTTL,