diff --git a/core/cmd/shell_local_test.go b/core/cmd/shell_local_test.go index 1682c8dbe16..6ad93300679 100644 --- a/core/cmd/shell_local_test.go +++ b/core/cmd/shell_local_test.go @@ -75,7 +75,7 @@ func genTestEVMRelayers(t *testing.T, cfg chainlink.GeneralConfig, ds sqlutil.Da f := chainlink.RelayerFactory{ Logger: lggr, LoopRegistry: plugins.NewLoopRegistry(lggr, cfg.AppID().String(), cfg.Feature().LogPoller(), cfg.Database(), - cfg.Mercury(), cfg.Pyroscope(), cfg.AutoPprof(), cfg.Tracing(), cfg.Telemetry(), nil, "", cfg.LOOPP()), + cfg.Mercury(), cfg.Pyroscope(), cfg.AutoPprof(), cfg.Tracing(), cfg.Telemetry(), cfg.Metering(), nil, "", cfg.LOOPP()), CapabilitiesRegistry: capabilities.NewRegistry(lggr), } diff --git a/core/config/app_config.go b/core/config/app_config.go index 7965d096c73..9ace57f66d1 100644 --- a/core/config/app_config.go +++ b/core/config/app_config.go @@ -61,6 +61,7 @@ type AppConfig interface { WebServer() WebServer Tracing() Tracing Telemetry() Telemetry + Metering() Metering CRE() CRE CCV() CCV Billing() Billing diff --git a/core/config/docs/core.toml b/core/config/docs/core.toml index 8a285a216e6..f4e389b92b9 100644 --- a/core/config/docs/core.toml +++ b/core/config/docs/core.toml @@ -922,6 +922,28 @@ Enabled = false # Default # By default, we only forward the go runtime metrics. Empty means forward everything. Prefixes = ["go_"] # Default +# Metering configures durable resource metering emission and the coarse +# deployment/node identity dimensions stamped on emitted MeterRecords and +# MeterSnapshots. These are plumbed to LOOP plugins via env. +[Metering] +# MeterRecordsEnabled enables durable MeterRecord emission for LOOP plugins. +MeterRecordsEnabled = false # Default +# MeterSnapshotsEnabled enables durable MeterSnapshot emission for LOOP plugins. +# Requires MeterRecordsEnabled = true. +MeterSnapshotsEnabled = false # Default +# Product is the deployment product identity dimension, e.g. 'cre'. +Product = '' # Default +# Tenant is the human-readable tenant name, e.g. 'mainline'. +Tenant = '' # Default +# NumericTenantID is the numbered tenant identifier represented as a string. +NumericTenantID = '' # Default +# Environment is the deployment environment identity dimension, e.g. 'production'. +Environment = '' # Default +# Zone is the deployment zone identity dimension, e.g. 'wf-zone-a'. +Zone = '' # Default +# NodeID is the node's logical name, e.g. 'clp-cre-wf-zone-a-1' (not the CSA public key). +NodeID = '' # Default + [CRE.Streams] # WsURL is the websockets url for the streams sdk config WsURL = "streams.url" # Example diff --git a/core/config/env/env.go b/core/config/env/env.go index dc5b4951fbc..eed36c80152 100644 --- a/core/config/env/env.go +++ b/core/config/env/env.go @@ -14,6 +14,10 @@ var ( DatabaseAllowSimplePasswords = Var("CL_DATABASE_ALLOW_SIMPLE_PASSWORDS") IgnorePrereleaseVersionCheck = Var("CL_IGNORE_PRE_RELEASE_VERSION_CHECK") SkipAppVersionCheck = Var("CL_SKIP_APP_VERSION_CHECK") + // MeterRecordsEnabled gates emission of metering.v1.MeterRecord events for + // durable CRE resources. Accepts strconv.ParseBool values; default false. + // Temporary deploy gate; promotion to TOML config is tracked by SHARED-2718. + MeterRecordsEnabled = Var("CL_METER_RECORDS_ENABLED") DatabaseURL = Secret("CL_DATABASE_URL") DatabaseBackupURL = Secret("CL_DATABASE_BACKUP_URL") diff --git a/core/config/metering_config.go b/core/config/metering_config.go new file mode 100644 index 00000000000..9f11e1c3419 --- /dev/null +++ b/core/config/metering_config.go @@ -0,0 +1,16 @@ +package config + +// Metering exposes durable resource-metering configuration: the emission +// toggles and the coarse deployment/node identity dimensions stamped on emitted +// MeterRecords and MeterSnapshots. These are passed via loop.EnvConfig to every LOOP +// plugin. +type Metering interface { + MeterRecordsEnabled() bool + MeterSnapshotsEnabled() bool + Product() string + Tenant() string + NumericTenantID() string + Environment() string + Zone() string + NodeID() string +} diff --git a/core/config/toml/types.go b/core/config/toml/types.go index 6b3bb050f1e..6111804b62f 100644 --- a/core/config/toml/types.go +++ b/core/config/toml/types.go @@ -65,6 +65,7 @@ type Core struct { Mercury Mercury `toml:",omitempty"` Capabilities Capabilities `toml:",omitempty"` Telemetry Telemetry `toml:",omitempty"` + Metering Metering `toml:",omitempty"` Workflows Workflows `toml:",omitempty"` CRE CreConfig `toml:",omitempty"` Billing Billing `toml:",omitempty"` @@ -113,6 +114,7 @@ func (c *Core) SetFrom(f *Core) { c.JobDistributor.setFrom(&f.JobDistributor) c.Tracing.setFrom(&f.Tracing) c.Telemetry.setFrom(&f.Telemetry) + c.Metering.setFrom(&f.Metering) c.CRE.setFrom(&f.CRE) c.Billing.setFrom(&f.Billing) c.BridgeStatusReporter.setFrom(&f.BridgeStatusReporter) @@ -3084,6 +3086,70 @@ func (b *Telemetry) ValidateConfig() (err error) { return err } +// Metering configures durable resource metering emission and the coarse +// deployment/node identity dimensions stamped on emitted MeterRecords and +// MeterSnapshots. These are passed via loop.EnvConfig to every LOOP plugin. +type Metering struct { + // MeterRecordsEnabled enables durable MeterRecord emission for LOOP plugins. + MeterRecordsEnabled *bool + // MeterSnapshotsEnabled enables durable MeterSnapshot emission. Requires + // MeterRecordsEnabled to be true. + MeterSnapshotsEnabled *bool + // Product is the deployment product identity dimension, e.g. "cre". + Product *string + // Tenant is the human-readable tenant name, e.g. "mainline". + Tenant *string + // NumericTenantID is the numbered tenant identifier as a string. + NumericTenantID *string + // Environment is the deployment environment dimension, e.g. "production". + Environment *string + // Zone is the deployment zone dimension, e.g. "wf-zone-a". + Zone *string + // NodeID is the node's logical name, e.g. "clp-cre-wf-zone-a-1" (NOT the CSA + // public key). The billing service uses it to look up the node's CSA key in + // the workflow registry; the CSA key itself is attached to events as the + // node_csa_key attribute. + NodeID *string +} + +func (b *Metering) setFrom(f *Metering) { + if v := f.MeterRecordsEnabled; v != nil { + b.MeterRecordsEnabled = v + } + if v := f.MeterSnapshotsEnabled; v != nil { + b.MeterSnapshotsEnabled = v + } + if v := f.Product; v != nil { + b.Product = v + } + if v := f.Tenant; v != nil { + b.Tenant = v + } + if v := f.NumericTenantID; v != nil { + b.NumericTenantID = v + } + if v := f.Environment; v != nil { + b.Environment = v + } + if v := f.Zone; v != nil { + b.Zone = v + } + if v := f.NodeID; v != nil { + b.NodeID = v + } +} + +func (b *Metering) ValidateConfig() (err error) { + if b.MeterSnapshotsEnabled != nil && *b.MeterSnapshotsEnabled && (b.MeterRecordsEnabled == nil || !*b.MeterRecordsEnabled) { + err = errors.Join(err, configutils.ErrInvalid{ + Name: "MeterSnapshotsEnabled", + Value: true, + Msg: "requires MeterRecordsEnabled to be true", + }) + } + return err +} + type PrometheusBridge struct { Enabled *bool Prefixes []string diff --git a/core/scripts/go.mod b/core/scripts/go.mod index 2760670ca01..7352bd2ca4b 100644 --- a/core/scripts/go.mod +++ b/core/scripts/go.mod @@ -47,7 +47,7 @@ require ( github.com/smartcontractkit/chain-selectors v1.0.104 github.com/smartcontractkit/chainlink-automation v0.8.1 github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260624154507-ea7ff77a0ddb - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 github.com/smartcontractkit/chainlink-common/keystore v1.2.0 github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 @@ -507,6 +507,7 @@ require ( github.com/smartcontractkit/chainlink-protos/chainlink-ccv/verifier v0.0.0-20251211142334-5c3421fe2c8d // indirect github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36 // indirect github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 // indirect + github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 // indirect github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 // indirect github.com/smartcontractkit/chainlink-protos/orchestrator v0.11.0 // indirect github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd // indirect diff --git a/core/scripts/go.sum b/core/scripts/go.sum index cdbdafec702..afa95889506 100644 --- a/core/scripts/go.sum +++ b/core/scripts/go.sum @@ -1580,8 +1580,8 @@ github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260 github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb/go.mod h1:67YbnoglYD61Pz/jTVCgav9wFq7S35OU8UyQSvPllRw= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094 h1:uMn1w/n95p+pSxK6hYNqji/sDaab8D0Cxuph9qUkM2g= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094/go.mod h1:3XQmy2zQHrmufebP7dM+x595+gghxzyIKxK9wv91Rh8= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 h1:ywWOEZYL4DhVwf/TQ5jXsGlx6CzERsYbqd6ov3OS1sc= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0/go.mod h1:ncZiIgraMh4F9lWDoJwB1TX395qZFR5bvnZcHbD0A1Q= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 h1:uvUnQlWgJGJBgtXpxpgGTipUZPxp/JugxBK7KUe41DQ= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6/go.mod h1:2hczeHjBapPah8d3EOoRAfzI4gH3caZToMhAUG8WS3I= 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.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1626,6 +1626,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0 h1:FC+WdJ8Y github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0/go.mod h1:2ahgl5bI9+fMCD+dBC785Lak38Tb7ApdTe5I8a09Qp0= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 h1:NJdGFhzT6zMaTod4QkBqVD2sg0I25iw1boOYtTpEwRo= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305/go.mod h1:qSTSwX3cBP3FKQwQacdjArqv0g6QnukjV4XuzO6UyoY= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 h1:B355/rV/lwpZl3C5iVsFQZU7+LeA+5BTTIzTDYlDOrA= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 h1:bnSl5p3mFekSJ6QcbZ1TmHn2ffYiX8xk6hNzVmyhstQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= github.com/smartcontractkit/chainlink-protos/op-catalog v0.1.0 h1:hGEJFD2X3oNIPXQbtIPxCJyg5CcKglRCYBmESS+gmeQ= diff --git a/core/services/chainlink/application.go b/core/services/chainlink/application.go index 74168bdd70e..fe5da2b1893 100644 --- a/core/services/chainlink/application.go +++ b/core/services/chainlink/application.go @@ -287,7 +287,7 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err } loopRegistry := plugins.NewLoopRegistry(globalLogger, cfg.AppID().String(), cfg.Feature().LogPoller(), cfg.Database(), cfg.Mercury(), cfg.Pyroscope(), cfg.AutoPprof(), cfg.Tracing(), cfg.Telemetry(), - beholderAuthHeaders, csaPubKeyHex, cfg.LOOPP()) + cfg.Metering(), beholderAuthHeaders, csaPubKeyHex, cfg.LOOPP()) relayerFactory := RelayerFactory{ Logger: opts.Logger, @@ -676,6 +676,16 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err atomicSettings, creServices.OCRConfigService, cfg.Capabilities().Local(), + // Host-injected deployment/node metering identity for spawned capability + // LOOPs. Sourced once here from node config: product constant, env/zone + // from [Telemetry.ResourceAttributes], node_id = the same CSA pubkey hex + // the node uses for beholder auth and the engine's node_id. + standardcapabilities.NodeIdentity{ + Product: "cre", + Environment: cfg.Telemetry().ResourceAttributes()["env"], + Zone: cfg.Telemetry().ResourceAttributes()["zone"], + NodeID: csaPubKeyHex, + }, ) delegates[job.StandardCapabilities] = stdcapDelegate if creServices.SetDelegatesDeps != nil { diff --git a/core/services/chainlink/config_general.go b/core/services/chainlink/config_general.go index fd75e41e5bb..dcc99287d46 100644 --- a/core/services/chainlink/config_general.go +++ b/core/services/chainlink/config_general.go @@ -588,6 +588,10 @@ func (g *generalConfig) Telemetry() coreconfig.Telemetry { return &telemetryConfig{s: g.c.Telemetry} } +func (g *generalConfig) Metering() coreconfig.Metering { + return &meteringConfig{s: g.c.Metering} +} + func (g *generalConfig) CRE() coreconfig.CRE { return &creConfig{s: g.secrets.CRE, c: g.c.CRE} } diff --git a/core/services/chainlink/config_metering.go b/core/services/chainlink/config_metering.go new file mode 100644 index 00000000000..fa0c8f36843 --- /dev/null +++ b/core/services/chainlink/config_metering.go @@ -0,0 +1,65 @@ +package chainlink + +import ( + "github.com/smartcontractkit/chainlink/v2/core/config/toml" +) + +type meteringConfig struct { + s toml.Metering +} + +func (b *meteringConfig) MeterRecordsEnabled() bool { + if b.s.MeterRecordsEnabled == nil { + return false + } + return *b.s.MeterRecordsEnabled +} + +func (b *meteringConfig) MeterSnapshotsEnabled() bool { + if b.s.MeterSnapshotsEnabled == nil { + return false + } + return *b.s.MeterSnapshotsEnabled +} + +func (b *meteringConfig) Product() string { + if b.s.Product == nil { + return "" + } + return *b.s.Product +} + +func (b *meteringConfig) Tenant() string { + if b.s.Tenant == nil { + return "" + } + return *b.s.Tenant +} + +func (b *meteringConfig) NumericTenantID() string { + if b.s.NumericTenantID == nil { + return "" + } + return *b.s.NumericTenantID +} + +func (b *meteringConfig) Environment() string { + if b.s.Environment == nil { + return "" + } + return *b.s.Environment +} + +func (b *meteringConfig) Zone() string { + if b.s.Zone == nil { + return "" + } + return *b.s.Zone +} + +func (b *meteringConfig) NodeID() string { + if b.s.NodeID == nil { + return "" + } + return *b.s.NodeID +} diff --git a/core/services/chainlink/config_metering_test.go b/core/services/chainlink/config_metering_test.go new file mode 100644 index 00000000000..56d61af1925 --- /dev/null +++ b/core/services/chainlink/config_metering_test.go @@ -0,0 +1,44 @@ +package chainlink + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/smartcontractkit/chainlink/v2/core/config/toml" +) + +func TestMeteringConfig(t *testing.T) { + t.Run("defaults", func(t *testing.T) { + mc := meteringConfig{s: toml.Metering{}} + assert.False(t, mc.MeterRecordsEnabled()) + assert.False(t, mc.MeterSnapshotsEnabled()) + assert.Empty(t, mc.Product()) + assert.Empty(t, mc.Tenant()) + assert.Empty(t, mc.NumericTenantID()) + assert.Empty(t, mc.Environment()) + assert.Empty(t, mc.Zone()) + assert.Empty(t, mc.NodeID()) + }) + + t.Run("explicit values", func(t *testing.T) { + mc := meteringConfig{s: toml.Metering{ + MeterRecordsEnabled: ptr(true), + MeterSnapshotsEnabled: ptr(true), + Product: ptr("cre"), + Tenant: ptr("mainline"), + NumericTenantID: ptr("42"), + Environment: ptr("production"), + Zone: ptr("wf-zone-a"), + NodeID: ptr("csa-pubkey-1"), + }} + assert.True(t, mc.MeterRecordsEnabled()) + assert.True(t, mc.MeterSnapshotsEnabled()) + assert.Equal(t, "cre", mc.Product()) + assert.Equal(t, "mainline", mc.Tenant()) + assert.Equal(t, "42", mc.NumericTenantID()) + assert.Equal(t, "production", mc.Environment()) + assert.Equal(t, "wf-zone-a", mc.Zone()) + assert.Equal(t, "csa-pubkey-1", mc.NodeID()) + }) +} diff --git a/core/services/chainlink/config_test.go b/core/services/chainlink/config_test.go index 27a27e70cc7..a5e654065e9 100644 --- a/core/services/chainlink/config_test.go +++ b/core/services/chainlink/config_test.go @@ -578,6 +578,16 @@ func TestConfig_Marshal(t *testing.T) { Prefixes: []string{"ocr_"}, }, } + full.Metering = toml.Metering{ + MeterRecordsEnabled: ptr(true), + MeterSnapshotsEnabled: ptr(true), + Product: ptr("cre"), + Tenant: ptr("mainline"), + NumericTenantID: ptr("42"), + Environment: ptr("production"), + Zone: ptr("wf-zone-a"), + NodeID: ptr("csa-pubkey-1"), + } full.CRE = toml.CreConfig{ UseLocalTimeProvider: ptr(true), EnableDKGRecipient: ptr(false), diff --git a/core/services/chainlink/mocks/general_config.go b/core/services/chainlink/mocks/general_config.go index 7bb1c4a4a1d..13228848409 100644 --- a/core/services/chainlink/mocks/general_config.go +++ b/core/services/chainlink/mocks/general_config.go @@ -2557,6 +2557,53 @@ func (_c *GeneralConfig_Telemetry_Call) RunAndReturn(run func() config.Telemetry return _c } +// Metering provides a mock function with no fields +func (_m *GeneralConfig) Metering() config.Metering { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for Metering") + } + + var r0 config.Metering + if rf, ok := ret.Get(0).(func() config.Metering); ok { + r0 = rf() + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(config.Metering) + } + } + + return r0 +} + +// GeneralConfig_Metering_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Metering' +type GeneralConfig_Metering_Call struct { + *mock.Call +} + +// Metering is a helper method to define mock.On call +func (_e *GeneralConfig_Expecter) Metering() *GeneralConfig_Metering_Call { + return &GeneralConfig_Metering_Call{Call: _e.mock.On("Metering")} +} + +func (_c *GeneralConfig_Metering_Call) Run(run func()) *GeneralConfig_Metering_Call { + _c.Call.Run(func(args mock.Arguments) { + run() + }) + return _c +} + +func (_c *GeneralConfig_Metering_Call) Return(_a0 config.Metering) *GeneralConfig_Metering_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *GeneralConfig_Metering_Call) RunAndReturn(run func() config.Metering) *GeneralConfig_Metering_Call { + _c.Call.Return(run) + return _c +} + // TelemetryIngress provides a mock function with no fields func (_m *GeneralConfig) TelemetryIngress() config.TelemetryIngress { ret := _m.Called() diff --git a/core/services/chainlink/testdata/config-empty-effective.toml b/core/services/chainlink/testdata/config-empty-effective.toml index 6aeec786180..cb9ed7589f7 100644 --- a/core/services/chainlink/testdata/config-empty-effective.toml +++ b/core/services/chainlink/testdata/config-empty-effective.toml @@ -364,6 +364,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Tenant = '' +NumericTenantID = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/core/services/chainlink/testdata/config-full.toml b/core/services/chainlink/testdata/config-full.toml index a2b99dafcbd..907bb21bc98 100644 --- a/core/services/chainlink/testdata/config-full.toml +++ b/core/services/chainlink/testdata/config-full.toml @@ -403,6 +403,16 @@ Foo = 'bar' Enabled = true Prefixes = ['ocr_'] +[Metering] +MeterRecordsEnabled = true +MeterSnapshotsEnabled = true +Product = 'cre' +Tenant = 'mainline' +NumericTenantID = '42' +Environment = 'production' +Zone = 'wf-zone-a' +NodeID = 'csa-pubkey-1' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/core/services/chainlink/testdata/config-multi-chain-effective.toml b/core/services/chainlink/testdata/config-multi-chain-effective.toml index ec7cb65fbe0..d42d768567e 100644 --- a/core/services/chainlink/testdata/config-multi-chain-effective.toml +++ b/core/services/chainlink/testdata/config-multi-chain-effective.toml @@ -364,6 +364,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Tenant = '' +NumericTenantID = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/core/services/cre/confidential_relay_peerid_test.go b/core/services/cre/confidential_relay_peerid_test.go index 8b92cfeaeec..47692762125 100644 --- a/core/services/cre/confidential_relay_peerid_test.go +++ b/core/services/cre/confidential_relay_peerid_test.go @@ -51,6 +51,8 @@ func (s stubConfig) Workflows() config.Workflows { return nil } func (s stubConfig) CRE() config.CRE { return nil } func (s stubConfig) P2P() config.P2P { return s.p2p } func (s stubConfig) Sharding() config.Sharding { return nil } +func (s stubConfig) Telemetry() config.Telemetry { return nil } +func (s stubConfig) Metering() config.Metering { return nil } func peerIDFromByte(b byte) p2pkey.PeerID { var id p2pkey.PeerID diff --git a/core/services/cre/cre.go b/core/services/cre/cre.go index 8a00ee1af6a..e3da8377357 100644 --- a/core/services/cre/cre.go +++ b/core/services/cre/cre.go @@ -18,11 +18,13 @@ import ( "github.com/smartcontractkit/chainlink-common/keystore/corekeys/p2pkey" "github.com/smartcontractkit/chainlink-common/keystore/corekeys/workflowkey" + "github.com/smartcontractkit/chainlink-common/pkg/beholder" "github.com/smartcontractkit/chainlink-common/pkg/billing" "github.com/smartcontractkit/chainlink-common/pkg/custmsg" "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/loop" nodeauthjwt "github.com/smartcontractkit/chainlink-common/pkg/nodeauth/jwt" + "github.com/smartcontractkit/chainlink-common/pkg/resourcemanager" commonsrv "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink-common/pkg/services/orgresolver" "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" @@ -43,6 +45,7 @@ import ( remotetypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" capStreams "github.com/smartcontractkit/chainlink/v2/core/capabilities/streams" "github.com/smartcontractkit/chainlink/v2/core/config" + "github.com/smartcontractkit/chainlink/v2/core/config/env" "github.com/smartcontractkit/chainlink/v2/core/services/job" "github.com/smartcontractkit/chainlink/v2/core/services/keystore" "github.com/smartcontractkit/chainlink/v2/core/services/ocr/capregconfig" @@ -297,6 +300,10 @@ func (s *Services) newSubservices( return srvs, nil } + // Build the syncer's base metering identity once from node metering config; + // the handler resolves the workflow DON id later from the don notifier. + meterIdentity := newSyncerMeterIdentity(cfg) + wfSyncer, billingClient, wfSyncerSrvcs, err := newWorkflowRegistrySyncer( cfg, relayerChainInterops, @@ -310,6 +317,7 @@ func (s *Services) newSubservices( opts.LimitsFactory, s.OrgResolver, s.GatewayConnectorWrapper, + meterIdentity, ) if err != nil { return nil, err @@ -338,6 +346,8 @@ type Config interface { CRE() config.CRE P2P() config.P2P Sharding() config.Sharding + Telemetry() config.Telemetry + Metering() config.Metering } // RelayerChainInterops is the minimal interface needed for relayer chain interops @@ -718,6 +728,48 @@ func newBillingClient(lggr logger.Logger, cfg Config, opts Opts) (metering.Billi return billing.NewWorkflowClient(lggr, cfg.Billing().URL(), workflowOpts...) } +// newSyncerMeterIdentity builds the syncer's base metering identity from node +// metering config. Product/tenant/environment/zone and logical node_id are read +// from [Metering]. The workflow DON id (don_id) is resolved later by the +// handler from the don notifier (the engine runs on the workflow DON), so it is +// intentionally left empty here. Service/resource_pool are stamped by +// syncer.WithIdentity. +func newSyncerMeterIdentity(cfg Config) resourcemanager.ResourceIdentity { + m := cfg.Metering() + if m == nil { + return resourcemanager.ResourceIdentity{} + } + id := resourcemanager.ResourceIdentity{ + Product: m.Product(), + Tenant: m.Tenant(), + NumericTenantID: m.NumericTenantID(), + Environment: m.Environment(), + Zone: m.Zone(), + } + if nodeID := m.NodeID(); nodeID != "" { + id.Don = &resourcemanager.DonIdentity{NodeID: nodeID} + } + return id +} + +// meterRecordsEnabled reads the CL_METER_RECORDS_ENABLED env var gating emission of +// metering.v1.MeterRecord events. Unset, empty, or invalid values disable emission. +// Promotion of this gate to TOML config is tracked by SHARED-2718. +func meterRecordsEnabled(lggr logger.Logger) bool { + v := env.MeterRecordsEnabled.Get() + if v == "" { + return false + } + enabled, err := strconv.ParseBool(v) + if err != nil { + // Warn, not error, matching the capability producers: a bad gate value + // disables emission but must never disturb the node. + lggr.Warnw("Invalid CL_METER_RECORDS_ENABLED value; meter record emission disabled", "value", v, "err", err) + return false + } + return enabled +} + func newShardOrchestratorClient(cfg Config, lggr logger.Logger) (*shardorchestrator.Client, error) { shardID := cfg.Sharding().ShardIndex() if shardID == 0 { @@ -933,6 +985,7 @@ func newWorkflowRegistrySyncerV2( lf limits.Factory, orgResolver orgresolver.OrgResolver, gatewayConnectorWrapper *gatewayconnector.ServiceWrapper, + meterIdentity resourcemanager.ResourceIdentity, ) (syncerV2.WorkflowRegistrySyncer, []commonsrv.Service, error) { capCfg := cfg.Capabilities() wfReg := capCfg.WorkflowRegistry() @@ -1008,6 +1061,7 @@ func newWorkflowRegistrySyncerV2( } shardRoutingSteady = shardownership.NewSteadySignal(shardownership.WithSteadySignalMetrics(steadyMetrics)) } + meterRecords := meterRecordsEnabled(lggr) handlerOpts := []syncerV2.EventHandlerOption{ syncerV2.WithBillingClient(billingClient), @@ -1017,6 +1071,17 @@ func newWorkflowRegistrySyncerV2( syncerV2.WithLocalSecretOverrides(lggr, cfg.CRE().LocalSecretOverrides()), syncerV2.WithShardExecutionGuard(shardOrchestratorClient, shardingEnabled, shardIndex), syncerV2.WithShardRoutingSteady(shardRoutingSteady), + // The handler owns this ResourceManager's lifecycle (starts it, registers + // itself as the snapshotted Meterable, closes it). A positive + // SnapshotInterval enables the periodic absolute-state snapshot loop; the + // RM is otherwise a no-op when metering is disabled. + syncerV2.WithResourceManager(resourcemanager.NewResourceManager(lggr, resourcemanager.ResourceManagerConfig{ + MeterRecordsEnabled: meterRecords, + MeterSnapshotsEnabled: meterRecords, + Emitter: beholder.GetEmitter(), + SnapshotInterval: resourcemanager.DefaultSnapshotInterval, + })), + syncerV2.WithIdentity(meterIdentity), } mc := capCfg.WorkflowRegistry().ModuleCache() @@ -1161,6 +1226,7 @@ func newWorkflowRegistrySyncer( lf limits.Factory, orgResolver orgresolver.OrgResolver, gatewayConnectorWrapper *gatewayconnector.ServiceWrapper, + meterIdentity resourcemanager.ResourceIdentity, ) (syncerV2.WorkflowRegistrySyncer, metering.BillingClient, []commonsrv.Service, error) { capCfg := cfg.Capabilities() @@ -1209,6 +1275,7 @@ func newWorkflowRegistrySyncer( lf, orgResolver, gatewayConnectorWrapper, + meterIdentity, ) return syncer, billingClient, srvcs, err default: diff --git a/core/services/standardcapabilities/delegate.go b/core/services/standardcapabilities/delegate.go index 63a37c1bcef..7282ff6e1cf 100644 --- a/core/services/standardcapabilities/delegate.go +++ b/core/services/standardcapabilities/delegate.go @@ -46,6 +46,20 @@ type RelayGetter interface { GetIDToRelayerMap() map[types.RelayID]loop.Relayer } +// NodeIdentity is the host-injected deployment/node metering identity sourced at +// node startup. It is retained on the delegate for compatibility with launcher +// wiring while metering identity delivery is handled through loop.EnvConfig. +type NodeIdentity struct { + // Product is the deployment product, e.g. "cre". + Product string + // Environment is the deployment environment, from [Telemetry.ResourceAttributes]["env"]. + Environment string + // Zone is the deployment zone, from [Telemetry.ResourceAttributes]["zone"]. + Zone string + // NodeID is the node's logical name (not the CSA public key). + NodeID string +} + type Delegate struct { logger logger.Logger ds sqlutil.DataSource @@ -66,6 +80,7 @@ type Delegate struct { creSettings core.SettingsBroadcaster ocrConfigService capregconfig.OCRConfigService localCfg coreconfig.LocalCapabilities + nodeIdentity NodeIdentity initErr error isNewlyCreatedJob bool @@ -98,6 +113,7 @@ func NewDelegate( creSettings core.SettingsBroadcaster, ocrConfigService capregconfig.OCRConfigService, localCfg coreconfig.LocalCapabilities, + nodeIdentity NodeIdentity, opts ...func(*gateway.RoundRobinSelector), ) *Delegate { initErr := registerOptionalMockStreamsTrigger(logger, localCfg, registry) @@ -125,6 +141,7 @@ func NewDelegate( creSettings: creSettings, ocrConfigService: ocrConfigService, localCfg: localCfg, + nodeIdentity: nodeIdentity, initErr: initErr, selectorOpts: opts, } diff --git a/core/services/standardcapabilities/standard_capabilities.go b/core/services/standardcapabilities/standard_capabilities.go index 3d7e7f8032c..01ea074dc44 100644 --- a/core/services/standardcapabilities/standard_capabilities.go +++ b/core/services/standardcapabilities/standard_capabilities.go @@ -82,12 +82,36 @@ func NewStandardCapabilities( } } +// initialiseDependencies builds the StandardCapabilitiesDependencies delivered to +// the capability LOOP via Initialise. +func (s *StandardCapabilities) initialiseDependencies() core.StandardCapabilitiesDependencies { + return core.StandardCapabilitiesDependencies{ + Config: s.config, + Store: s.store, + CapabilityRegistry: s.CapabilitiesRegistry, + RelayerSet: s.relayerSet, + OracleFactory: s.oracleFactory, + GatewayConnector: s.gatewayConnector, + P2PKeystore: s.keystore, + OrgResolver: s.orgResolver, + CRESettings: s.creSettings, + TriggerEventStore: s.triggerEventStore, + CapabilityDonID: s.capabilityDonID, + } +} + func (s *StandardCapabilities) Start(ctx context.Context) error { return s.StartOnce("StandardCapabilities", func() error { envVars, err := plugins.ParseEnvFile(env.CapabilitiesPlugin.Env.Get()) if err != nil { return fmt.Errorf("failed to parse capabilities env file: %w", err) } + // Pass through the node's meter-record emission gate so capability LOOPPs + // inherit it: plugins.NewCmdFactory builds the child env from CmdConfig.Env + // only and does not inherit os.Environ(). + if v := env.MeterRecordsEnabled.Get(); v != "" { + envVars = append(envVars, fmt.Sprintf("%s=%s", env.MeterRecordsEnabled, v)) + } cmdFn, opts, err := s.pluginRegistrar.RegisterLOOP(plugins.CmdConfig{ ID: s.log.Name(), Cmd: s.command, @@ -119,19 +143,7 @@ func (s *StandardCapabilities) Start(ctx context.Context) error { return } - dependencies := core.StandardCapabilitiesDependencies{ - Config: s.config, - Store: s.store, - CapabilityRegistry: s.CapabilitiesRegistry, - RelayerSet: s.relayerSet, - OracleFactory: s.oracleFactory, - GatewayConnector: s.gatewayConnector, - P2PKeystore: s.keystore, - OrgResolver: s.orgResolver, - CRESettings: s.creSettings, - TriggerEventStore: s.triggerEventStore, - CapabilityDonID: s.capabilityDonID, - } + dependencies := s.initialiseDependencies() if err = s.capabilitiesLoop.Service.Initialise(cctx, dependencies); err != nil { s.log.Errorf("error initialising standard capabilities service: %v", err) return diff --git a/core/services/standardcapabilities/standard_capabilities_test.go b/core/services/standardcapabilities/standard_capabilities_test.go index ce0159b54f7..3884c702d3c 100644 --- a/core/services/standardcapabilities/standard_capabilities_test.go +++ b/core/services/standardcapabilities/standard_capabilities_test.go @@ -74,6 +74,7 @@ func TestStandardCapabilities_ForwardsPluginEnvFile(t *testing.T) { t.Run("env file unset results in empty CmdConfig.Env", func(t *testing.T) { t.Setenv(string(env.CapabilitiesPlugin.Env), "") + t.Setenv(string(env.MeterRecordsEnabled), "") cfg, err := startAndCapture(t) require.Error(t, err, "expected synthetic Start failure from capturingRegistrar") @@ -83,6 +84,33 @@ func TestStandardCapabilities_ForwardsPluginEnvFile(t *testing.T) { "no operator-provided env vars should be forwarded when CL_CAPABILITIES_ENV is unset") }) + t.Run("CL_METER_RECORDS_ENABLED set on the node is passed through to the LOOPP", func(t *testing.T) { + envFile := writeEnvFile(t, "FOO=bar\n") + t.Setenv(string(env.CapabilitiesPlugin.Env), envFile) + t.Setenv(string(env.MeterRecordsEnabled), "true") + + cfg, err := startAndCapture(t) + require.Error(t, err, "expected synthetic Start failure from capturingRegistrar") + require.Contains(t, err.Error(), capturingRegistrarErr) + + require.Contains(t, cfg.Env, "CL_METER_RECORDS_ENABLED=true", + "the node's meter-record emission gate must reach capability LOOPPs, which do not inherit os.Environ()") + require.Contains(t, cfg.Env, "FOO=bar", + "the pass-through must not displace entries from the operator-supplied env file") + }) + + t.Run("CL_METER_RECORDS_ENABLED unset is not forwarded", func(t *testing.T) { + t.Setenv(string(env.CapabilitiesPlugin.Env), "") + t.Setenv(string(env.MeterRecordsEnabled), "") + + cfg, err := startAndCapture(t) + require.Error(t, err, "expected synthetic Start failure from capturingRegistrar") + require.Contains(t, err.Error(), capturingRegistrarErr) + + require.Empty(t, cfg.Env, + "no CL_METER_RECORDS_ENABLED entry should be forwarded when the gate is unset on the node") + }) + t.Run("missing env file fails Start before RegisterLOOP", func(t *testing.T) { missingPath := filepath.Join(t.TempDir(), "does-not-exist.env") t.Setenv(string(env.CapabilitiesPlugin.Env), missingPath) @@ -106,6 +134,24 @@ func TestStandardCapabilities_ForwardsPluginEnvFile(t *testing.T) { }) } +func TestStandardCapabilities_InitialiseDependenciesRoundTrip(t *testing.T) { + want := core.StandardCapabilitiesDependencies{ + Config: "test-config", + } + + std := NewStandardCapabilities( + logger.TestLogger(t), + "not/found/path/to/binary", + want.Config, + &capturingRegistrar{}, + want, + ) + + got := std.initialiseDependencies() + + require.Equal(t, want.Config, got.Config, "config should be re-delivered to LOOP Initialise dependencies") +} + const capturingRegistrarErr = "capturingRegistrar: stop after capture" // capturingRegistrar records the CmdConfig handed to RegisterLOOP and then diff --git a/core/services/workflows/syncer/v2/engine_registry.go b/core/services/workflows/syncer/v2/engine_registry.go index afd712afe2a..6dd9adfc428 100644 --- a/core/services/workflows/syncer/v2/engine_registry.go +++ b/core/services/workflows/syncer/v2/engine_registry.go @@ -18,7 +18,7 @@ type ServiceWithMetadata struct { services.Service } -// engineEntry holds the engine and its associated source for internal storage +// engineEntry holds the engine and its associated source for internal storage. type engineEntry struct { engine services.Service source string diff --git a/core/services/workflows/syncer/v2/handler.go b/core/services/workflows/syncer/v2/handler.go index 55284f71ed0..ea7e27d8d89 100644 --- a/core/services/workflows/syncer/v2/handler.go +++ b/core/services/workflows/syncer/v2/handler.go @@ -7,7 +7,9 @@ import ( "fmt" "io" "maps" + "strconv" "sync" + "sync/atomic" "time" "go.opentelemetry.io/otel" @@ -19,6 +21,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/contexts" "github.com/smartcontractkit/chainlink-common/pkg/custmsg" "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/resourcemanager" "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink-common/pkg/services/orgresolver" "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" @@ -28,6 +31,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/workflows/dontime" generichost "github.com/smartcontractkit/chainlink-common/pkg/workflows/host" "github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host" + meteringpb "github.com/smartcontractkit/chainlink-protos/metering/go" eventsv2 "github.com/smartcontractkit/chainlink-protos/workflows/go/v2" "github.com/smartcontractkit/chainlink/v2/core/capabilities" "github.com/smartcontractkit/chainlink/v2/core/capabilities/confidentialrelay" @@ -64,6 +68,18 @@ type DrainableService interface { var ErrDrainInProgress = errors.New("drain in progress") +// Service-level metering identity constants for the workflow syncer. Service is +// the stable service constant; ResourcePool identifies the workflow_specs_v2 +// pool. The billing unit (ResourceType) and per-resource id (ResourceID) are +// carried on each Utilization. The coarse deployment/node/DON dimensions +// (product, tenant, environment, zone, don_id, node_id) are supplied at +// construction via WithIdentity. +const ( + meterService = "workflow-syncer-v2" + meterResourcePool = "workflow_specs_v2" + meterResourceType = "operations" +) + // eventHandler is a handler for WorkflowRegistryEvent events. Each event type has a corresponding method that handles the event. type eventHandler struct { services.Service @@ -88,8 +104,26 @@ type eventHandler struct { workflowEncryptionKey workflowkey.Key workflowDonSubscriber capabilities.DonSubscriber billingClient metering.BillingClient - orgResolver orgresolver.OrgResolver - secretsFetcher v2.SecretsFetcher + resourceManager *resourcemanager.ResourceManager + // meterIdentity is the base metering identity for this node's syncer: the + // six coarse dimensions (product, tenant, environment, zone, don_id, + // node_id) plus service/resource_pool. Per-resource billing fields + // (resource_type/resource_id/event_id/org_id/value) are attached at emit/ + // snapshot time via UtilizationFields. It is populated by WithIdentity in + // cre.go; Service/ResourcePool are forced to syncer constants regardless of + // the option. + meterIdentity resourcemanager.ResourceIdentity + // resolvedDonID holds the workflow DON id once resolved from the don notifier + // at start (the engine runs on the workflow DON). It is resolved + // asynchronously so node boot is not blocked while waiting for the DON to be + // set, and read on the hot snapshot/emit paths; an atomic keeps that read + // lock-free. Nil until resolved; baseIdentity folds it into meterIdentity. + resolvedDonID atomic.Pointer[string] + // rmUnregister removes this handler from the ResourceManager's snapshot + // registry; set in start, called in close. Nil until started. + rmUnregister func() + orgResolver orgresolver.OrgResolver + secretsFetcher v2.SecretsFetcher // localSecretOverrides is keyed by owner address; values are secret id -> secret value localSecretOverrides map[string]map[string]string @@ -153,6 +187,27 @@ func WithBillingClient(client metering.BillingClient) func(*eventHandler) { } } +// WithResourceManager overrides the default (disabled) ResourceManager used to +// emit metering.v1.MeterRecord events for the workflow_specs_v2 storage lifecycle. +func WithResourceManager(rm *resourcemanager.ResourceManager) func(*eventHandler) { + return func(e *eventHandler) { + e.resourceManager = rm + } +} + +// WithIdentity supplies the coarse metering identity dimensions sourced once at +// construction in cre.go (product/tenant/environment/zone/node_id from node +// metering config; don_id resolved later by the handler). The syncer +// Service/ResourcePool are stable constants and always overwrite whatever the +// caller passes. +func WithIdentity(id resourcemanager.ResourceIdentity) func(*eventHandler) { + return func(e *eventHandler) { + id.Service = meterService + id.ResourcePool = meterResourcePool + e.meterIdentity = id + } +} + func WithShardExecutionGuard(client shardorchestrator.ClientInterface, shardingEnabled bool, shardID uint32) func(*eventHandler) { return func(e *eventHandler) { e.shardOrchestratorClient = client @@ -306,7 +361,14 @@ func NewEventHandler( workflowArtifactsStore: workflowArtifacts, workflowEncryptionKey: workflowEncryptionKey, workflowDonSubscriber: workflowDonSubscriber, - tracer: noop.NewTracerProvider().Tracer(""), // default to noop, enable via WithDebugMode + resourceManager: resourcemanager.NewResourceManager(lggr, resourcemanager.ResourceManagerConfig{}), // default to disabled, enable via WithResourceManager + // Default identity carries only the service-level constants; the coarse + // deployment/node/DON dimensions are filled in by WithIdentity in cre.go. + meterIdentity: resourcemanager.ResourceIdentity{ + Service: meterService, + ResourcePool: meterResourcePool, + }, + tracer: noop.NewTracerProvider().Tracer(""), // default to noop, enable via WithDebugMode } metricsInst, metricsErr := newMetrics() if metricsErr != nil { @@ -327,20 +389,72 @@ func NewEventHandler( return eh, nil } -func (h *eventHandler) start(_ context.Context) error { +func (h *eventHandler) start(ctx context.Context) error { if h.moduleLRU != nil { h.moduleLRU.Start() } + // The handler is the single owner of its ResourceManager: it starts the RM + // (which owns the snapshot tick) and registers itself as the Meterable that + // is snapshotted. The RM is a no-op when metering is disabled, so this is + // safe regardless of the gate. + if h.resourceManager != nil { + if err := h.resourceManager.Start(ctx); err != nil { + return fmt.Errorf("failed to start resource manager: %w", err) + } + h.rmUnregister = h.resourceManager.Register(h) + h.resolveWorkflowDonID() + } return nil } +// resolveWorkflowDonID asynchronously resolves the workflow DON id (the engine +// runs on the workflow DON) and folds it into the metering identity. It +// subscribes to the don notifier and stores the first DON's id, so node boot is +// never blocked waiting for the DON to be set. Until it resolves, emitted records +// carry an empty don_id (the host-injection fallback semantics); once resolved, +// every subsequent record and snapshot carries it. Resolution happens at most +// once per start. +func (h *eventHandler) resolveWorkflowDonID() { + if h.workflowDonSubscriber == nil || h.resolvedDonID.Load() != nil { + return + } + h.eng.Go(func(ctx context.Context) { + ch, unsubscribe, err := h.workflowDonSubscriber.Subscribe(ctx) + if err != nil { + h.lggr.Warnw("failed to subscribe to workflow DON for metering identity; don_id will be empty", "err", err) + return + } + defer unsubscribe() + select { + case <-ctx.Done(): + return + case don := <-ch: + donID := strconv.FormatUint(uint64(don.ID), 10) + h.resolvedDonID.Store(&donID) + h.lggr.Debugw("resolved workflow DON id for metering identity", "donID", donID) + } + }) +} + func (h *eventHandler) close() error { if h.moduleLRU != nil { h.moduleLRU.Close() } es := h.engineRegistry.PopAll() - cs := make([]io.Closer, 0, len(es)+1) + // Emit a graceful-close RELEASE for each still-running engine before popping + // them, so a clean shutdown pairs every snapshot RESERVE. Best-effort and + // fail-open: metering must never gate shutdown. + h.emitGracefulCloseReleases(context.Background(), es) + // Stop snapshotting this handler, then close the ResourceManager service. + if h.rmUnregister != nil { + h.rmUnregister() + h.rmUnregister = nil + } + cs := make([]io.Closer, 0, len(es)+2) cs = append(cs, h.engineLimiters) + if h.resourceManager != nil { + cs = append(cs, h.resourceManager) + } for _, e := range es { cs = append(cs, e) } @@ -500,7 +614,7 @@ func (h *eventHandler) Handle(ctx context.Context, event Event) error { } }() - if herr = h.workflowDeletedEvent(ctx, payload); herr != nil { + if herr = h.workflowDeletedEvent(ctx, payload, WorkflowDeleted); herr != nil { if errors.Is(herr, ErrDrainInProgress) { logCustMsg(ctx, cma, fmt.Sprintf("workflow deletion deferred: %v", herr), h.lggr) } else { @@ -524,7 +638,7 @@ func (h *eventHandler) workflowActivatedEvent( ) error { // Convert WorkflowActivatedEvent to WorkflowRegisteredEvent since they have identical fields registeredPayload := WorkflowRegisteredEvent(payload) - return h.workflowRegisteredEvent(ctx, registeredPayload) + return h.workflowRegisteredEvent(ctx, registeredPayload, WorkflowActivated) } // workflowRegisteredEvent handles the WorkflowRegisteredEvent event type. @@ -532,9 +646,13 @@ func (h *eventHandler) workflowActivatedEvent( // workflowRegisteredEvent proceeds in two phases: // - phase 1 synchronizes the database state // - phase 2 synchronizes the state of the engine registry. +// originatingEvent names the event that triggered this call (e.g. WorkflowActivated +// when delegated from workflowActivatedEvent) and identifies the originating intent +// in emitted meter records. func (h *eventHandler) workflowRegisteredEvent( ctx context.Context, payload WorkflowRegisteredEvent, + originatingEvent WorkflowRegistryEventName, ) error { ctx, span := h.tracer.Start(ctx, "workflow_registered", trace.WithAttributes( @@ -557,6 +675,7 @@ func (h *eventHandler) workflowRegisteredEvent( if innerErr != nil { return innerErr } + h.emitMeterRecord(ctx, meteringpb.MeterAction_METER_ACTION_RESERVE, originatingEvent, payload.WorkflowID.Hex()) spec = newSpec case spec.WorkflowID != payload.WorkflowID.Hex(): @@ -564,6 +683,7 @@ func (h *eventHandler) workflowRegisteredEvent( if innerErr != nil { return innerErr } + h.emitMeterRecord(ctx, meteringpb.MeterAction_METER_ACTION_RESERVE, originatingEvent, payload.WorkflowID.Hex()) spec = newSpec case spec.Status != status: @@ -571,6 +691,7 @@ func (h *eventHandler) workflowRegisteredEvent( if _, innerErr := h.workflowArtifactsStore.UpsertWorkflowSpec(ctx, spec); innerErr != nil { return fmt.Errorf("failed to update workflow spec: %w", innerErr) } + h.emitMeterRecord(ctx, meteringpb.MeterAction_METER_ACTION_UPDATE, originatingEvent, payload.WorkflowID.Hex()) } // Next, let's synchronize the engine. @@ -820,13 +941,17 @@ func (h *eventHandler) workflowPausedEvent( ctx context.Context, payload WorkflowPausedEvent, ) error { - return h.workflowDeletedEvent(ctx, WorkflowDeletedEvent{WorkflowID: payload.WorkflowID}) + return h.workflowDeletedEvent(ctx, WorkflowDeletedEvent{WorkflowID: payload.WorkflowID}, WorkflowPaused) } // workflowDeletedEvent handles the WorkflowDeletedEvent event type. This method must remain idempotent. +// originatingEvent names the event that triggered this call (e.g. WorkflowPaused +// when delegated from workflowPausedEvent) and identifies the originating intent +// in emitted meter records. func (h *eventHandler) workflowDeletedEvent( ctx context.Context, payload WorkflowDeletedEvent, + originatingEvent WorkflowRegistryEventName, ) error { // The order in the handler is slightly different to the order in `tryEngineCleanup`. // This is because the engine requires its corresponding DB record to be present to be successfully @@ -865,6 +990,7 @@ func (h *eventHandler) workflowDeletedEvent( if err := h.workflowArtifactsStore.DeleteWorkflowArtifacts(ctx, payload.WorkflowID.Hex()); err != nil { return fmt.Errorf("failed to delete workflow artifacts: %w", err) } + h.emitMeterRecord(ctx, meteringpb.MeterAction_METER_ACTION_RELEASE, originatingEvent, workflowID) h.cleanupModuleCache(payload.WorkflowID.Hex()) @@ -885,6 +1011,125 @@ func (h *eventHandler) workflowDeletedEvent( return nil } +// emitMeterRecord emits a metering.v1.MeterRecord for a workflow_specs_v2 mutation. +// Callers must invoke it only after the database mutation has succeeded. Emission is +// fail-open: it never returns an error and must never affect event handling. +// Consumer-side dedup can use the emitted identity + action + +// utilization(resource_id/event_id). +// +// Known lost-record windows (consumer-side reconciliation tracked by SHARED-2141): +// - A node crash between the workflow spec database write and this emit loses the +// record (e.g. the RESERVE for a new spec): the syncer has no recovery path that +// re-emits records for mutations persisted before the crash. +// - A workflow paused or deleted while the node is down loses its RELEASE: +// reconciliation only synthesizes events for engine-registry entries, and the +// restarted node has no engine registered for the already-removed workflow. +func (h *eventHandler) emitMeterRecord(ctx context.Context, action meteringpb.MeterAction, originatingEvent WorkflowRegistryEventName, workflowID string) { + if h.resourceManager == nil { + return + } + // resource_id = workflow_id (the syncer has no shared physical resource); + // event_id = originating event name so lifecycle edges remain distinguishable. + identity := h.baseIdentity() + utilizations := []*meteringpb.Utilization{ + resourcemanager.NewUtilizationInt(1, resourcemanager.UtilizationFields{ + ResourceType: meterResourceType, + ResourceID: workflowID, + EventID: string(originatingEvent), + }), + } + h.resourceManager.EmitMeterRecord(ctx, identity, action, + utilizations) +} + +// baseIdentity returns the handler's metering identity with the workflow DON id +// folded in once it has been resolved from the don notifier. meterIdentity itself +// is immutable after construction (set via WithIdentity); only the DON id is +// learned later, so it is read from an atomic and overlaid here. +func (h *eventHandler) baseIdentity() resourcemanager.ResourceIdentity { + id := h.meterIdentity + if donID := h.resolvedDonID.Load(); donID != nil { + nodeID := "" + if id.Don != nil { + nodeID = id.Don.NodeID + } + id.Don = &resourcemanager.DonIdentity{ + DonID: *donID, + NodeID: nodeID, + } + } + return id +} + +// ResourceIdentity implements resourcemanager.Meterable: it returns the syncer's +// base identity (coarse deployment/DON/node dimensions + service/resource_pool). +// The ResourceManager uses it as the top-level identity of each emitted +// Snapshot. +func (h *eventHandler) ResourceIdentity() resourcemanager.ResourceIdentity { + return h.baseIdentity() +} + +// GetUtilization implements resourcemanager.Meterable: it returns one +// SnapshotEntry per running engine, each with one Utilization at value 1 +// (each running workflow holds one workflow_specs_v2 reservation). It is a pure +// in-memory read over the engine registry — each entry's resource_id is the +// workflow_id, which fully identifies the resource (Design 1; reconciling +// persisted-but-not-running specs is the Design 2 follow-up, see +// snapshotDesign2Followup below). +func (h *eventHandler) GetUtilization(_ context.Context) []resourcemanager.SnapshotEntry { + base := h.baseIdentity() + engines := h.engineRegistry.GetAll() + entries := make([]resourcemanager.SnapshotEntry, 0, len(engines)) + for _, e := range engines { + workflowID := e.WorkflowID.Hex() + entries = append(entries, resourcemanager.SnapshotEntry{ + Identity: base, + Utilizations: []*meteringpb.Utilization{ + resourcemanager.NewUtilizationInt(1, resourcemanager.UtilizationFields{ + ResourceType: meterResourceType, + ResourceID: workflowID, + }), + }, + }) + } + return entries +} + +// snapshotDesign2Followup documents the deferred reconciliation work. Design 1 +// (implemented here) snapshots only engines that are currently running in this +// node's engine registry. Design 2 — the follow-up — would additionally +// enumerate persisted-but-not-running specs via WorkflowSpecsDS.ListAll so that +// snapshots also account for specs the database has but for which no engine is +// live (e.g. paused specs, or specs whose engine failed to start). That requires +// a store-backed read on the tick (not a cheap in-memory snapshot) and a policy +// for the utilization value of a non-running spec, so it is intentionally left +// out of the in-memory GetUtilization above. Tracked as the syncer Design 2 +// follow-up. +const snapshotDesign2Followup = "Design 2: reconcile persisted-but-not-running specs via WorkflowSpecsDS.ListAll" + +// emitGracefulCloseReleases emits a RELEASE meter record for every engine still +// in the registry at shutdown, so a clean close pairs every snapshot RESERVE. It +// is fail-open like all metering emission. Called from close before the engines +// are popped. +func (h *eventHandler) emitGracefulCloseReleases(ctx context.Context, engines []ServiceWithMetadata) { + if h.resourceManager == nil { + return + } + identity := h.baseIdentity() + for _, e := range engines { + workflowID := e.WorkflowID.Hex() + utilizations := []*meteringpb.Utilization{ + resourcemanager.NewUtilizationInt(1, resourcemanager.UtilizationFields{ + ResourceType: meterResourceType, + ResourceID: workflowID, + EventID: string(WorkflowDeleted), + }), + } + h.resourceManager.EmitMeterRecord(ctx, identity, meteringpb.MeterAction_METER_ACTION_RELEASE, + utilizations) + } +} + // tryEngineCleanup attempts to stop the workflow engine for the given workflow ID. Does nothing if the // workflow engine is not running. func (h *eventHandler) tryEngineCleanup(workflowID types.WorkflowID) error { @@ -1004,7 +1249,7 @@ func (h *eventHandler) tryEngineCreate(ctx context.Context, spec *job.WorkflowSp } } - // Engine is fully initialized, add to registry with source tracking + // Engine is fully initialized, add to registry with source tracking. if err := h.engineRegistry.Add(wid, source, engine); err != nil { if closeErr := engine.Close(); closeErr != nil { return fmt.Errorf("failed to close workflow engine: %w during invariant violation: %w", closeErr, err) diff --git a/core/services/workflows/syncer/v2/handler_metering_test.go b/core/services/workflows/syncer/v2/handler_metering_test.go new file mode 100644 index 00000000000..9f31cc5b03b --- /dev/null +++ b/core/services/workflows/syncer/v2/handler_metering_test.go @@ -0,0 +1,456 @@ +package v2 + +import ( + "context" + "encoding/hex" + "errors" + "math/big" + "sync" + "testing" + "time" + + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + + "github.com/smartcontractkit/chainlink-common/keystore/corekeys/workflowkey" + "github.com/smartcontractkit/chainlink-common/pkg/custmsg" + commonlogger "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/resourcemanager" + "github.com/smartcontractkit/chainlink-common/pkg/services/servicetest" + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" + pkgworkflows "github.com/smartcontractkit/chainlink-common/pkg/workflows" + meteringpb "github.com/smartcontractkit/chainlink-protos/metering/go" + + "github.com/smartcontractkit/chainlink/v2/core/capabilities" + "github.com/smartcontractkit/chainlink/v2/core/logger" + "github.com/smartcontractkit/chainlink/v2/core/services/job" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/ratelimiter" + workflowstore "github.com/smartcontractkit/chainlink/v2/core/services/workflows/store" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/syncerlimiter" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/types" + v2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2" +) + +// recordingEmitter is a fake resourcemanager.Emitter that decodes and stores +// every emitted MeterRecord. If err is set, Emit fails instead. +type recordingEmitter struct { + mu sync.Mutex + err error + records []*meteringpb.MeterRecord +} + +func (r *recordingEmitter) Emit(_ context.Context, body []byte, _ ...any) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.err != nil { + return r.err + } + var record meteringpb.MeterRecord + if err := proto.Unmarshal(body, &record); err != nil { + return err + } + r.records = append(r.records, &record) + return nil +} + +func (r *recordingEmitter) Records() []*meteringpb.MeterRecord { + r.mu.Lock() + defer r.mu.Unlock() + records := make([]*meteringpb.MeterRecord, len(r.records)) + copy(records, r.records) + return records +} + +func newMeteringResourceManager(t *testing.T, enabled bool, emitter resourcemanager.Emitter) *resourcemanager.ResourceManager { + t.Helper() + return resourcemanager.NewResourceManager(commonlogger.Test(t), resourcemanager.ResourceManagerConfig{ + MeterRecordsEnabled: enabled, + Emitter: emitter, + }) +} + +// recordingSnapshotEmitter is a fake resourcemanager.Emitter that decodes and +// stores every emitted MeterSnapshot (one per active resource). Used to drive +// the Meterable snapshot path. +type recordingSnapshotEmitter struct { + mu sync.Mutex + snapshots []*meteringpb.MeterSnapshot +} + +func (r *recordingSnapshotEmitter) Emit(_ context.Context, body []byte, _ ...any) error { + r.mu.Lock() + defer r.mu.Unlock() + var snapshot meteringpb.MeterSnapshot + if err := proto.Unmarshal(body, &snapshot); err != nil { + return err + } + r.snapshots = append(r.snapshots, &snapshot) + return nil +} + +func (r *recordingSnapshotEmitter) Snapshots() []*meteringpb.MeterSnapshot { + r.mu.Lock() + defer r.mu.Unlock() + snapshots := make([]*meteringpb.MeterSnapshot, len(r.snapshots)) + copy(snapshots, r.snapshots) + return snapshots +} + +func newMeteringTestHandler(t *testing.T, artifactsStore WorkflowArtifactsStore, rm *resourcemanager.ResourceManager) *eventHandler { + t.Helper() + lggr := logger.TestLogger(t) + lf := limits.Factory{Logger: lggr} + registry := capabilities.NewRegistry(lggr) + registry.SetLocalRegistry(&capabilities.TestMetadataRegistry{}) + limiters, err := v2.NewLimiters(lf, nil) + require.NoError(t, err) + rl, err := ratelimiter.NewRateLimiter(rlConfig) + require.NoError(t, err) + workflowLimits, err := syncerlimiter.NewWorkflowLimits(lggr, wlConfig, lf) + require.NoError(t, err) + + h, err := NewEventHandler( + lggr, + workflowstore.NewInMemoryStore(lggr, clockwork.NewFakeClock()), + nil, + true, + registry, + NewEngineRegistry(), + custmsg.NewLabeler(), + limiters, + nil, + rl, + workflowLimits, + artifactsStore, + workflowkey.MustNewXXXTestingOnly(big.NewInt(1)), + &testDonNotifier{}, + WithResourceManager(rm), + WithEngineFactoryFn(mockEngineFactory), + ) + require.NoError(t, err) + return h +} + +func requireMeterRecord(t *testing.T, record *meteringpb.MeterRecord, action meteringpb.MeterAction, originatingEvent WorkflowRegistryEventName, workflowID string) { + t.Helper() + require.NotNil(t, record.Identity) + assert.Equal(t, "workflow-syncer-v2", record.Identity.Service) + assert.Equal(t, "workflow_specs_v2", record.Identity.ResourcePool) + assert.Equal(t, action, record.Action) + assert.NotNil(t, record.Timestamp) + require.Len(t, record.Utilizations, 1) + util := record.Utilizations[0] + assert.Equal(t, "1", util.Value) + assert.Equal(t, "operations", util.ResourceType) + // resource_id = workflow_id for the syncer (no shared physical resource). + assert.Equal(t, workflowID, util.ResourceId) + assert.Equal(t, string(originatingEvent), util.EventId) +} + +func Test_meterRecords(t *testing.T) { + t.Parallel() + + wfOwner := []byte{0xaa, 0xbb, 0xcc, 0xdd} + wfOwnerHex := hex.EncodeToString(wfOwner) + + t.Run("registered event creating a new spec emits RESERVE", func(t *testing.T) { + t.Parallel() + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{}, newMeteringResourceManager(t, true, emitter)) + + wfID := types.WorkflowID{1} + err := h.workflowRegisteredEvent(t.Context(), WorkflowRegisteredEvent{ + Status: WorkflowStatusPaused, + WorkflowID: wfID, + WorkflowOwner: wfOwner, + WorkflowName: "wf-name", + }, WorkflowRegistered) + require.NoError(t, err) + + records := emitter.Records() + require.Len(t, records, 1) + requireMeterRecord(t, records[0], meteringpb.MeterAction_METER_ACTION_RESERVE, WorkflowRegistered, wfID.Hex()) + }) + + t.Run("retried registered event emits equivalent utilization identity", func(t *testing.T) { + t.Parallel() + emitter := &recordingEmitter{} + // The stub never returns a stored spec, so each call replays the + // new-spec path exactly as a reprocessed event would. + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{}, newMeteringResourceManager(t, true, emitter)) + + event := WorkflowRegisteredEvent{ + Status: WorkflowStatusPaused, + WorkflowID: types.WorkflowID{2}, + WorkflowOwner: wfOwner, + WorkflowName: "wf-name", + } + require.NoError(t, h.workflowRegisteredEvent(t.Context(), event, WorkflowRegistered)) + require.NoError(t, h.workflowRegisteredEvent(t.Context(), event, WorkflowRegistered)) + + records := emitter.Records() + require.Len(t, records, 2) + require.Len(t, records[0].Utilizations, 1) + require.Len(t, records[1].Utilizations, 1) + assert.Equal(t, records[0].Action, records[1].Action) + assert.Equal(t, records[0].Identity.GetService(), records[1].Identity.GetService()) + assert.Equal(t, records[0].Identity.GetResourcePool(), records[1].Identity.GetResourcePool()) + assert.Equal(t, records[0].Utilizations[0].GetResourceId(), records[1].Utilizations[0].GetResourceId()) + assert.Equal(t, records[0].Utilizations[0].GetEventId(), records[1].Utilizations[0].GetEventId()) + }) + + t.Run("activated event with existing spec emits UPDATE", func(t *testing.T) { + t.Parallel() + binary := []byte("binary-data") + config := []byte("") + giveWFID, err := pkgworkflows.GenerateWorkflowID(wfOwner, "wf-name", binary, config, "") + require.NoError(t, err) + wfID := types.WorkflowID(giveWFID) + + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{ + spec: &job.WorkflowSpec{ + Workflow: hex.EncodeToString(binary), + Config: string(config), + WorkflowID: wfID.Hex(), + Status: job.WorkflowSpecStatusPaused, + WorkflowOwner: wfOwnerHex, + WorkflowName: "wf-name", + }, + }, newMeteringResourceManager(t, true, emitter)) + + err = h.workflowActivatedEvent(t.Context(), WorkflowActivatedEvent{ + Status: WorkflowStatusActive, + WorkflowID: wfID, + WorkflowOwner: wfOwner, + WorkflowName: "wf-name", + }) + require.NoError(t, err) + + records := emitter.Records() + require.Len(t, records, 1) + requireMeterRecord(t, records[0], meteringpb.MeterAction_METER_ACTION_UPDATE, WorkflowActivated, wfID.Hex()) + }) + + t.Run("paused event emits RELEASE after artifacts are deleted", func(t *testing.T) { + t.Parallel() + wfID := types.WorkflowID{3} + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{ + spec: &job.WorkflowSpec{ + WorkflowID: wfID.Hex(), + Status: job.WorkflowSpecStatusActive, + WorkflowOwner: wfOwnerHex, + }, + }, newMeteringResourceManager(t, true, emitter)) + + require.NoError(t, h.workflowPausedEvent(t.Context(), WorkflowPausedEvent{WorkflowID: wfID})) + + records := emitter.Records() + require.Len(t, records, 1) + requireMeterRecord(t, records[0], meteringpb.MeterAction_METER_ACTION_RELEASE, WorkflowPaused, wfID.Hex()) + }) + + t.Run("deleted event emits RELEASE after artifacts are deleted", func(t *testing.T) { + t.Parallel() + wfID := types.WorkflowID{4} + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{ + spec: &job.WorkflowSpec{ + WorkflowID: wfID.Hex(), + Status: job.WorkflowSpecStatusActive, + WorkflowOwner: wfOwnerHex, + }, + }, newMeteringResourceManager(t, true, emitter)) + + require.NoError(t, h.workflowDeletedEvent(t.Context(), WorkflowDeletedEvent{WorkflowID: wfID}, WorkflowDeleted)) + + records := emitter.Records() + require.Len(t, records, 1) + requireMeterRecord(t, records[0], meteringpb.MeterAction_METER_ACTION_RELEASE, WorkflowDeleted, wfID.Hex()) + }) + + t.Run("no record when creating a new spec fails", func(t *testing.T) { + t.Parallel() + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{upsertErr: assert.AnError}, newMeteringResourceManager(t, true, emitter)) + + err := h.workflowRegisteredEvent(t.Context(), WorkflowRegisteredEvent{ + Status: WorkflowStatusPaused, + WorkflowID: types.WorkflowID{5}, + WorkflowOwner: wfOwner, + WorkflowName: "wf-name", + }, WorkflowRegistered) + require.ErrorIs(t, err, assert.AnError) + assert.Empty(t, emitter.Records()) + }) + + t.Run("no record when a status-only update fails", func(t *testing.T) { + t.Parallel() + wfID := types.WorkflowID{6} + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{ + spec: &job.WorkflowSpec{ + WorkflowID: wfID.Hex(), + Status: job.WorkflowSpecStatusActive, + WorkflowOwner: wfOwnerHex, + }, + upsertErr: assert.AnError, + }, newMeteringResourceManager(t, true, emitter)) + + err := h.workflowRegisteredEvent(t.Context(), WorkflowRegisteredEvent{ + Status: WorkflowStatusPaused, + WorkflowID: wfID, + WorkflowOwner: wfOwner, + WorkflowName: "wf-name", + }, WorkflowRegistered) + require.ErrorIs(t, err, assert.AnError) + assert.Empty(t, emitter.Records()) + }) + + t.Run("no record when deleting artifacts fails", func(t *testing.T) { + t.Parallel() + wfID := types.WorkflowID{7} + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{ + spec: &job.WorkflowSpec{ + WorkflowID: wfID.Hex(), + Status: job.WorkflowSpecStatusActive, + WorkflowOwner: wfOwnerHex, + }, + deleteErr: assert.AnError, + }, newMeteringResourceManager(t, true, emitter)) + + err := h.workflowDeletedEvent(t.Context(), WorkflowDeletedEvent{WorkflowID: wfID}, WorkflowDeleted) + require.ErrorIs(t, err, assert.AnError) + assert.Empty(t, emitter.Records()) + }) + + t.Run("no record while a delete is deferred by drain; exactly one on the successful retry", func(t *testing.T) { + t.Parallel() + wfID := types.WorkflowID{8} + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{ + spec: &job.WorkflowSpec{ + WorkflowID: wfID.Hex(), + Status: job.WorkflowSpecStatusActive, + WorkflowOwner: wfOwnerHex, + }, + }, newMeteringResourceManager(t, true, emitter)) + + drainable := &mockDrainableEngine{} + drainable.activeExecutions.Store(1) + require.NoError(t, h.engineRegistry.Add(wfID, "test-source", drainable)) + + err := h.workflowDeletedEvent(t.Context(), WorkflowDeletedEvent{WorkflowID: wfID}, WorkflowDeleted) + require.ErrorIs(t, err, ErrDrainInProgress) + assert.Empty(t, emitter.Records()) + + drainable.activeExecutions.Store(0) + require.NoError(t, h.workflowDeletedEvent(t.Context(), WorkflowDeletedEvent{WorkflowID: wfID}, WorkflowDeleted)) + + records := emitter.Records() + require.Len(t, records, 1) + requireMeterRecord(t, records[0], meteringpb.MeterAction_METER_ACTION_RELEASE, WorkflowDeleted, wfID.Hex()) + }) + + t.Run("emit failure never fails event handling", func(t *testing.T) { + t.Parallel() + wfID := types.WorkflowID{9} + emitter := &recordingEmitter{err: errors.New("beholder unavailable")} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{ + spec: &job.WorkflowSpec{ + WorkflowID: wfID.Hex(), + Status: job.WorkflowSpecStatusActive, + WorkflowOwner: wfOwnerHex, + }, + }, newMeteringResourceManager(t, true, emitter)) + + require.NoError(t, h.workflowRegisteredEvent(t.Context(), WorkflowRegisteredEvent{ + Status: WorkflowStatusPaused, + WorkflowID: wfID, + WorkflowOwner: wfOwner, + WorkflowName: "wf-name", + }, WorkflowRegistered)) + require.NoError(t, h.workflowDeletedEvent(t.Context(), WorkflowDeletedEvent{WorkflowID: wfID}, WorkflowDeleted)) + assert.Empty(t, emitter.Records()) + }) + + t.Run("disabled resource manager emits nothing", func(t *testing.T) { + t.Parallel() + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{}, newMeteringResourceManager(t, false, emitter)) + + require.NoError(t, h.workflowRegisteredEvent(t.Context(), WorkflowRegisteredEvent{ + Status: WorkflowStatusPaused, + WorkflowID: types.WorkflowID{10}, + WorkflowOwner: wfOwner, + WorkflowName: "wf-name", + }, WorkflowRegistered)) + assert.Empty(t, emitter.Records()) + }) + + t.Run("snapshot emits one MeterSnapshot per running engine", func(t *testing.T) { + t.Parallel() + emitter := &recordingSnapshotEmitter{} + clock := clockwork.NewFakeClockAt(time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)) + rm := resourcemanager.NewResourceManager(commonlogger.Test(t), resourcemanager.ResourceManagerConfig{ + MeterRecordsEnabled: true, + MeterSnapshotsEnabled: true, + Emitter: emitter, + SnapshotInterval: time.Minute, + Clock: clock, + }) + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{}, rm) + unregister := rm.Register(h) + t.Cleanup(unregister) + + // Register two engines; the running engine registry is the in-memory source + // the snapshot reads (no per-tick GetWorkflowSpec). + wfID1 := types.WorkflowID{20} + wfID2 := types.WorkflowID{21} + require.NoError(t, h.engineRegistry.Add(wfID1, "test-source", &fakeService{})) + require.NoError(t, h.engineRegistry.Add(wfID2, "test-source", &fakeService{})) + + servicetest.Run(t, rm) + require.NoError(t, clock.BlockUntilContext(t.Context(), 1)) + clock.Advance(time.Minute) + + require.Eventually(t, func() bool { + return len(emitter.Snapshots()) == 2 + }, time.Second, time.Millisecond) + snapshots := emitter.Snapshots() + require.Len(t, snapshots, 2) + + byWorkflowID := map[string]*meteringpb.MeterSnapshot{} + for _, snap := range snapshots { + require.NotNil(t, snap.Identity) + assert.Equal(t, "workflow-syncer-v2", snap.Identity.Service) + assert.Equal(t, "workflow_specs_v2", snap.Identity.ResourcePool) + require.Len(t, snap.Utilization, 1) + assert.Equal(t, "1", snap.Utilization[0].Value) + // resource_id = workflow_id fully identifies the resource; no labels. + byWorkflowID[snap.Utilization[0].ResourceId] = snap + } + require.NotNil(t, byWorkflowID[wfID1.Hex()], "snapshot must contain an entry for the first running engine") + require.NotNil(t, byWorkflowID[wfID2.Hex()], "snapshot must contain an entry for the second running engine") + }) + + t.Run("graceful close emits a RELEASE per running engine", func(t *testing.T) { + t.Parallel() + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{}, newMeteringResourceManager(t, true, emitter)) + + wfID := types.WorkflowID{30} + require.NoError(t, h.engineRegistry.Add(wfID, "test-source", &fakeService{})) + + es := h.engineRegistry.GetAll() + h.emitGracefulCloseReleases(t.Context(), es) + + records := emitter.Records() + require.Len(t, records, 1) + requireMeterRecord(t, records[0], meteringpb.MeterAction_METER_ACTION_RELEASE, WorkflowDeleted, wfID.Hex()) + }) +} diff --git a/core/services/workflows/syncer/v2/handler_test.go b/core/services/workflows/syncer/v2/handler_test.go index 25d576fe5bb..8dc521088b7 100644 --- a/core/services/workflows/syncer/v2/handler_test.go +++ b/core/services/workflows/syncer/v2/handler_test.go @@ -247,7 +247,7 @@ func Test_workflowRegisteredHandler(t *testing.T) { signedURLParameter := "?auth=abc123" defaultValidationFn := func(t *testing.T, ctx context.Context, event WorkflowRegisteredEvent, h *eventHandler, s *artifacts.Store, wfOwner []byte, wfName string, wfID types.WorkflowID, _ *mockFetcher) { - err := h.workflowRegisteredEvent(ctx, event) + err := h.workflowRegisteredEvent(ctx, event, WorkflowRegistered) require.NoError(t, err) // Verify the record is updated in the database @@ -389,7 +389,7 @@ func Test_workflowRegisteredHandler(t *testing.T) { validationFn: func(t *testing.T, ctx context.Context, event WorkflowRegisteredEvent, h *eventHandler, s *artifacts.Store, wfOwner []byte, wfName string, wfID types.WorkflowID, fetcher *mockFetcher, binaryURL string, configURL string, ) { - err := h.workflowRegisteredEvent(ctx, event) + err := h.workflowRegisteredEvent(ctx, event, WorkflowRegistered) require.Error(t, err) require.ErrorIs(t, err, assert.AnError) }, @@ -427,7 +427,7 @@ func Test_workflowRegisteredHandler(t *testing.T) { me := &mockEngine{} err := h.engineRegistry.Add(wfID, event.Source, me) require.NoError(t, err) - err = h.workflowRegisteredEvent(ctx, event) + err = h.workflowRegisteredEvent(ctx, event, WorkflowRegistered) require.NoError(t, err) }, }, @@ -466,7 +466,7 @@ func Test_workflowRegisteredHandler(t *testing.T) { oldWfIDBytes := [32]byte{0, 1, 2, 3, 5} err := h.engineRegistry.Add(oldWfIDBytes, event.Source, me) require.NoError(t, err) - err = h.workflowRegisteredEvent(ctx, event) + err = h.workflowRegisteredEvent(ctx, event, WorkflowRegistered) require.NoError(t, err) engineInRegistry, ok := h.engineRegistry.Get(wfID) assert.True(t, ok) @@ -505,7 +505,7 @@ func Test_workflowRegisteredHandler(t *testing.T) { validationFn: func(t *testing.T, ctx context.Context, event WorkflowRegisteredEvent, h *eventHandler, s *artifacts.Store, wfOwner []byte, wfName string, wfID types.WorkflowID, fetcher *mockFetcher, binaryURL string, configURL string, ) { - err := h.workflowRegisteredEvent(ctx, event) + err := h.workflowRegisteredEvent(ctx, event, WorkflowRegistered) require.NoError(t, err) // Verify the record is updated in the database @@ -568,7 +568,7 @@ func Test_workflowRegisteredHandler(t *testing.T) { _, err := s.UpsertWorkflowSpec(ctx, entry) require.NoError(t, err) - err = h.workflowRegisteredEvent(ctx, event) + err = h.workflowRegisteredEvent(ctx, event, WorkflowRegistered) require.NoError(t, err) // Verify the record is updated in the database @@ -794,7 +794,7 @@ func Test_workflowRegisteredHandler_confidentialRouting(t *testing.T) { } ctx = contexts.WithCRE(ctx, contexts.CRE{Owner: hex.EncodeToString(wfOwner), Workflow: wfIDString}) - err = h.workflowRegisteredEvent(ctx, event) + err = h.workflowRegisteredEvent(ctx, event, WorkflowRegistered) require.NoError(t, err) assert.Eventually(t, confidential.ran.Load, 10*time.Second, time.Millisecond) @@ -883,7 +883,7 @@ func Test_workflowRegisteredHandler_confidentialRouting(t *testing.T) { } ctx = contexts.WithCRE(ctx, contexts.CRE{Owner: hex.EncodeToString(wfOwner), Workflow: wfIDString}) - err = h.workflowRegisteredEvent(ctx, event) + err = h.workflowRegisteredEvent(ctx, event, WorkflowRegistered) require.NoError(t, err) assert.Eventually(t, action.ran.Load, 10*time.Second, time.Millisecond) @@ -1109,7 +1109,7 @@ func Test_workflowDeletedHandler(t *testing.T) { ) require.NoError(t, err) ctx = contexts.WithCRE(ctx, contexts.CRE{Owner: hex.EncodeToString(wfOwner), Workflow: wfIDString}) - err = h.workflowRegisteredEvent(ctx, active) + err = h.workflowRegisteredEvent(ctx, active, WorkflowRegistered) require.NoError(t, err) // Verify the record is updated in the database @@ -1128,7 +1128,7 @@ func Test_workflowDeletedHandler(t *testing.T) { deleteEvent := WorkflowDeletedEvent{ WorkflowID: giveWFID, } - err = h.workflowDeletedEvent(ctx, deleteEvent) + err = h.workflowDeletedEvent(ctx, deleteEvent, WorkflowDeleted) require.NoError(t, err) // Verify the record is deleted in the database @@ -1183,7 +1183,7 @@ func Test_workflowDeletedHandler(t *testing.T) { deleteEvent := WorkflowDeletedEvent{ WorkflowID: giveWFID, } - err = h.workflowDeletedEvent(ctx, deleteEvent) + err = h.workflowDeletedEvent(ctx, deleteEvent, WorkflowDeleted) require.NoError(t, err) // Verify the record is deleted in the database @@ -1264,7 +1264,7 @@ func Test_workflowDeletedHandler(t *testing.T) { ) require.NoError(t, err) ctx = contexts.WithCRE(ctx, contexts.CRE{Owner: hex.EncodeToString(wfOwner), Workflow: wfIDString}) - err = h.workflowRegisteredEvent(ctx, active) + err = h.workflowRegisteredEvent(ctx, active, WorkflowRegistered) require.NoError(t, err) // Verify the record is updated in the database @@ -1283,7 +1283,7 @@ func Test_workflowDeletedHandler(t *testing.T) { deleteEvent := WorkflowDeletedEvent{ WorkflowID: giveWFID, } - err = h.workflowDeletedEvent(ctx, deleteEvent) + err = h.workflowDeletedEvent(ctx, deleteEvent, WorkflowDeleted) require.Error(t, err, failWith) // Verify the record is still in the DB @@ -1298,6 +1298,7 @@ func Test_workflowDeletedHandler(t *testing.T) { type stubWorkflowArtifactsStore struct { spec *job.WorkflowSpec + upsertErr error deleteErr error deleteCalls atomic.Int32 } @@ -1314,6 +1315,9 @@ func (s *stubWorkflowArtifactsStore) GetWorkflowSpec(context.Context, string) (* } func (s *stubWorkflowArtifactsStore) UpsertWorkflowSpec(context.Context, *job.WorkflowSpec) (int64, error) { + if s.upsertErr != nil { + return 0, s.upsertErr + } return 1, nil } @@ -1342,7 +1346,7 @@ func Test_workflowDeletedEvent_DrainInProgress(t *testing.T) { workflowArtifactsStore: artifactStore, } - err := h.workflowDeletedEvent(t.Context(), WorkflowDeletedEvent{WorkflowID: workflowID}) + err := h.workflowDeletedEvent(t.Context(), WorkflowDeletedEvent{WorkflowID: workflowID}, WorkflowDeleted) require.Error(t, err) require.ErrorIs(t, err, ErrDrainInProgress) assert.Equal(t, int32(1), drainable.drainCalls.Load()) @@ -1368,7 +1372,7 @@ func Test_workflowDeletedEvent_IgnoresErrAlreadyStopped(t *testing.T) { workflowArtifactsStore: artifactStore, } - err := h.workflowDeletedEvent(t.Context(), WorkflowDeletedEvent{WorkflowID: workflowID}) + err := h.workflowDeletedEvent(t.Context(), WorkflowDeletedEvent{WorkflowID: workflowID}, WorkflowDeleted) require.NoError(t, err) assert.Equal(t, int32(1), drainable.closeCalls.Load()) assert.Equal(t, int32(1), artifactStore.deleteCalls.Load()) @@ -1406,7 +1410,7 @@ func Test_workflowRegisteredEvent_DrainingEngineNotTreatedAsHealthy(t *testing.T err := h.workflowRegisteredEvent(t.Context(), WorkflowRegisteredEvent{ Status: WorkflowStatusActive, WorkflowID: workflowID, - }) + }, WorkflowRegistered) require.Error(t, err) require.Contains(t, err.Error(), "could not clean up old engine") assert.Equal(t, int32(1), drainable.closeCalls.Load()) diff --git a/core/web/resolver/testdata/config-empty-effective.toml b/core/web/resolver/testdata/config-empty-effective.toml index 6aeec786180..cb9ed7589f7 100644 --- a/core/web/resolver/testdata/config-empty-effective.toml +++ b/core/web/resolver/testdata/config-empty-effective.toml @@ -364,6 +364,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Tenant = '' +NumericTenantID = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/core/web/resolver/testdata/config-full.toml b/core/web/resolver/testdata/config-full.toml index dc1cc00ae4d..66a6e5a19fd 100644 --- a/core/web/resolver/testdata/config-full.toml +++ b/core/web/resolver/testdata/config-full.toml @@ -382,6 +382,16 @@ Foo = 'bar' Enabled = true Prefixes = ['ocr_'] +[Metering] +MeterRecordsEnabled = true +MeterSnapshotsEnabled = true +Product = 'cre' +Tenant = 'mainline' +NumericTenantID = '42' +Environment = 'production' +Zone = 'wf-zone-a' +NodeID = 'csa-pubkey-1' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/core/web/resolver/testdata/config-multi-chain-effective.toml b/core/web/resolver/testdata/config-multi-chain-effective.toml index a98ec27c405..c7a8e8dc3eb 100644 --- a/core/web/resolver/testdata/config-multi-chain-effective.toml +++ b/core/web/resolver/testdata/config-multi-chain-effective.toml @@ -364,6 +364,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Tenant = '' +NumericTenantID = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/deployment/go.mod b/deployment/go.mod index 3521ee2f1cb..d380046bfbd 100644 --- a/deployment/go.mod +++ b/deployment/go.mod @@ -47,7 +47,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb github.com/smartcontractkit/chainlink-ccip/deployment v0.0.0-20260624154507-ea7ff77a0ddb - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 github.com/smartcontractkit/chainlink-common/keystore v1.2.0 github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 @@ -440,6 +440,7 @@ require ( github.com/smartcontractkit/chainlink-protos/chainlink-ccv/verifier v0.0.0-20251211142334-5c3421fe2c8d // indirect github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36 // indirect github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 // indirect + github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 // indirect github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 // indirect github.com/smartcontractkit/chainlink-protos/op-catalog v0.1.0 // indirect github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd // indirect diff --git a/deployment/go.sum b/deployment/go.sum index 37355a0b094..3824cdbb1a7 100644 --- a/deployment/go.sum +++ b/deployment/go.sum @@ -1385,8 +1385,8 @@ github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094 h github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094/go.mod h1:3XQmy2zQHrmufebP7dM+x595+gghxzyIKxK9wv91Rh8= github.com/smartcontractkit/chainlink-ccv/deployment v0.0.2-0.20260616151800-9a3a31c4e194 h1:QxZkbKtQyPtVLYP4eMwc+VbXY7M5ve1deSiLZ2pOA+Y= github.com/smartcontractkit/chainlink-ccv/deployment v0.0.2-0.20260616151800-9a3a31c4e194/go.mod h1:bNMFRxwWdgVFdSsFZRmsUUPoBUncU3RM765K99svIKM= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 h1:ywWOEZYL4DhVwf/TQ5jXsGlx6CzERsYbqd6ov3OS1sc= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0/go.mod h1:ncZiIgraMh4F9lWDoJwB1TX395qZFR5bvnZcHbD0A1Q= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 h1:uvUnQlWgJGJBgtXpxpgGTipUZPxp/JugxBK7KUe41DQ= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6/go.mod h1:2hczeHjBapPah8d3EOoRAfzI4gH3caZToMhAUG8WS3I= 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.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1431,6 +1431,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0 h1:FC+WdJ8Y github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0/go.mod h1:2ahgl5bI9+fMCD+dBC785Lak38Tb7ApdTe5I8a09Qp0= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 h1:NJdGFhzT6zMaTod4QkBqVD2sg0I25iw1boOYtTpEwRo= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305/go.mod h1:qSTSwX3cBP3FKQwQacdjArqv0g6QnukjV4XuzO6UyoY= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 h1:B355/rV/lwpZl3C5iVsFQZU7+LeA+5BTTIzTDYlDOrA= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 h1:bnSl5p3mFekSJ6QcbZ1TmHn2ffYiX8xk6hNzVmyhstQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= github.com/smartcontractkit/chainlink-protos/op-catalog v0.1.0 h1:hGEJFD2X3oNIPXQbtIPxCJyg5CcKglRCYBmESS+gmeQ= diff --git a/docs/CONFIG.md b/docs/CONFIG.md index af944ec92e5..b5a8cb50c0d 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -2566,6 +2566,71 @@ Prefixes = ["go_"] # Default Prefixes is a set of filters to restrict which prometheus metrics are forwarded based on prefix matching. By default, we only forward the go runtime metrics. Empty means forward everything. +## Metering +```toml +[Metering] +MeterRecordsEnabled = false # Default +MeterSnapshotsEnabled = false # Default +Product = '' # Default +Tenant = '' # Default +NumericTenantID = '' # Default +Environment = '' # Default +Zone = '' # Default +NodeID = '' # Default +``` +Metering configures durable resource metering emission and the coarse +deployment/node identity dimensions stamped on emitted MeterRecords and +MeterSnapshots. These are plumbed to LOOP plugins via env. + +### MeterRecordsEnabled +```toml +MeterRecordsEnabled = false # Default +``` +MeterRecordsEnabled enables durable MeterRecord emission for LOOP plugins. + +### MeterSnapshotsEnabled +```toml +MeterSnapshotsEnabled = false # Default +``` +MeterSnapshotsEnabled enables durable MeterSnapshot emission for LOOP plugins. +Requires MeterRecordsEnabled = true. + +### Product +```toml +Product = '' # Default +``` +Product is the deployment product identity dimension, e.g. 'cre'. + +### Tenant +```toml +Tenant = '' # Default +``` +Tenant is the human-readable tenant name, e.g. 'mainline'. + +### NumericTenantID +```toml +NumericTenantID = '' # Default +``` +NumericTenantID is the numbered tenant identifier represented as a string. + +### Environment +```toml +Environment = '' # Default +``` +Environment is the deployment environment identity dimension, e.g. 'production'. + +### Zone +```toml +Zone = '' # Default +``` +Zone is the deployment zone identity dimension, e.g. 'wf-zone-a'. + +### NodeID +```toml +NodeID = '' # Default +``` +NodeID is the node's logical name, e.g. 'clp-cre-wf-zone-a-1' (not the CSA public key). + ## CRE.Streams ```toml [CRE.Streams] diff --git a/go.mod b/go.mod index be25ac97daa..78218df496e 100644 --- a/go.mod +++ b/go.mod @@ -85,7 +85,7 @@ 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.20260706093831-dad26b360094 - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 github.com/smartcontractkit/chainlink-common/keystore v1.2.0 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 @@ -100,6 +100,7 @@ require ( github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260623200841-e0322b819f62 github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36 github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 + github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 github.com/smartcontractkit/chainlink-protos/orchestrator v0.11.0 github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd diff --git a/go.sum b/go.sum index 7ac9b6f3c84..1d03191a326 100644 --- a/go.sum +++ b/go.sum @@ -1162,8 +1162,8 @@ 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.20260706093831-dad26b360094 h1:uMn1w/n95p+pSxK6hYNqji/sDaab8D0Cxuph9qUkM2g= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094/go.mod h1:3XQmy2zQHrmufebP7dM+x595+gghxzyIKxK9wv91Rh8= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 h1:ywWOEZYL4DhVwf/TQ5jXsGlx6CzERsYbqd6ov3OS1sc= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0/go.mod h1:ncZiIgraMh4F9lWDoJwB1TX395qZFR5bvnZcHbD0A1Q= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 h1:uvUnQlWgJGJBgtXpxpgGTipUZPxp/JugxBK7KUe41DQ= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6/go.mod h1:2hczeHjBapPah8d3EOoRAfzI4gH3caZToMhAUG8WS3I= 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.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1204,6 +1204,8 @@ github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546- github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36/go.mod h1:vL1bDgPSJjV0EqHYs4dDlR+EEE0cJchgvGLYXhwIjXY= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 h1:NJdGFhzT6zMaTod4QkBqVD2sg0I25iw1boOYtTpEwRo= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305/go.mod h1:qSTSwX3cBP3FKQwQacdjArqv0g6QnukjV4XuzO6UyoY= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 h1:B355/rV/lwpZl3C5iVsFQZU7+LeA+5BTTIzTDYlDOrA= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 h1:bnSl5p3mFekSJ6QcbZ1TmHn2ffYiX8xk6hNzVmyhstQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= github.com/smartcontractkit/chainlink-protos/orchestrator v0.11.0 h1:NXKTdIESAiCkVnPS6dyZP+NXVek3GzXa6P4uFAs0o8Y= diff --git a/integration-tests/go.mod b/integration-tests/go.mod index 2f4828c4fd7..efe2be9ea9a 100644 --- a/integration-tests/go.mod +++ b/integration-tests/go.mod @@ -33,7 +33,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260624154507-ea7ff77a0ddb github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 github.com/smartcontractkit/chainlink-common/keystore v1.2.0 github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260623170329-4577ef4ba0ae @@ -60,7 +60,10 @@ require ( gopkg.in/guregu/null.v4 v4.0.0 ) -require github.com/smartcontractkit/chainlink-sui/codec v0.0.0-20260706170510-94f3a0409ea5 // indirect +require ( + github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 // indirect + github.com/smartcontractkit/chainlink-sui/codec v0.0.0-20260706170510-94f3a0409ea5 // indirect +) require ( cloud.google.com/go/compute/metadata v0.9.0 // indirect diff --git a/integration-tests/go.sum b/integration-tests/go.sum index 8a628bc6bb3..09f39fdb0c3 100644 --- a/integration-tests/go.sum +++ b/integration-tests/go.sum @@ -1372,8 +1372,8 @@ github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094 h github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094/go.mod h1:3XQmy2zQHrmufebP7dM+x595+gghxzyIKxK9wv91Rh8= github.com/smartcontractkit/chainlink-ccv/deployment v0.0.2-0.20260616151800-9a3a31c4e194 h1:QxZkbKtQyPtVLYP4eMwc+VbXY7M5ve1deSiLZ2pOA+Y= github.com/smartcontractkit/chainlink-ccv/deployment v0.0.2-0.20260616151800-9a3a31c4e194/go.mod h1:bNMFRxwWdgVFdSsFZRmsUUPoBUncU3RM765K99svIKM= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 h1:ywWOEZYL4DhVwf/TQ5jXsGlx6CzERsYbqd6ov3OS1sc= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0/go.mod h1:ncZiIgraMh4F9lWDoJwB1TX395qZFR5bvnZcHbD0A1Q= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 h1:uvUnQlWgJGJBgtXpxpgGTipUZPxp/JugxBK7KUe41DQ= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6/go.mod h1:2hczeHjBapPah8d3EOoRAfzI4gH3caZToMhAUG8WS3I= 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.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1418,6 +1418,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0 h1:FC+WdJ8Y github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0/go.mod h1:2ahgl5bI9+fMCD+dBC785Lak38Tb7ApdTe5I8a09Qp0= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 h1:NJdGFhzT6zMaTod4QkBqVD2sg0I25iw1boOYtTpEwRo= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305/go.mod h1:qSTSwX3cBP3FKQwQacdjArqv0g6QnukjV4XuzO6UyoY= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 h1:B355/rV/lwpZl3C5iVsFQZU7+LeA+5BTTIzTDYlDOrA= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 h1:bnSl5p3mFekSJ6QcbZ1TmHn2ffYiX8xk6hNzVmyhstQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= github.com/smartcontractkit/chainlink-protos/op-catalog v0.1.0 h1:hGEJFD2X3oNIPXQbtIPxCJyg5CcKglRCYBmESS+gmeQ= diff --git a/integration-tests/load/go.mod b/integration-tests/load/go.mod index 0d2d85116d1..12f549a887e 100644 --- a/integration-tests/load/go.mod +++ b/integration-tests/load/go.mod @@ -24,7 +24,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260624154507-ea7ff77a0ddb github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260623170329-4577ef4ba0ae github.com/smartcontractkit/chainlink-testing-framework/framework v0.16.6-0.20260630120514-36abe27604df @@ -502,6 +502,7 @@ require ( github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36 // indirect github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0 // indirect github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 // indirect + github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 // indirect github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 // indirect github.com/smartcontractkit/chainlink-protos/op-catalog v0.1.0 // indirect github.com/smartcontractkit/chainlink-protos/orchestrator v0.11.0 // indirect diff --git a/integration-tests/load/go.sum b/integration-tests/load/go.sum index c9737b0f1c1..ea97729af9b 100644 --- a/integration-tests/load/go.sum +++ b/integration-tests/load/go.sum @@ -1634,8 +1634,8 @@ github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094 h github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094/go.mod h1:3XQmy2zQHrmufebP7dM+x595+gghxzyIKxK9wv91Rh8= github.com/smartcontractkit/chainlink-ccv/deployment v0.0.2-0.20260616151800-9a3a31c4e194 h1:QxZkbKtQyPtVLYP4eMwc+VbXY7M5ve1deSiLZ2pOA+Y= github.com/smartcontractkit/chainlink-ccv/deployment v0.0.2-0.20260616151800-9a3a31c4e194/go.mod h1:bNMFRxwWdgVFdSsFZRmsUUPoBUncU3RM765K99svIKM= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 h1:ywWOEZYL4DhVwf/TQ5jXsGlx6CzERsYbqd6ov3OS1sc= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0/go.mod h1:ncZiIgraMh4F9lWDoJwB1TX395qZFR5bvnZcHbD0A1Q= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 h1:uvUnQlWgJGJBgtXpxpgGTipUZPxp/JugxBK7KUe41DQ= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6/go.mod h1:2hczeHjBapPah8d3EOoRAfzI4gH3caZToMhAUG8WS3I= 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.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1680,6 +1680,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0 h1:FC+WdJ8Y github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0/go.mod h1:2ahgl5bI9+fMCD+dBC785Lak38Tb7ApdTe5I8a09Qp0= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 h1:NJdGFhzT6zMaTod4QkBqVD2sg0I25iw1boOYtTpEwRo= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305/go.mod h1:qSTSwX3cBP3FKQwQacdjArqv0g6QnukjV4XuzO6UyoY= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 h1:B355/rV/lwpZl3C5iVsFQZU7+LeA+5BTTIzTDYlDOrA= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 h1:bnSl5p3mFekSJ6QcbZ1TmHn2ffYiX8xk6hNzVmyhstQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= github.com/smartcontractkit/chainlink-protos/op-catalog v0.1.0 h1:hGEJFD2X3oNIPXQbtIPxCJyg5CcKglRCYBmESS+gmeQ= diff --git a/plugins/loop_registry.go b/plugins/loop_registry.go index 3968018cdbc..149c74a4baa 100644 --- a/plugins/loop_registry.go +++ b/plugins/loop_registry.go @@ -40,6 +40,7 @@ type LoopRegistry struct { autoPPROF config.AutoPprof cfgTracing config.Tracing cfgTelemetry config.Telemetry + cfgMetering config.Metering telemetryAuthHeaders map[string]string telemetryAuthPubKeyHex string cfgLOOPP config.LOOPP @@ -47,7 +48,7 @@ type LoopRegistry struct { func NewLoopRegistry(lggr logger.Logger, appID string, featureLogPoller bool, dbConfig config.Database, mercury de.Mercury, pyroscope config.Pyroscope, autoPPROF config.AutoPprof, tracing config.Tracing, telemetry config.Telemetry, - telemetryAuthHeaders map[string]string, telemetryAuthPubKeyHex string, looppCfg config.LOOPP) *LoopRegistry { + metering config.Metering, telemetryAuthHeaders map[string]string, telemetryAuthPubKeyHex string, looppCfg config.LOOPP) *LoopRegistry { return &LoopRegistry{ registry: map[string]*RegisteredLoop{}, lggr: logger.Named(lggr, "LoopRegistry"), @@ -59,6 +60,7 @@ func NewLoopRegistry(lggr logger.Logger, appID string, featureLogPoller bool, db autoPPROF: autoPPROF, cfgTracing: tracing, cfgTelemetry: telemetry, + cfgMetering: metering, telemetryAuthHeaders: telemetryAuthHeaders, telemetryAuthPubKeyHex: telemetryAuthPubKeyHex, cfgLOOPP: looppCfg, @@ -166,6 +168,20 @@ func (m *LoopRegistry) Register(id string) (*RegisteredLoop, error) { envCfg.TelemetryPrometheusBridgeEnabled = m.cfgTelemetry.PrometheusBridge().Enabled() envCfg.TelemetryPrometheusBridgePrefixes = m.cfgTelemetry.PrometheusBridge().Prefixes() } + + // Metering config (emission toggles + deployment/node identity dimensions) + // is passed over the env channel to every LOOP plugin, rather than the + // standard-capabilities boundary. + if m.cfgMetering != nil { + envCfg.MeterRecordsEnabled = m.cfgMetering.MeterRecordsEnabled() + envCfg.MeterSnapshotsEnabled = m.cfgMetering.MeterSnapshotsEnabled() + envCfg.MeterProduct = m.cfgMetering.Product() + envCfg.MeterTenant = m.cfgMetering.Tenant() + envCfg.MeterNumericTenantID = m.cfgMetering.NumericTenantID() + envCfg.MeterEnvironment = m.cfgMetering.Environment() + envCfg.MeterZone = m.cfgMetering.Zone() + envCfg.MeterNodeID = m.cfgMetering.NodeID() + } m.lggr.Debugf("Registered loopp %q with port %d", id, envCfg.PrometheusPort) // Add auth header after logging config diff --git a/plugins/loop_registry_test.go b/plugins/loop_registry_test.go index d762dd85c57..ede01b4ed87 100644 --- a/plugins/loop_registry_test.go +++ b/plugins/loop_registry_test.go @@ -92,6 +92,17 @@ func (m mockCfgTelemetry) PrometheusBridge() config.PrometheusBridge { return mockPrometheusBridge{} } +type mockCfgMetering struct{} + +func (m mockCfgMetering) MeterRecordsEnabled() bool { return true } +func (m mockCfgMetering) MeterSnapshotsEnabled() bool { return true } +func (m mockCfgMetering) Product() string { return "cre" } +func (m mockCfgMetering) Tenant() string { return "mainline" } +func (m mockCfgMetering) NumericTenantID() string { return "42" } +func (m mockCfgMetering) Environment() string { return "production" } +func (m mockCfgMetering) Zone() string { return "wf-zone-a" } +func (m mockCfgMetering) NodeID() string { return "csa-pubkey-1" } + type mockPrometheusBridge struct{} func (m mockPrometheusBridge) Enabled() bool { return true } @@ -185,6 +196,7 @@ func TestLoopRegistry_Register(t *testing.T) { mockCfgMercury := &mockCfgMercury{} mockCfgTracing := &mockCfgTracing{} mockCfgTelemetry := &mockCfgTelemetry{} + mockCfgMetering := &mockCfgMetering{} registry := make(map[string]*RegisteredLoop) // Create a LoopRegistry instance with mockCfgTracing @@ -197,6 +209,7 @@ func TestLoopRegistry_Register(t *testing.T) { cfgMercury: mockCfgMercury, cfgTracing: mockCfgTracing, cfgTelemetry: mockCfgTelemetry, + cfgMetering: mockCfgMetering, } // Test case 1: Register new loop @@ -250,6 +263,14 @@ func TestLoopRegistry_Register(t *testing.T) { require.Equal(t, 512, envCfg.TelemetryLogExportMaxBatchSize) require.Equal(t, 5*time.Second, envCfg.TelemetryLogExportInterval) require.Equal(t, 2048, envCfg.TelemetryLogMaxQueueSize) + require.True(t, envCfg.MeterRecordsEnabled) + require.True(t, envCfg.MeterSnapshotsEnabled) + require.Equal(t, "cre", envCfg.MeterProduct) + require.Equal(t, "mainline", envCfg.MeterTenant) + require.Equal(t, "42", envCfg.MeterNumericTenantID) + require.Equal(t, "production", envCfg.MeterEnvironment) + require.Equal(t, "wf-zone-a", envCfg.MeterZone) + require.Equal(t, "csa-pubkey-1", envCfg.MeterNodeID) require.Equal(t, "example.com/chip-ingress", envCfg.ChipIngressEndpoint) require.False(t, envCfg.ChipIngressBatchEmitterEnabled) diff --git a/plugins/plugins.private.yaml b/plugins/plugins.private.yaml index 2882d359901..4a531844367 100644 --- a/plugins/plugins.private.yaml +++ b/plugins/plugins.private.yaml @@ -7,6 +7,44 @@ defaults: goflags: "-ldflags=-s" plugins: + cron: + - moduleURI: "github.com/smartcontractkit/capabilities/cron" + gitRef: "38038ef2e97d15e74491df228460f182c8cb81a5" + installPath: "." + flags: "-tags timetzdata" + readcontract: + - moduleURI: "github.com/smartcontractkit/capabilities/readcontract" + gitRef: "3fad562d65996bf447372023b5123c0c6e8850c7" + installPath: "." + consensus: + - moduleURI: "github.com/smartcontractkit/capabilities/consensus" + gitRef: "69e882e0f24cf9ae897366c4dcbbd3c7e035ae33" + installPath: "." + workflowevent: + - enabled: false + moduleURI: "github.com/smartcontractkit/capabilities/workflowevent" + gitRef: "69e882e0f24cf9ae897366c4dcbbd3c7e035ae33" + installPath: "." + httpaction: + - moduleURI: "github.com/smartcontractkit/capabilities/http_action" + gitRef: "69e882e0f24cf9ae897366c4dcbbd3c7e035ae33" + installPath: "." + httptrigger: + - moduleURI: "github.com/smartcontractkit/capabilities/http_trigger" + gitRef: "38038ef2e97d15e74491df228460f182c8cb81a5" + installPath: "." + evm: + - moduleURI: "github.com/smartcontractkit/capabilities/chain_capabilities/evm" + gitRef: "38038ef2e97d15e74491df228460f182c8cb81a5" + installPath: "." + solana: + - moduleURI: "github.com/smartcontractkit/capabilities/chain_capabilities/solana" + gitRef: "3fccd7ffaa5662ce63d1cf081a9d65a9cd5a58d0" + installPath: "." + aptos: + - moduleURI: "github.com/smartcontractkit/capabilities/chain_capabilities/aptos" + gitRef: "3fad562d65996bf447372023b5123c0c6e8850c7" + installPath: "." confidential-http: - moduleURI: "github.com/smartcontractkit/confidential-compute/enclave/apps/confidential-http/capability" gitRef: "8812c6483a6e6cded22b3fe9615cf1d18b054fd5" diff --git a/system-tests/lib/go.mod b/system-tests/lib/go.mod index 6cb2227f59b..c6a8366bfd4 100644 --- a/system-tests/lib/go.mod +++ b/system-tests/lib/go.mod @@ -37,7 +37,7 @@ require ( github.com/smartcontractkit/chain-selectors v1.0.104 github.com/smartcontractkit/chainlink-aptos v0.0.0-20260706100550-d43558069754 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 github.com/smartcontractkit/chainlink-common/keystore v1.2.0 github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260623170329-4577ef4ba0ae @@ -473,6 +473,7 @@ require ( github.com/smartcontractkit/chainlink-protos/chainlink-ccv/message-rules v0.0.0-20260505131349-78e491b80735 // indirect github.com/smartcontractkit/chainlink-protos/chainlink-ccv/verifier v0.0.0-20251211142334-5c3421fe2c8d // indirect github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36 // indirect + github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 // indirect github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 // indirect github.com/smartcontractkit/chainlink-protos/orchestrator v0.11.0 // indirect github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd // indirect diff --git a/system-tests/lib/go.sum b/system-tests/lib/go.sum index b2c2cff2960..0829f3bbc41 100644 --- a/system-tests/lib/go.sum +++ b/system-tests/lib/go.sum @@ -1545,8 +1545,8 @@ github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260 github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb/go.mod h1:67YbnoglYD61Pz/jTVCgav9wFq7S35OU8UyQSvPllRw= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094 h1:uMn1w/n95p+pSxK6hYNqji/sDaab8D0Cxuph9qUkM2g= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094/go.mod h1:3XQmy2zQHrmufebP7dM+x595+gghxzyIKxK9wv91Rh8= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 h1:ywWOEZYL4DhVwf/TQ5jXsGlx6CzERsYbqd6ov3OS1sc= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0/go.mod h1:ncZiIgraMh4F9lWDoJwB1TX395qZFR5bvnZcHbD0A1Q= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 h1:uvUnQlWgJGJBgtXpxpgGTipUZPxp/JugxBK7KUe41DQ= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6/go.mod h1:2hczeHjBapPah8d3EOoRAfzI4gH3caZToMhAUG8WS3I= 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.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1591,6 +1591,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0 h1:FC+WdJ8Y github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0/go.mod h1:2ahgl5bI9+fMCD+dBC785Lak38Tb7ApdTe5I8a09Qp0= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 h1:NJdGFhzT6zMaTod4QkBqVD2sg0I25iw1boOYtTpEwRo= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305/go.mod h1:qSTSwX3cBP3FKQwQacdjArqv0g6QnukjV4XuzO6UyoY= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 h1:B355/rV/lwpZl3C5iVsFQZU7+LeA+5BTTIzTDYlDOrA= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 h1:bnSl5p3mFekSJ6QcbZ1TmHn2ffYiX8xk6hNzVmyhstQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= github.com/smartcontractkit/chainlink-protos/op-catalog v0.1.0 h1:hGEJFD2X3oNIPXQbtIPxCJyg5CcKglRCYBmESS+gmeQ= diff --git a/system-tests/tests/go.mod b/system-tests/tests/go.mod index 85dcd38638e..d1b522c2d62 100644 --- a/system-tests/tests/go.mod +++ b/system-tests/tests/go.mod @@ -62,7 +62,7 @@ require ( github.com/rs/zerolog v1.35.1 github.com/smartcontractkit/chain-selectors v1.0.104 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 github.com/smartcontractkit/chainlink-common/keystore v1.2.0 github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 @@ -165,6 +165,7 @@ require ( github.com/smartcontractkit/chainlink-protos/chainlink-ccv/verifier v0.0.0-20251211142334-5c3421fe2c8d // indirect github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36 // indirect github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0 // indirect + github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 // indirect github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 // indirect github.com/smartcontractkit/chainlink-solana/contracts v0.0.0-20260513123719-d347eaf314e1 // indirect github.com/smartcontractkit/chainlink-sui/codec v0.0.0-20260706170510-94f3a0409ea5 // indirect diff --git a/system-tests/tests/go.sum b/system-tests/tests/go.sum index 3c65ac57379..7895b293a88 100644 --- a/system-tests/tests/go.sum +++ b/system-tests/tests/go.sum @@ -1559,8 +1559,8 @@ github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260 github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb/go.mod h1:67YbnoglYD61Pz/jTVCgav9wFq7S35OU8UyQSvPllRw= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094 h1:uMn1w/n95p+pSxK6hYNqji/sDaab8D0Cxuph9qUkM2g= github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260706093831-dad26b360094/go.mod h1:3XQmy2zQHrmufebP7dM+x595+gghxzyIKxK9wv91Rh8= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0 h1:ywWOEZYL4DhVwf/TQ5jXsGlx6CzERsYbqd6ov3OS1sc= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0/go.mod h1:ncZiIgraMh4F9lWDoJwB1TX395qZFR5bvnZcHbD0A1Q= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6 h1:uvUnQlWgJGJBgtXpxpgGTipUZPxp/JugxBK7KUe41DQ= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6/go.mod h1:2hczeHjBapPah8d3EOoRAfzI4gH3caZToMhAUG8WS3I= 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.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1605,6 +1605,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0 h1:FC+WdJ8Y github.com/smartcontractkit/chainlink-protos/job-distributor v0.19.0/go.mod h1:2ahgl5bI9+fMCD+dBC785Lak38Tb7ApdTe5I8a09Qp0= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 h1:NJdGFhzT6zMaTod4QkBqVD2sg0I25iw1boOYtTpEwRo= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305/go.mod h1:qSTSwX3cBP3FKQwQacdjArqv0g6QnukjV4XuzO6UyoY= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 h1:B355/rV/lwpZl3C5iVsFQZU7+LeA+5BTTIzTDYlDOrA= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 h1:bnSl5p3mFekSJ6QcbZ1TmHn2ffYiX8xk6hNzVmyhstQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= github.com/smartcontractkit/chainlink-protos/op-catalog v0.1.0 h1:hGEJFD2X3oNIPXQbtIPxCJyg5CcKglRCYBmESS+gmeQ= diff --git a/testdata/scripts/config/merge_raw_configs.txtar b/testdata/scripts/config/merge_raw_configs.txtar index 57977a1f485..3b74face983 100644 --- a/testdata/scripts/config/merge_raw_configs.txtar +++ b/testdata/scripts/config/merge_raw_configs.txtar @@ -511,6 +511,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/testdata/scripts/node/validate/default.txtar b/testdata/scripts/node/validate/default.txtar index c0a1abab84b..9e0f05c374c 100644 --- a/testdata/scripts/node/validate/default.txtar +++ b/testdata/scripts/node/validate/default.txtar @@ -376,6 +376,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/testdata/scripts/node/validate/defaults-override.txtar b/testdata/scripts/node/validate/defaults-override.txtar index 86086a4fca2..5e7958e30a7 100644 --- a/testdata/scripts/node/validate/defaults-override.txtar +++ b/testdata/scripts/node/validate/defaults-override.txtar @@ -437,6 +437,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/testdata/scripts/node/validate/disk-based-logging-disabled.txtar b/testdata/scripts/node/validate/disk-based-logging-disabled.txtar index 36c2900e57f..a6aec07ebde 100644 --- a/testdata/scripts/node/validate/disk-based-logging-disabled.txtar +++ b/testdata/scripts/node/validate/disk-based-logging-disabled.txtar @@ -420,6 +420,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 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 9fb002e0af8..47e3d90fc7a 100644 --- a/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar +++ b/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar @@ -420,6 +420,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/testdata/scripts/node/validate/disk-based-logging.txtar b/testdata/scripts/node/validate/disk-based-logging.txtar index dfdb9b48384..37ed971264b 100644 --- a/testdata/scripts/node/validate/disk-based-logging.txtar +++ b/testdata/scripts/node/validate/disk-based-logging.txtar @@ -420,6 +420,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/testdata/scripts/node/validate/fallback-override.txtar b/testdata/scripts/node/validate/fallback-override.txtar index c2320cbce92..629b75c1c10 100644 --- a/testdata/scripts/node/validate/fallback-override.txtar +++ b/testdata/scripts/node/validate/fallback-override.txtar @@ -522,6 +522,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/testdata/scripts/node/validate/invalid-ocr-p2p.txtar b/testdata/scripts/node/validate/invalid-ocr-p2p.txtar index 5e7eaaa1bad..488f4199388 100644 --- a/testdata/scripts/node/validate/invalid-ocr-p2p.txtar +++ b/testdata/scripts/node/validate/invalid-ocr-p2p.txtar @@ -405,6 +405,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/testdata/scripts/node/validate/invalid.txtar b/testdata/scripts/node/validate/invalid.txtar index aae5e8883ca..cec424c990c 100644 --- a/testdata/scripts/node/validate/invalid.txtar +++ b/testdata/scripts/node/validate/invalid.txtar @@ -416,6 +416,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/testdata/scripts/node/validate/valid.txtar b/testdata/scripts/node/validate/valid.txtar index 9b0595833cb..c8b9da305ce 100644 --- a/testdata/scripts/node/validate/valid.txtar +++ b/testdata/scripts/node/validate/valid.txtar @@ -417,6 +417,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/testdata/scripts/node/validate/warnings.txtar b/testdata/scripts/node/validate/warnings.txtar index 6e000e852ba..32598b0d36e 100644 --- a/testdata/scripts/node/validate/warnings.txtar +++ b/testdata/scripts/node/validate/warnings.txtar @@ -399,6 +399,14 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200