diff --git a/core/cmd/shell.go b/core/cmd/shell.go index 4494588bdbd..2e18564cd1c 100644 --- a/core/cmd/shell.go +++ b/core/cmd/shell.go @@ -155,6 +155,13 @@ func newBeholderClient( ChipIngressEmitterGRPCEndpoint: cfgTelemetry.ChipIngressEndpoint(), ChipIngressInsecureConnection: cfgTelemetry.ChipIngressInsecureConnection(), ChipIngressBatchEmitterEnabled: cfgTelemetry.ChipIngressBatchEmitterEnabled(), + ChipIngressBufferSize: cfgTelemetry.ChipIngressBufferSize(), + ChipIngressMaxBatchSize: cfgTelemetry.ChipIngressMaxBatchSize(), + ChipIngressMaxConcurrentSends: cfgTelemetry.ChipIngressMaxConcurrentSends(), + ChipIngressSendInterval: cfgTelemetry.ChipIngressSendInterval(), + ChipIngressSendTimeout: cfgTelemetry.ChipIngressSendTimeout(), + ChipIngressDrainTimeout: cfgTelemetry.ChipIngressDrainTimeout(), + ChipIngressMaxGRPCRequestSize: cfgTelemetry.ChipIngressMaxGRPCRequestSize(), ChipIngressLogger: lggr, LogStreamingEnabled: cfgTelemetry.LogStreamingEnabled(), LogLevel: cfgTelemetry.LogLevel(), diff --git a/core/config/docs/core.toml b/core/config/docs/core.toml index 67f235bf8b1..c45724a539e 100644 --- a/core/config/docs/core.toml +++ b/core/config/docs/core.toml @@ -887,6 +887,20 @@ ChipIngressInsecureConnection = false # Default # ChipIngressBatchEmitterEnabled enables batching for chip-ingress events. # When false, events are sent individually (legacy behavior). ChipIngressBatchEmitterEnabled = true # Default +# ChipIngressBufferSize is the in-memory queue size for chip-ingress events. +ChipIngressBufferSize = 1000 # Default +# ChipIngressMaxBatchSize is the max events per PublishBatch RPC. +ChipIngressMaxBatchSize = 500 # Default +# ChipIngressMaxConcurrentSends limits parallel PublishBatch calls. +ChipIngressMaxConcurrentSends = 10 # Default +# ChipIngressSendInterval is the max wait before flushing an incomplete batch. +ChipIngressSendInterval = '100ms' # Default +# ChipIngressSendTimeout is the per-RPC timeout for PublishBatch. +ChipIngressSendTimeout = '3s' # Default +# ChipIngressDrainTimeout is the max shutdown wait to flush queued events. +ChipIngressDrainTimeout = '10s' # Default +# ChipIngressMaxGRPCRequestSize is the max serialized PublishBatch request size in bytes. Batches exceeding this are split before send (min 1 MiB enforced by batch client). +ChipIngressMaxGRPCRequestSize = 10485760 # Default # DurableEmitterEnabled enables persisting outbound CHIP events to Postgres for at-least-once delivery. DurableEmitterEnabled = false # Default # HeartbeatInterval is the interval at which a the application heartbeat is sent to telemetry backends. diff --git a/core/config/telemetry_config.go b/core/config/telemetry_config.go index 1036eac775e..adfc431b1bf 100644 --- a/core/config/telemetry_config.go +++ b/core/config/telemetry_config.go @@ -19,6 +19,13 @@ type Telemetry interface { ChipIngressEndpoint() string ChipIngressInsecureConnection() bool ChipIngressBatchEmitterEnabled() bool + ChipIngressBufferSize() uint + ChipIngressMaxBatchSize() uint + ChipIngressMaxConcurrentSends() int + ChipIngressSendInterval() time.Duration + ChipIngressSendTimeout() time.Duration + ChipIngressDrainTimeout() time.Duration + ChipIngressMaxGRPCRequestSize() uint DurableEmitterEnabled() bool HeartbeatInterval() time.Duration LogStreamingEnabled() bool diff --git a/core/config/toml/types.go b/core/config/toml/types.go index 80b8338de4d..cf89b80c680 100644 --- a/core/config/toml/types.go +++ b/core/config/toml/types.go @@ -2968,6 +2968,13 @@ type Telemetry struct { ChipIngressEndpoint *string ChipIngressInsecureConnection *bool ChipIngressBatchEmitterEnabled *bool + ChipIngressBufferSize *uint + ChipIngressMaxBatchSize *uint + ChipIngressMaxConcurrentSends *int + ChipIngressSendInterval *commonconfig.Duration + ChipIngressSendTimeout *commonconfig.Duration + ChipIngressDrainTimeout *commonconfig.Duration + ChipIngressMaxGRPCRequestSize *uint DurableEmitterEnabled *bool HeartbeatInterval *commonconfig.Duration LogLevel *string @@ -3018,6 +3025,27 @@ func (b *Telemetry) setFrom(f *Telemetry) { if v := f.ChipIngressBatchEmitterEnabled; v != nil { b.ChipIngressBatchEmitterEnabled = v } + if v := f.ChipIngressBufferSize; v != nil { + b.ChipIngressBufferSize = v + } + if v := f.ChipIngressMaxBatchSize; v != nil { + b.ChipIngressMaxBatchSize = v + } + if v := f.ChipIngressMaxConcurrentSends; v != nil { + b.ChipIngressMaxConcurrentSends = v + } + if v := f.ChipIngressSendInterval; v != nil { + b.ChipIngressSendInterval = v + } + if v := f.ChipIngressSendTimeout; v != nil { + b.ChipIngressSendTimeout = v + } + if v := f.ChipIngressDrainTimeout; v != nil { + b.ChipIngressDrainTimeout = v + } + if v := f.ChipIngressMaxGRPCRequestSize; v != nil { + b.ChipIngressMaxGRPCRequestSize = v + } if v := f.DurableEmitterEnabled; v != nil { b.DurableEmitterEnabled = v } @@ -3064,6 +3092,27 @@ func (b *Telemetry) ValidateConfig() (err error) { if ratio := b.TraceSampleRatio; ratio != nil && (*ratio < 0 || *ratio > 1) { err = errors.Join(err, configutils.ErrInvalid{Name: "TraceSampleRatio", Value: *ratio, Msg: "must be between 0 and 1"}) } + if v := b.ChipIngressBufferSize; v != nil && *v == 0 { + err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressBufferSize", Value: *v, Msg: "must be greater than 0"}) + } + if v := b.ChipIngressMaxBatchSize; v != nil && *v == 0 { + err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressMaxBatchSize", Value: *v, Msg: "must be greater than 0"}) + } + if v := b.ChipIngressMaxConcurrentSends; v != nil && *v <= 0 { + err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressMaxConcurrentSends", Value: *v, Msg: "must be greater than 0"}) + } + if v := b.ChipIngressSendInterval; v != nil && v.Duration() <= 0 { + err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressSendInterval", Value: v.Duration(), Msg: "must be greater than 0"}) + } + if v := b.ChipIngressSendTimeout; v != nil && v.Duration() <= 0 { + err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressSendTimeout", Value: v.Duration(), Msg: "must be greater than 0"}) + } + if v := b.ChipIngressDrainTimeout; v != nil && v.Duration() <= 0 { + err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressDrainTimeout", Value: v.Duration(), Msg: "must be greater than 0"}) + } + if v := b.ChipIngressMaxGRPCRequestSize; v != nil && *v == 0 { + err = errors.Join(err, configutils.ErrInvalid{Name: "ChipIngressMaxGRPCRequestSize", Value: *v, Msg: "must be greater than 0"}) + } return err } diff --git a/core/services/chainlink/config_telemetry.go b/core/services/chainlink/config_telemetry.go index df61e0d3269..24d8eee2548 100644 --- a/core/services/chainlink/config_telemetry.go +++ b/core/services/chainlink/config_telemetry.go @@ -107,6 +107,55 @@ func (b *telemetryConfig) ChipIngressBatchEmitterEnabled() bool { return *b.s.ChipIngressBatchEmitterEnabled } +func (b *telemetryConfig) ChipIngressBufferSize() uint { + if b.s.ChipIngressBufferSize == nil { + return 0 + } + return *b.s.ChipIngressBufferSize +} + +func (b *telemetryConfig) ChipIngressMaxBatchSize() uint { + if b.s.ChipIngressMaxBatchSize == nil { + return 0 + } + return *b.s.ChipIngressMaxBatchSize +} + +func (b *telemetryConfig) ChipIngressMaxConcurrentSends() int { + if b.s.ChipIngressMaxConcurrentSends == nil { + return 0 + } + return *b.s.ChipIngressMaxConcurrentSends +} + +func (b *telemetryConfig) ChipIngressSendInterval() time.Duration { + if b.s.ChipIngressSendInterval == nil { + return 0 + } + return b.s.ChipIngressSendInterval.Duration() +} + +func (b *telemetryConfig) ChipIngressSendTimeout() time.Duration { + if b.s.ChipIngressSendTimeout == nil { + return 0 + } + return b.s.ChipIngressSendTimeout.Duration() +} + +func (b *telemetryConfig) ChipIngressDrainTimeout() time.Duration { + if b.s.ChipIngressDrainTimeout == nil { + return 0 + } + return b.s.ChipIngressDrainTimeout.Duration() +} + +func (b *telemetryConfig) ChipIngressMaxGRPCRequestSize() uint { + if b.s.ChipIngressMaxGRPCRequestSize == nil { + return 0 + } + return *b.s.ChipIngressMaxGRPCRequestSize +} + func (b *telemetryConfig) DurableEmitterEnabled() bool { if b.s.DurableEmitterEnabled == nil { return false diff --git a/core/services/chainlink/config_telemetry_test.go b/core/services/chainlink/config_telemetry_test.go index 5baf19de43c..31c8e924e1f 100644 --- a/core/services/chainlink/config_telemetry_test.go +++ b/core/services/chainlink/config_telemetry_test.go @@ -245,6 +245,129 @@ func ptrFloat(f float64) *float64 { return &f } +func ptrUint(u uint) *uint { + return &u +} + +func TestTelemetryConfig_ChipIngressBufferSize(t *testing.T) { + tests := []struct { + name string + telemetry toml.Telemetry + expected uint + }{ + {"ChipIngressBufferSizeSet", toml.Telemetry{ChipIngressBufferSize: ptrUint(1000)}, 1000}, + {"ChipIngressBufferSizeNil", toml.Telemetry{ChipIngressBufferSize: nil}, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tc := telemetryConfig{s: tt.telemetry} + assert.Equal(t, tt.expected, tc.ChipIngressBufferSize()) + }) + } +} + +func TestTelemetryConfig_ChipIngressMaxBatchSize(t *testing.T) { + tests := []struct { + name string + telemetry toml.Telemetry + expected uint + }{ + {"ChipIngressMaxBatchSizeSet", toml.Telemetry{ChipIngressMaxBatchSize: ptrUint(500)}, 500}, + {"ChipIngressMaxBatchSizeNil", toml.Telemetry{ChipIngressMaxBatchSize: nil}, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tc := telemetryConfig{s: tt.telemetry} + assert.Equal(t, tt.expected, tc.ChipIngressMaxBatchSize()) + }) + } +} + +func TestTelemetryConfig_ChipIngressMaxConcurrentSends(t *testing.T) { + tests := []struct { + name string + telemetry toml.Telemetry + expected int + }{ + {"ChipIngressMaxConcurrentSendsSet", toml.Telemetry{ChipIngressMaxConcurrentSends: ptrInt(10)}, 10}, + {"ChipIngressMaxConcurrentSendsNil", toml.Telemetry{ChipIngressMaxConcurrentSends: nil}, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tc := telemetryConfig{s: tt.telemetry} + assert.Equal(t, tt.expected, tc.ChipIngressMaxConcurrentSends()) + }) + } +} + +func TestTelemetryConfig_ChipIngressSendInterval(t *testing.T) { + tests := []struct { + name string + telemetry toml.Telemetry + expected time.Duration + }{ + {"ChipIngressSendIntervalSet", toml.Telemetry{ChipIngressSendInterval: ptrDuration(100 * time.Millisecond)}, 100 * time.Millisecond}, + {"ChipIngressSendIntervalNil", toml.Telemetry{ChipIngressSendInterval: nil}, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tc := telemetryConfig{s: tt.telemetry} + assert.Equal(t, tt.expected, tc.ChipIngressSendInterval()) + }) + } +} + +func TestTelemetryConfig_ChipIngressSendTimeout(t *testing.T) { + tests := []struct { + name string + telemetry toml.Telemetry + expected time.Duration + }{ + {"ChipIngressSendTimeoutSet", toml.Telemetry{ChipIngressSendTimeout: ptrDuration(3 * time.Second)}, 3 * time.Second}, + {"ChipIngressSendTimeoutNil", toml.Telemetry{ChipIngressSendTimeout: nil}, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tc := telemetryConfig{s: tt.telemetry} + assert.Equal(t, tt.expected, tc.ChipIngressSendTimeout()) + }) + } +} + +func TestTelemetryConfig_ChipIngressDrainTimeout(t *testing.T) { + tests := []struct { + name string + telemetry toml.Telemetry + expected time.Duration + }{ + {"ChipIngressDrainTimeoutSet", toml.Telemetry{ChipIngressDrainTimeout: ptrDuration(10 * time.Second)}, 10 * time.Second}, + {"ChipIngressDrainTimeoutNil", toml.Telemetry{ChipIngressDrainTimeout: nil}, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tc := telemetryConfig{s: tt.telemetry} + assert.Equal(t, tt.expected, tc.ChipIngressDrainTimeout()) + }) + } +} + +func TestTelemetryConfig_ChipIngressMaxGRPCRequestSize(t *testing.T) { + tests := []struct { + name string + telemetry toml.Telemetry + expected uint + }{ + {"ChipIngressMaxGRPCRequestSizeSet", toml.Telemetry{ChipIngressMaxGRPCRequestSize: ptrUint(10485760)}, 10485760}, + {"ChipIngressMaxGRPCRequestSizeNil", toml.Telemetry{ChipIngressMaxGRPCRequestSize: nil}, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tc := telemetryConfig{s: tt.telemetry} + assert.Equal(t, tt.expected, tc.ChipIngressMaxGRPCRequestSize()) + }) + } +} + func TestTelemetryConfig_HeartbeatInterval(t *testing.T) { tests := []struct { name string diff --git a/core/services/chainlink/config_test.go b/core/services/chainlink/config_test.go index 127261bc388..def1fbfbf7c 100644 --- a/core/services/chainlink/config_test.go +++ b/core/services/chainlink/config_test.go @@ -562,6 +562,13 @@ func TestConfig_Marshal(t *testing.T) { ChipIngressEndpoint: ptr("example.com/chip-ingress"), ChipIngressInsecureConnection: ptr(false), ChipIngressBatchEmitterEnabled: ptr(true), + ChipIngressBufferSize: ptr[uint](1000), + ChipIngressMaxBatchSize: ptr[uint](500), + ChipIngressMaxConcurrentSends: ptr[int](10), + ChipIngressSendInterval: commoncfg.MustNewDuration(100 * time.Millisecond), + ChipIngressSendTimeout: commoncfg.MustNewDuration(3 * time.Second), + ChipIngressDrainTimeout: commoncfg.MustNewDuration(10 * time.Second), + ChipIngressMaxGRPCRequestSize: ptr[uint](10485760), DurableEmitterEnabled: ptr(false), HeartbeatInterval: commoncfg.MustNewDuration(1 * time.Second), LogStreamingEnabled: ptr(false), diff --git a/core/services/chainlink/testdata/config-empty-effective.toml b/core/services/chainlink/testdata/config-empty-effective.toml index f6eb5121505..a0caa94ea83 100644 --- a/core/services/chainlink/testdata/config-empty-effective.toml +++ b/core/services/chainlink/testdata/config-empty-effective.toml @@ -349,6 +349,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/core/services/chainlink/testdata/config-full.toml b/core/services/chainlink/testdata/config-full.toml index 782db0949d7..e94538920f6 100644 --- a/core/services/chainlink/testdata/config-full.toml +++ b/core/services/chainlink/testdata/config-full.toml @@ -384,6 +384,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = 'example.com/chip-ingress' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/core/services/chainlink/testdata/config-multi-chain-effective.toml b/core/services/chainlink/testdata/config-multi-chain-effective.toml index 86121aaa1b8..5be7dcdaa56 100644 --- a/core/services/chainlink/testdata/config-multi-chain-effective.toml +++ b/core/services/chainlink/testdata/config-multi-chain-effective.toml @@ -349,6 +349,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/docs/CONFIG.md b/docs/CONFIG.md index db2207dee18..07fef8be3b5 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -2386,6 +2386,13 @@ AuthHeadersTTL = '0s' # Default ChipIngressEndpoint = '' # Default ChipIngressInsecureConnection = false # Default ChipIngressBatchEmitterEnabled = true # Default +ChipIngressBufferSize = 1000 # Default +ChipIngressMaxBatchSize = 500 # Default +ChipIngressMaxConcurrentSends = 10 # Default +ChipIngressSendInterval = '100ms' # Default +ChipIngressSendTimeout = '3s' # Default +ChipIngressDrainTimeout = '10s' # Default +ChipIngressMaxGRPCRequestSize = 10485760 # Default DurableEmitterEnabled = false # Default HeartbeatInterval = '1s' # Default LogLevel = "info" # Default @@ -2470,6 +2477,48 @@ ChipIngressBatchEmitterEnabled = true # Default ChipIngressBatchEmitterEnabled enables batching for chip-ingress events. When false, events are sent individually (legacy behavior). +### ChipIngressBufferSize +```toml +ChipIngressBufferSize = 1000 # Default +``` +ChipIngressBufferSize is the in-memory queue size for chip-ingress events. + +### ChipIngressMaxBatchSize +```toml +ChipIngressMaxBatchSize = 500 # Default +``` +ChipIngressMaxBatchSize is the max events per PublishBatch RPC. + +### ChipIngressMaxConcurrentSends +```toml +ChipIngressMaxConcurrentSends = 10 # Default +``` +ChipIngressMaxConcurrentSends limits parallel PublishBatch calls. + +### ChipIngressSendInterval +```toml +ChipIngressSendInterval = '100ms' # Default +``` +ChipIngressSendInterval is the max wait before flushing an incomplete batch. + +### ChipIngressSendTimeout +```toml +ChipIngressSendTimeout = '3s' # Default +``` +ChipIngressSendTimeout is the per-RPC timeout for PublishBatch. + +### ChipIngressDrainTimeout +```toml +ChipIngressDrainTimeout = '10s' # Default +``` +ChipIngressDrainTimeout is the max shutdown wait to flush queued events. + +### ChipIngressMaxGRPCRequestSize +```toml +ChipIngressMaxGRPCRequestSize = 10485760 # Default +``` +ChipIngressMaxGRPCRequestSize is the max serialized PublishBatch request size in bytes. Batches exceeding this are split before send (min 1 MiB enforced by batch client). + ### DurableEmitterEnabled ```toml DurableEmitterEnabled = false # Default diff --git a/go.mod b/go.mod index 9f99c8b05a4..d969c6ff2d0 100644 --- a/go.mod +++ b/go.mod @@ -85,9 +85,9 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260415165642-49f23e4d76cc github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260415165642-49f23e4d76cc github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260622154332-695181f87033 - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260625162847-bdf9e82b2f75 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260629213606-78f56464b8fc github.com/smartcontractkit/chainlink-common/keystore v1.2.0 - github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260601211238-9f526774fef0 + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260623170329-4577ef4ba0ae github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 diff --git a/go.sum b/go.sum index 24ed43e4254..98eeaa320c3 100644 --- a/go.sum +++ b/go.sum @@ -1162,12 +1162,12 @@ github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260 github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260415165642-49f23e4d76cc/go.mod h1:67YbnoglYD61Pz/jTVCgav9wFq7S35OU8UyQSvPllRw= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260622154332-695181f87033 h1:WjZwKtUA/0TPvzgCt8bcdq+BHMIL65S0oU79mxgZn/Y= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260622154332-695181f87033/go.mod h1:15M0qBycFN5jkNjaYFkutYkGAmhuT401IfaJvz32lcg= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260625162847-bdf9e82b2f75 h1:+wn8Kfs9ImNEWlonkTmtjDQOdEaDzwdpedab+wAzCKI= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260625162847-bdf9e82b2f75/go.mod h1:wUK7w5xRrFPD2qQfdt1fLXzQzWSb4PaZaxa4nsqCWVs= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260629213606-78f56464b8fc h1:pVVJgCQN9VeOtxZVAjnDkE3gLcX/jb9+LPP8+Tj0XqM= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260629213606-78f56464b8fc/go.mod h1:ncZiIgraMh4F9lWDoJwB1TX395qZFR5bvnZcHbD0A1Q= github.com/smartcontractkit/chainlink-common/keystore v1.2.0 h1:1BH/b14CkGjArfzznlioQpIJiynECWVT48JUP9E277U= github.com/smartcontractkit/chainlink-common/keystore v1.2.0/go.mod h1:9R/74vN+bJ5PbkOyM/pUy/AeAZaRwYb/k4XPeXcbDio= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260601211238-9f526774fef0 h1:NExKM/D0HneOq/N5LGTbkV4VOa0UHCvfTNEb4GqYpto= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260601211238-9f526774fef0/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= +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-data-streams v0.1.15-0.20260522094612-5f9f748bd87a h1:8bIqv4r7SgDWkXL2Qz/Ijw+YjZY1uroIte3E2v2keVk= github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260623170329-4577ef4ba0ae h1:VrOaSNVhElURmca9ZVVyKlSiOg9Hz372+oFJD38fYRo= diff --git a/plugins/loop_registry.go b/plugins/loop_registry.go index 3968018cdbc..5fa6625787c 100644 --- a/plugins/loop_registry.go +++ b/plugins/loop_registry.go @@ -156,6 +156,13 @@ func (m *LoopRegistry) Register(id string) (*RegisteredLoop, error) { envCfg.ChipIngressInsecureConnection = m.cfgTelemetry.ChipIngressInsecureConnection() envCfg.ChipIngressBatchEmitterEnabled = m.cfgTelemetry.ChipIngressBatchEmitterEnabled() envCfg.ChipIngressDurableEmitterEnabled = m.cfgTelemetry.DurableEmitterEnabled() + envCfg.ChipIngressBufferSize = m.cfgTelemetry.ChipIngressBufferSize() + envCfg.ChipIngressMaxBatchSize = m.cfgTelemetry.ChipIngressMaxBatchSize() + envCfg.ChipIngressMaxConcurrentSends = m.cfgTelemetry.ChipIngressMaxConcurrentSends() + envCfg.ChipIngressSendInterval = m.cfgTelemetry.ChipIngressSendInterval() + envCfg.ChipIngressSendTimeout = m.cfgTelemetry.ChipIngressSendTimeout() + envCfg.ChipIngressDrainTimeout = m.cfgTelemetry.ChipIngressDrainTimeout() + envCfg.ChipIngressMaxGRPCRequestSize = m.cfgTelemetry.ChipIngressMaxGRPCRequestSize() envCfg.TelemetryLogStreamingEnabled = m.cfgTelemetry.LogStreamingEnabled() envCfg.TelemetryLogLevel = m.cfgTelemetry.LogLevel() envCfg.TelemetryLogBatchProcessor = m.cfgTelemetry.LogBatchProcessor() diff --git a/plugins/loop_registry_test.go b/plugins/loop_registry_test.go index d762dd85c57..00aadb7d2fd 100644 --- a/plugins/loop_registry_test.go +++ b/plugins/loop_registry_test.go @@ -76,6 +76,20 @@ func (m mockCfgTelemetry) ChipIngressInsecureConnection() bool { return false } func (m mockCfgTelemetry) ChipIngressBatchEmitterEnabled() bool { return false } +func (m mockCfgTelemetry) ChipIngressBufferSize() uint { return 1000 } + +func (m mockCfgTelemetry) ChipIngressMaxBatchSize() uint { return 500 } + +func (m mockCfgTelemetry) ChipIngressMaxConcurrentSends() int { return 10 } + +func (m mockCfgTelemetry) ChipIngressSendInterval() time.Duration { return 100 * time.Millisecond } + +func (m mockCfgTelemetry) ChipIngressSendTimeout() time.Duration { return 3 * time.Second } + +func (m mockCfgTelemetry) ChipIngressDrainTimeout() time.Duration { return 10 * time.Second } + +func (m mockCfgTelemetry) ChipIngressMaxGRPCRequestSize() uint { return 10485760 } + func (m mockCfgTelemetry) HeartbeatInterval() time.Duration { return 5 * time.Second } @@ -253,4 +267,11 @@ func TestLoopRegistry_Register(t *testing.T) { require.Equal(t, "example.com/chip-ingress", envCfg.ChipIngressEndpoint) require.False(t, envCfg.ChipIngressBatchEmitterEnabled) + require.Equal(t, uint(1000), envCfg.ChipIngressBufferSize) + require.Equal(t, uint(500), envCfg.ChipIngressMaxBatchSize) + require.Equal(t, 10, envCfg.ChipIngressMaxConcurrentSends) + require.Equal(t, 100*time.Millisecond, envCfg.ChipIngressSendInterval) + require.Equal(t, 3*time.Second, envCfg.ChipIngressSendTimeout) + require.Equal(t, 10*time.Second, envCfg.ChipIngressDrainTimeout) + require.Equal(t, uint(10485760), envCfg.ChipIngressMaxGRPCRequestSize) } diff --git a/testdata/scripts/config/merge_raw_configs.txtar b/testdata/scripts/config/merge_raw_configs.txtar index 06ac519d21d..e1f9d80da48 100644 --- a/testdata/scripts/config/merge_raw_configs.txtar +++ b/testdata/scripts/config/merge_raw_configs.txtar @@ -496,6 +496,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/testdata/scripts/node/validate/default.txtar b/testdata/scripts/node/validate/default.txtar index 87476ac18f8..c404a145d5c 100644 --- a/testdata/scripts/node/validate/default.txtar +++ b/testdata/scripts/node/validate/default.txtar @@ -361,6 +361,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/testdata/scripts/node/validate/defaults-override.txtar b/testdata/scripts/node/validate/defaults-override.txtar index a0bd0f22ca8..138d26a45b7 100644 --- a/testdata/scripts/node/validate/defaults-override.txtar +++ b/testdata/scripts/node/validate/defaults-override.txtar @@ -422,6 +422,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/testdata/scripts/node/validate/disk-based-logging-disabled.txtar b/testdata/scripts/node/validate/disk-based-logging-disabled.txtar index b514320e81e..80914efa3f0 100644 --- a/testdata/scripts/node/validate/disk-based-logging-disabled.txtar +++ b/testdata/scripts/node/validate/disk-based-logging-disabled.txtar @@ -405,6 +405,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar b/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar index 09d630f0410..f330aa67646 100644 --- a/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar +++ b/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar @@ -405,6 +405,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/testdata/scripts/node/validate/disk-based-logging.txtar b/testdata/scripts/node/validate/disk-based-logging.txtar index efa55d6c4a3..65d693a043a 100644 --- a/testdata/scripts/node/validate/disk-based-logging.txtar +++ b/testdata/scripts/node/validate/disk-based-logging.txtar @@ -405,6 +405,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/testdata/scripts/node/validate/fallback-override.txtar b/testdata/scripts/node/validate/fallback-override.txtar index 4f5406cf44d..94ce28b02f2 100644 --- a/testdata/scripts/node/validate/fallback-override.txtar +++ b/testdata/scripts/node/validate/fallback-override.txtar @@ -507,6 +507,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/testdata/scripts/node/validate/invalid-ocr-p2p.txtar b/testdata/scripts/node/validate/invalid-ocr-p2p.txtar index 63ac3270c61..2159aa5084f 100644 --- a/testdata/scripts/node/validate/invalid-ocr-p2p.txtar +++ b/testdata/scripts/node/validate/invalid-ocr-p2p.txtar @@ -390,6 +390,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/testdata/scripts/node/validate/invalid.txtar b/testdata/scripts/node/validate/invalid.txtar index 2ea583aa574..6b016be7154 100644 --- a/testdata/scripts/node/validate/invalid.txtar +++ b/testdata/scripts/node/validate/invalid.txtar @@ -401,6 +401,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/testdata/scripts/node/validate/valid.txtar b/testdata/scripts/node/validate/valid.txtar index d8a68cdc8a4..56d429996ef 100644 --- a/testdata/scripts/node/validate/valid.txtar +++ b/testdata/scripts/node/validate/valid.txtar @@ -402,6 +402,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info' diff --git a/testdata/scripts/node/validate/warnings.txtar b/testdata/scripts/node/validate/warnings.txtar index cc7c437f38a..73a40a85fa4 100644 --- a/testdata/scripts/node/validate/warnings.txtar +++ b/testdata/scripts/node/validate/warnings.txtar @@ -384,6 +384,13 @@ AuthHeadersTTL = '0s' ChipIngressEndpoint = '' ChipIngressInsecureConnection = false ChipIngressBatchEmitterEnabled = true +ChipIngressBufferSize = 1000 +ChipIngressMaxBatchSize = 500 +ChipIngressMaxConcurrentSends = 10 +ChipIngressSendInterval = '100ms' +ChipIngressSendTimeout = '3s' +ChipIngressDrainTimeout = '10s' +ChipIngressMaxGRPCRequestSize = 10485760 DurableEmitterEnabled = false HeartbeatInterval = '1s' LogLevel = 'info'