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 fafb339e6a9..a24424881c9 100644 --- a/core/config/docs/core.toml +++ b/core/config/docs/core.toml @@ -929,6 +929,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. +[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 = 'cre' # 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/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 1fbd21c1bc8..14b4be5cc48 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) @@ -3096,6 +3098,68 @@ 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) + 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 d45de59febb..7ded20dcc73 100644 --- a/core/scripts/go.mod +++ b/core/scripts/go.mod @@ -47,7 +47,7 @@ require ( github.com/smartcontractkit/chain-selectors v1.0.106 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.20260715145851-8219609496a4 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260707105132-2da5d31f22fc github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 @@ -508,6 +508,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-20260710151514-27b5a126dabe // indirect github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 // 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 92af015246e..70c38662439 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.1.1-0.20260715132147-59df1b138100 h1:sH546tdm3VaYGaRLnZ6ZYBZiE25vEkMap76Xq/vQWx0= github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260715132147-59df1b138100/go.mod h1:0v6RGdYa9NezVnBPIAVyxB3A9furKWwaSkLha+URYGk= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4 h1:LzDSHQ5bM4pp8Wrr0mFjg60A4J47+IFPIuC/rHNF1qA= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4/go.mod h1:snfVBRRQTpC2x5O3bQHZe9SvJX5yv/SbG8oHkJTKLtE= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba h1:1sx4ZrmoGiBOa0obx8K7VTojaODpXB67KGTfqLObjss= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba/go.mod h1:tAsXbMsNrwIp7vCqeauNsT4f3gVATbTmIviCkZPDgdo= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 h1:DvfhsiIxB4JMuR+r1UciKV8jKiwhI/OZI/QhKawC4pM= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6/go.mod h1:6NefaCIMH4zFBN3T+cvsFGYoy3oTSPKDaxPAyBzlYTU= 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.20.1-0.202607011 github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.20260701185448-696c075849ea/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-20260710151514-27b5a126dabe h1:MDnY5wQbWTpFdDnMRicEnoMfSP5nM/KncARr4skP1ug= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 h1:/vkKPJoweLkRd56V4YHGRAtTG4+/JAlgklGEfvH6l4c= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16/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 aadc1aae3df..87e9d0fe25d 100644 --- a/core/services/chainlink/application.go +++ b/core/services/chainlink/application.go @@ -290,7 +290,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, 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..632b2b1fa3b --- /dev/null +++ b/core/services/chainlink/config_metering.go @@ -0,0 +1,67 @@ +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 +} + +// Product defaults to "unset" (resourcemanager.DefaultMeteringProduct) when unset, +// to catch config errors early. +func (b *meteringConfig) Product() string { + if b.s.Product == nil { + return "unset" + } + 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..42f64260f8c --- /dev/null +++ b/core/services/chainlink/config_metering_test.go @@ -0,0 +1,50 @@ +package chainlink + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/smartcontractkit/chainlink/v2/core/config/toml" +) + +func TestMeteringConfig(t *testing.T) { + t.Parallel() + t.Run("defaults", func(t *testing.T) { + t.Parallel() + mc := meteringConfig{s: toml.Metering{}} + assert.False(t, mc.MeterRecordsEnabled()) + assert.False(t, mc.MeterSnapshotsEnabled()) + // Product defaults to "cre" so metering can never be enabled with an + // empty product dimension and the syncer identity matches the plugins' + // fallback. + assert.Equal(t, "unset", 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) { + t.Parallel() + mc := meteringConfig{s: toml.Metering{ + MeterRecordsEnabled: new(true), + MeterSnapshotsEnabled: new(true), + Product: new("cre"), + Tenant: new("mainline"), + NumericTenantID: new("42"), + Environment: new("production"), + Zone: new("wf-zone-a"), + NodeID: new("clp-cre-wf-zone-a-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, "clp-cre-wf-zone-a-1", mc.NodeID()) + }) +} diff --git a/core/services/chainlink/config_test.go b/core/services/chainlink/config_test.go index 97a7d484250..8e8657281be 100644 --- a/core/services/chainlink/config_test.go +++ b/core/services/chainlink/config_test.go @@ -581,6 +581,16 @@ func TestConfig_Marshal(t *testing.T) { Prefixes: []string{"ocr_"}, }, } + full.Metering = toml.Metering{ + MeterRecordsEnabled: new(true), + MeterSnapshotsEnabled: new(true), + Product: new("cre"), + Tenant: new("mainline"), + NumericTenantID: new("42"), + Environment: new("production"), + Zone: new("wf-zone-a"), + NodeID: new("clp-cre-wf-zone-a-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 fb6219ca3fc..4d72c6bbb57 100644 --- a/core/services/chainlink/mocks/general_config.go +++ b/core/services/chainlink/mocks/general_config.go @@ -2604,6 +2604,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 2bf0926bf4a..7d607891451 100644 --- a/core/services/chainlink/testdata/config-empty-effective.toml +++ b/core/services/chainlink/testdata/config-empty-effective.toml @@ -367,6 +367,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'unset' +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 0b2444932c7..5dc3a9b14f5 100644 --- a/core/services/chainlink/testdata/config-full.toml +++ b/core/services/chainlink/testdata/config-full.toml @@ -406,6 +406,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 = 'clp-cre-wf-zone-a-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 45d7d4e2121..55f9cbe7314 100644 --- a/core/services/chainlink/testdata/config-multi-chain-effective.toml +++ b/core/services/chainlink/testdata/config-multi-chain-effective.toml @@ -367,6 +367,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'unset' +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..10095699655 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" @@ -297,6 +299,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 +316,7 @@ func (s *Services) newSubservices( opts.LimitsFactory, s.OrgResolver, s.GatewayConnectorWrapper, + meterIdentity, ) if err != nil { return nil, err @@ -338,6 +345,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 +727,30 @@ 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 +} + func newShardOrchestratorClient(cfg Config, lggr logger.Logger) (*shardorchestrator.Client, error) { shardID := cfg.Sharding().ShardIndex() if shardID == 0 { @@ -933,6 +966,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() @@ -1009,6 +1043,10 @@ func newWorkflowRegistrySyncerV2( shardRoutingSteady = shardownership.NewSteadySignal(shardownership.WithSteadySignalMetrics(steadyMetrics)) } + meteringCfg := cfg.Metering() + meterRecordsEnabled := meteringCfg != nil && meteringCfg.MeterRecordsEnabled() + meterSnapshotsEnabled := meteringCfg != nil && meteringCfg.MeterSnapshotsEnabled() + handlerOpts := []syncerV2.EventHandlerOption{ syncerV2.WithBillingClient(billingClient), syncerV2.WithWorkflowRegistry(capCfg.WorkflowRegistry().Address(), selector), @@ -1017,6 +1055,13 @@ func newWorkflowRegistrySyncerV2( syncerV2.WithLocalSecretOverrides(lggr, cfg.CRE().LocalSecretOverrides()), syncerV2.WithShardExecutionGuard(shardOrchestratorClient, shardingEnabled, shardIndex), syncerV2.WithShardRoutingSteady(shardRoutingSteady), + syncerV2.WithResourceManager(resourcemanager.NewResourceManager(lggr, resourcemanager.ResourceManagerConfig{ + MeterRecordsEnabled: meterRecordsEnabled, + MeterSnapshotsEnabled: meterSnapshotsEnabled, + Emitter: beholder.GetEmitter(), + SnapshotInterval: resourcemanager.DefaultSnapshotInterval, + })), + syncerV2.WithIdentity(meterIdentity), } mc := capCfg.WorkflowRegistry().ModuleCache() @@ -1161,6 +1206,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 +1255,7 @@ func newWorkflowRegistrySyncer( lf, orgResolver, gatewayConnectorWrapper, + meterIdentity, ) return syncer, billingClient, srvcs, err default: diff --git a/core/services/standardcapabilities/standard_capabilities.go b/core/services/standardcapabilities/standard_capabilities.go index 3d7e7f8032c..c9ec5d99a63 100644 --- a/core/services/standardcapabilities/standard_capabilities.go +++ b/core/services/standardcapabilities/standard_capabilities.go @@ -82,6 +82,24 @@ 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()) @@ -119,19 +137,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..72094b5b172 100644 --- a/core/services/standardcapabilities/standard_capabilities_test.go +++ b/core/services/standardcapabilities/standard_capabilities_test.go @@ -80,7 +80,7 @@ func TestStandardCapabilities_ForwardsPluginEnvFile(t *testing.T) { require.Contains(t, err.Error(), capturingRegistrarErr) require.Empty(t, cfg.Env, - "no operator-provided env vars should be forwarded when CL_CAPABILITIES_ENV is unset") + "no operator-provided env vars should be forwarded when CL_CAPABILITIES_ENV is unset; ") }) t.Run("missing env file fails Start before RegisterLOOP", func(t *testing.T) { @@ -106,6 +106,66 @@ func TestStandardCapabilities_ForwardsPluginEnvFile(t *testing.T) { }) } +func TestStandardCapabilities_InitialiseDependenciesRoundTrip(t *testing.T) { + t.Parallel() + 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") +} + +// TestStandardCapabilities_CapabilityDonIDDeliveredToLOOP asserts that the +// host-resolved capability DON ID is carried on the dependencies delivered to +// the capability LOOP at Initialise. Without this, a trigger producer silently +// falls back to the consumer workflow's DON for metering identity and event +// labels even when the node knows which DON it is serving. +func TestStandardCapabilities_CapabilityDonIDDeliveredToLOOP(t *testing.T) { + t.Parallel() + t.Run("nonzero DON ID round-trips when the DON is known", func(t *testing.T) { + t.Parallel() + const knownDonID = uint32(42) + std := NewStandardCapabilities( + logger.TestLogger(t), + "not/found/path/to/binary", + "{}", + &capturingRegistrar{}, + core.StandardCapabilitiesDependencies{CapabilityDonID: knownDonID}, + ) + + got := std.initialiseDependencies() + + require.Equal(t, knownDonID, got.CapabilityDonID, + "the host-resolved capability DON ID must reach the LOOP at Initialise") + }) + + t.Run("zero DON ID is preserved when the host could not resolve one", func(t *testing.T) { + t.Parallel() + std := NewStandardCapabilities( + logger.TestLogger(t), + "not/found/path/to/binary", + "{}", + &capturingRegistrar{}, + core.StandardCapabilitiesDependencies{}, + ) + + got := std.initialiseDependencies() + + require.Zero(t, got.CapabilityDonID, + "an unresolved DON ID stays zero so the LOOP can fall back to the workflow DON") + }) +} + const capturingRegistrarErr = "capturingRegistrar: stop after capture" // capturingRegistrar records the CmdConfig handed to RegisterLOOP and then diff --git a/core/services/workflows/artifacts/v2/orm.go b/core/services/workflows/artifacts/v2/orm.go index 31395bce8f3..a9467a3cd59 100644 --- a/core/services/workflows/artifacts/v2/orm.go +++ b/core/services/workflows/artifacts/v2/orm.go @@ -20,6 +20,11 @@ type WorkflowSpecsDS interface { // GetWorkflowSpecByID returns the workflow spec for the given workflowID. GetWorkflowSpec(ctx context.Context, id string) (*job.WorkflowSpec, error) + // GetWorkflowSpecList returns the persisted workflow specs. It projects only + // the identity columns needed by the metering snapshot path (workflow_id and + // workflow_owner); other fields are left zero. + GetWorkflowSpecList(ctx context.Context) ([]*job.WorkflowSpec, error) + // DeleteWorkflowSpec deletes the workflow spec for the given workflow ID. DeleteWorkflowSpec(ctx context.Context, id string) error @@ -129,6 +134,22 @@ func (orm *orm) GetWorkflowSpec(ctx context.Context, id string) (*job.WorkflowSp return &spec, nil } +// GetWorkflowSpecList returns all persisted workflow specs, projecting only the +// identity columns the metering snapshot path needs (workflow_id, workflow_owner). +func (orm *orm) GetWorkflowSpecList(ctx context.Context) ([]*job.WorkflowSpec, error) { + query := ` + SELECT workflow_id, workflow_owner + FROM workflow_specs_v2 + ` + + var specs []*job.WorkflowSpec + if err := orm.ds.SelectContext(ctx, &specs, query); err != nil { + return nil, err + } + + return specs, nil +} + func (orm *orm) DeleteWorkflowSpec(ctx context.Context, id string) error { query := ` DELETE FROM workflow_specs_v2 diff --git a/core/services/workflows/artifacts/v2/store.go b/core/services/workflows/artifacts/v2/store.go index 0eeb694cc6a..deb27c77632 100644 --- a/core/services/workflows/artifacts/v2/store.go +++ b/core/services/workflows/artifacts/v2/store.go @@ -234,6 +234,12 @@ func (h *Store) GetWorkflowSpec(ctx context.Context, workflowID string) (*job.Wo return spec, err } +// GetWorkflowSpecList returns the persisted workflow specs (identity columns +// only). It backs the metering snapshot path. +func (h *Store) GetWorkflowSpecList(ctx context.Context) ([]*job.WorkflowSpec, error) { + return h.orm.GetWorkflowSpecList(ctx) +} + func (h *Store) UpsertWorkflowSpec(ctx context.Context, spec *job.WorkflowSpec) (int64, error) { return h.orm.UpsertWorkflowSpec(ctx, spec) } 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 c7eb74bba53..85c160cbca3 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,19 @@ type eventHandler struct { workflowEncryptionKey workflowkey.Key workflowDonSubscriber capabilities.DonSubscriber billingClient metering.BillingClient - orgResolver orgresolver.OrgResolver - secretsFetcher v2.SecretsFetcher + resourceManager *resourcemanager.ResourceManager + 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 +180,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 @@ -253,6 +301,10 @@ func WithModuleEngineVersion(v string) func(*eventHandler) { type WorkflowArtifactsStore interface { FetchWorkflowArtifacts(ctx context.Context, workflowID, binaryIdentifier, configIdentifier string) ([]byte, []byte, error) GetWorkflowSpec(ctx context.Context, workflowID string) (*job.WorkflowSpec, error) + // GetWorkflowSpecList returns the persisted workflow specs. It backs the + // metering snapshot path (Meterable.GetUtilization), which enumerates the + // durable specs rather than running engines. + GetWorkflowSpecList(ctx context.Context) ([]*job.WorkflowSpec, error) UpsertWorkflowSpec(ctx context.Context, spec *job.WorkflowSpec) (int64, error) DeleteWorkflowArtifacts(ctx context.Context, workflowID string) error DeleteWorkflowArtifactsBatch(ctx context.Context, workflowIDs []string) error @@ -306,7 +358,15 @@ func NewEventHandler( workflowArtifactsStore: workflowArtifacts, workflowEncryptionKey: workflowEncryptionKey, workflowDonSubscriber: workflowDonSubscriber, - tracer: noop.NewTracerProvider().Tracer(""), // default to noop, enable via WithDebugMode + // resourceManager defaults to nil (metering off); it is set only via + // WithResourceManager. A nil manager keeps every emission site guarded + // and avoids registering duplicate OTel metering instruments. + // 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, + }, } metricsInst, metricsErr := newMetrics() if metricsErr != nil { @@ -327,20 +387,73 @@ 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) + } + // Register returns a func that unregisters. + 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) + // No metering is emitted on close: meter records anchor on workflow-spec + // storage transitions, not engine lifecycle, so stopping an engine at + // shutdown leaves the persisted spec (and therefore its metered level) + // untouched. A spec that is genuinely released stops being snapshotted. + // 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 +613,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 +637,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 +645,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 +674,10 @@ func (h *eventHandler) workflowRegisteredEvent( if innerErr != nil { return innerErr } + // A new spec's artifacts were persisted for the first time: the durable + // workflow_specs_v2 resource gained one unit, so emit a +1 delta. + h.emitSpecDelta(ctx, 1, payload.WorkflowID.Hex(), hex.EncodeToString(payload.WorkflowOwner), + resourcemanager.EventID("workflow-spec-register", payload.WorkflowID.Hex(), strconv.FormatUint(payload.CreatedAt, 10))) spec = newSpec case spec.WorkflowID != payload.WorkflowID.Hex(): @@ -564,6 +685,10 @@ func (h *eventHandler) workflowRegisteredEvent( if innerErr != nil { return innerErr } + // A different spec's artifacts were persisted under this key: the newly + // stored spec is a fresh durable resource, so emit a +1 delta. + h.emitSpecDelta(ctx, 1, payload.WorkflowID.Hex(), hex.EncodeToString(payload.WorkflowOwner), + resourcemanager.EventID("workflow-spec-register", payload.WorkflowID.Hex(), strconv.FormatUint(payload.CreatedAt, 10))) spec = newSpec case spec.Status != status: @@ -571,6 +696,9 @@ func (h *eventHandler) workflowRegisteredEvent( if _, innerErr := h.workflowArtifactsStore.UpsertWorkflowSpec(ctx, spec); innerErr != nil { return fmt.Errorf("failed to update workflow spec: %w", innerErr) } + // Status-only update (e.g. activate an already-stored spec): the spec + // stays stored, so there is no artifact-persistence transition and no + // meter delta. The level is unchanged and remains covered by snapshots. } // Next, let's synchronize the engine. @@ -820,13 +948,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 @@ -834,6 +966,16 @@ func (h *eventHandler) workflowDeletedEvent( // At the same time, popping the engine should occur last to allow deletes to be retried if any of the // prior steps fail. workflowID := payload.WorkflowID.Hex() + // Capture the stored owner before deletion so a real-delete -1 delta can + // resolve org attribution from durable state. Only needed for actual + // deletions (pause emits no delta); fail-open, an empty owner yields an + // empty org. + var deleteOwner string + if originatingEvent == WorkflowDeleted { + if existing, gerr := h.workflowArtifactsStore.GetWorkflowSpec(ctx, workflowID); gerr == nil && existing != nil { + deleteOwner = existing.WorkflowOwner + } + } e, ok := h.engineRegistry.Get(payload.WorkflowID) var drainable DrainableService var isDrainable bool @@ -865,6 +1007,18 @@ 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) } + // A real delete removed the persisted spec, so emit a -1 delta. Pause is + // delegated here too (originatingEvent == WorkflowPaused); its release is + // realized by the spec's absence from subsequent snapshots, so pause emits + // no delta. + if originatingEvent == WorkflowDeleted { + // WorkflowDeletedEvent carries only the workflow ID (no on-chain + // CreatedAt/block), so the delete event_id is derived from the + // DON-consistent workflowID alone. See the emitSpecDelta call for the + // re-delete-cycle limitation noted in the metering redesign. + h.emitSpecDelta(ctx, -1, workflowID, deleteOwner, + resourcemanager.EventID("workflow-spec-delete", workflowID)) + } h.cleanupModuleCache(payload.WorkflowID.Hex()) @@ -885,6 +1039,115 @@ func (h *eventHandler) workflowDeletedEvent( return nil } +// emitSpecDelta emits one metering.v1.MeterRecord (METER_ACTION_UPDATE) +// capturing a signed ±delta to the durable workflow_specs_v2 level for +// workflowID. Each record is a first-class billing event: a +1 when a new +// spec's artifacts are persisted, a -1 when a spec is deleted. It is the single +// emission helper for every syncer meter record; callers invoke it only after +// the workflow-spec storage mutation has committed. +// +// owner is the workflow owner as stored in durable state; the organization is +// resolved fail-open at emit time and is never persisted. +// Emission is fail-open and returns no error, so metering can never gate, +// delay, or fail the storage operation it records. +// +// eventID is the caller-supplied, deterministic cross-node dedup key. The syncer +// is a workflow-DON service driven by reconciliation against the shared on-chain +// WorkflowRegistry, so every workflow-DON node sees the same reconciliation +// event and derives the identical id from it (see the call sites). It is never +// generated here; an empty eventID is rejected by the ResourceManager. +func (h *eventHandler) emitSpecDelta(ctx context.Context, delta int64, workflowID, owner, eventID string) { + if h.resourceManager == nil { + return + } + var orgID string + if h.orgResolver != nil && owner != "" { + if resolved, err := h.orgResolver.Get(ctx, owner); err != nil { + h.lggr.Warnw("failed to resolve org ID for metering", "owner", owner, "err", err) + } else { + orgID = resolved + } + } + // resource_id = workflow_id (the syncer meters one durable spec per workflow). + h.resourceManager.EmitDelta(ctx, h.baseIdentity(), eventID, delta, resourcemanager.UtilizationFields{ + ResourceType: meterResourceType, + ResourceID: workflowID, + OrgID: orgID, + }) +} + +// 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 persisted workflow_specs_v2 spec, each at level 1, so the +// periodic absolute-state snapshot agrees with the +1/-1 deltas emitted on +// artifact-persistence transitions. The durable resource is the stored spec, +// not the running engine, so this enumerates persisted specs (via the store's +// list) rather than the engine registry — that way paused-but-stored specs and +// specs whose engine failed to start are still accounted for, and a delete is +// reflected by the spec's absence from the next snapshot. +// +// resource_id is the workflow_id; the organization is resolved from the spec's +// stored owner through the caching resolver, which serves from memory so this +// tick stays effectively no-network. Fail-open: a store error yields no entries +// for this tick. +func (h *eventHandler) GetUtilization(ctx context.Context) []resourcemanager.SnapshotEntry { + specs, err := h.workflowArtifactsStore.GetWorkflowSpecList(ctx) + if err != nil { + h.lggr.Warnw("failed to list persisted workflow specs for metering snapshot; skipping tick", "err", err) + return nil + } + base := h.baseIdentity() + entries := make([]resourcemanager.SnapshotEntry, 0, len(specs)) + for _, spec := range specs { + if spec == nil { + continue + } + var orgID string + if h.orgResolver != nil && spec.WorkflowOwner != "" { + if resolved, err := h.orgResolver.Get(ctx, spec.WorkflowOwner); err == nil { + orgID = resolved + } + } + entries = append(entries, resourcemanager.SnapshotEntry{ + Identity: base, + Utilizations: []*meteringpb.Utilization{ + resourcemanager.NewUtilizationInt(1, resourcemanager.UtilizationFields{ + ResourceType: meterResourceType, + ResourceID: spec.WorkflowID, + OrgID: orgID, + }), + }, + }) + } + return entries +} + // 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 +1267,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..6c4ea25f067 --- /dev/null +++ b/core/services/workflows/syncer/v2/handler_metering_test.go @@ -0,0 +1,427 @@ +package v2 + +import ( + "context" + "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" + meteringpb "github.com/smartcontractkit/chainlink-protos/metering/go" + + "github.com/smartcontractkit/chainlink/v2/core/capabilities" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/confidentialrelay" + "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, + &confidentialrelay.ExecutionHandlers{}, + NewEngineRegistry(), + custmsg.NewLabeler(), + limiters, + nil, + rl, + workflowLimits, + artifactsStore, + workflowkey.MustNewXXXTestingOnly(big.NewInt(1)), + &testDonNotifier{}, + WithResourceManager(rm), + WithEngineFactoryFn(mockEngineFactory), + ) + require.NoError(t, err) + return h +} + +// requireSpecDelta asserts that record is a workflow-syncer-v2 spec delta: a +// METER_ACTION_UPDATE carrying a single utilization with the given signed value +// and resource_id (= workflow_id). event_id must be the deterministic, +// cross-node-identical id wantEventID (an opaque string that is never +// format-validated). +func requireSpecDelta(t *testing.T, record *meteringpb.MeterRecord, value, workflowID, wantEventID 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) + // Every syncer meter record is a signed level delta (UPDATE), never a + // RESERVE/RELEASE lifecycle edge. + assert.Equal(t, meteringpb.MeterAction_METER_ACTION_UPDATE, record.Action) + assert.NotNil(t, record.Timestamp) + require.Len(t, record.Utilizations, 1) + util := record.Utilizations[0] + assert.Equal(t, value, 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) + // event_id is the deterministic reconciliation-derived id, identical on every + // workflow-DON node; it is an opaque string with no validated format. + assert.Equal(t, wantEventID, util.EventId) +} + +func Test_meterRecords(t *testing.T) { + t.Parallel() + + wfOwner := []byte{0xaa, 0xbb, 0xcc, 0xdd} + + t.Run("registered event persisting a new spec emits +1", 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) + requireSpecDelta(t, records[0], "1", wfID.Hex(), + resourcemanager.EventID("workflow-spec-register", wfID.Hex(), "0")) + }) + + t.Run("reprocessed registered event emits the IDENTICAL event_id (cross-node dedup)", 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 (or a second node + // reconciling the same on-chain event) would. + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{}, newMeteringResourceManager(t, true, emitter)) + + event := WorkflowRegisteredEvent{ + Status: WorkflowStatusPaused, + WorkflowID: types.WorkflowID{2}, + WorkflowOwner: wfOwner, + WorkflowName: "wf-name", + CreatedAt: 123, + } + 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) + // The same reconciliation event MUST yield the identical event_id, so the + // billing consumer dedups reprocessing and cross-node duplicates. It is + // derived deterministically from the on-chain workflowID + CreatedAt. + want := resourcemanager.EventID("workflow-spec-register", types.WorkflowID{2}.Hex(), "123") + assert.Equal(t, want, records[0].Utilizations[0].GetEventId()) + assert.Equal(t, records[0].Utilizations[0].GetEventId(), records[1].Utilizations[0].GetEventId()) + }) + + t.Run("activating an already-stored spec emits nothing (status-only update)", 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: "aabbccdd", + }, + }, newMeteringResourceManager(t, true, emitter)) + + // Same workflow ID already stored, only the status changes: the spec + // stays stored, so no artifact-persistence transition and no delta. + err := h.workflowRegisteredEvent(t.Context(), WorkflowRegisteredEvent{ + Status: WorkflowStatusPaused, + WorkflowID: wfID, + WorkflowOwner: wfOwner, + WorkflowName: "wf-name", + }, WorkflowRegistered) + require.NoError(t, err) + + assert.Empty(t, emitter.Records()) + }) + + t.Run("paused event emits nothing", 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: "aabbccdd", + }, + }, newMeteringResourceManager(t, true, emitter)) + + require.NoError(t, h.workflowPausedEvent(t.Context(), WorkflowPausedEvent{WorkflowID: wfID})) + + // Pause routes through the delete path, but the spec's release is + // realized by its absence from subsequent snapshots, not a delta. + assert.Empty(t, emitter.Records()) + }) + + t.Run("deleted event emits -1 after artifacts are deleted", func(t *testing.T) { + t.Parallel() + wfID := types.WorkflowID{5} + emitter := &recordingEmitter{} + h := newMeteringTestHandler(t, &stubWorkflowArtifactsStore{ + spec: &job.WorkflowSpec{ + WorkflowID: wfID.Hex(), + Status: job.WorkflowSpecStatusActive, + WorkflowOwner: "aabbccdd", + }, + }, newMeteringResourceManager(t, true, emitter)) + + require.NoError(t, h.workflowDeletedEvent(t.Context(), WorkflowDeletedEvent{WorkflowID: wfID}, WorkflowDeleted)) + + records := emitter.Records() + require.Len(t, records, 1) + requireSpecDelta(t, records[0], "-1", wfID.Hex(), + resourcemanager.EventID("workflow-spec-delete", wfID.Hex())) + }) + + t.Run("no record when persisting 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{6}, + 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: "aabbccdd", + }, + 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: "aabbccdd", + }, + }, 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) + requireSpecDelta(t, records[0], "-1", wfID.Hex(), + resourcemanager.EventID("workflow-spec-delete", 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: "aabbccdd", + }, + }, 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 persisted spec", 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, + }) + + // The snapshot enumerates persisted specs from the store, not running + // engines, so paused-but-stored specs are still accounted for. + wfID1 := types.WorkflowID{20} + wfID2 := types.WorkflowID{21} + store := &stubWorkflowArtifactsStore{ + specs: []*job.WorkflowSpec{ + {WorkflowID: wfID1.Hex(), WorkflowOwner: "aabbccdd"}, + {WorkflowID: wfID2.Hex(), WorkflowOwner: "aabbccdd"}, + }, + } + h := newMeteringTestHandler(t, store, rm) + unregister := rm.Register(h) + t.Cleanup(unregister) + + 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 persisted spec") + require.NotNil(t, byWorkflowID[wfID2.Hex()], "snapshot must contain an entry for the second persisted spec") + }) +} diff --git a/core/services/workflows/syncer/v2/handler_test.go b/core/services/workflows/syncer/v2/handler_test.go index d4dde4c6b7d..e72c708bf1d 100644 --- a/core/services/workflows/syncer/v2/handler_test.go +++ b/core/services/workflows/syncer/v2/handler_test.go @@ -245,7 +245,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 @@ -387,7 +387,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) }, @@ -425,7 +425,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) }, }, @@ -464,7 +464,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) @@ -503,7 +503,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 @@ -566,7 +566,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 @@ -792,7 +792,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) @@ -881,7 +881,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) @@ -1023,6 +1023,10 @@ func (m *mockArtifactStore) DeleteWorkflowArtifacts(ctx context.Context, workflo return m.artifactStore.DeleteWorkflowArtifacts(ctx, workflowID) } +func (m *mockArtifactStore) GetWorkflowSpecList(ctx context.Context) ([]*job.WorkflowSpec, error) { + return m.artifactStore.GetWorkflowSpecList(ctx) +} + func (m *mockArtifactStore) DeleteWorkflowArtifactsBatch(ctx context.Context, workflowIDs []string) error { return m.artifactStore.DeleteWorkflowArtifactsBatch(ctx, workflowIDs) } @@ -1107,7 +1111,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 @@ -1126,7 +1130,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 @@ -1181,7 +1185,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 @@ -1262,7 +1266,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 @@ -1281,7 +1285,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 @@ -1296,7 +1300,10 @@ func Test_workflowDeletedHandler(t *testing.T) { type stubWorkflowArtifactsStore struct { spec *job.WorkflowSpec + specs []*job.WorkflowSpec + upsertErr error deleteErr error + listErr error deleteCalls atomic.Int32 } @@ -1312,9 +1319,19 @@ 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 } +func (s *stubWorkflowArtifactsStore) GetWorkflowSpecList(context.Context) ([]*job.WorkflowSpec, error) { + if s.listErr != nil { + return nil, s.listErr + } + return s.specs, nil +} + func (s *stubWorkflowArtifactsStore) DeleteWorkflowArtifacts(context.Context, string) error { s.deleteCalls.Add(1) return s.deleteErr @@ -1340,7 +1357,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()) @@ -1366,7 +1383,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()) @@ -1404,7 +1421,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/services/workflows/syncer/v2/mocks/orm.go b/core/services/workflows/syncer/v2/mocks/orm.go index a4d02addf29..67c36249cd5 100644 --- a/core/services/workflows/syncer/v2/mocks/orm.go +++ b/core/services/workflows/syncer/v2/mocks/orm.go @@ -175,6 +175,64 @@ func (_c *ORM_GetWorkflowSpec_Call) RunAndReturn(run func(context.Context, strin return _c } +// GetWorkflowSpecList provides a mock function with given fields: ctx +func (_m *ORM) GetWorkflowSpecList(ctx context.Context) ([]*job.WorkflowSpec, error) { + ret := _m.Called(ctx) + + if len(ret) == 0 { + panic("no return value specified for GetWorkflowSpecList") + } + + var r0 []*job.WorkflowSpec + var r1 error + if rf, ok := ret.Get(0).(func(context.Context) ([]*job.WorkflowSpec, error)); ok { + return rf(ctx) + } + if rf, ok := ret.Get(0).(func(context.Context) []*job.WorkflowSpec); ok { + r0 = rf(ctx) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]*job.WorkflowSpec) + } + } + + if rf, ok := ret.Get(1).(func(context.Context) error); ok { + r1 = rf(ctx) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// ORM_GetWorkflowSpecList_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetWorkflowSpecList' +type ORM_GetWorkflowSpecList_Call struct { + *mock.Call +} + +// GetWorkflowSpecList is a helper method to define mock.On call +// - ctx context.Context +func (_e *ORM_Expecter) GetWorkflowSpecList(ctx interface{}) *ORM_GetWorkflowSpecList_Call { + return &ORM_GetWorkflowSpecList_Call{Call: _e.mock.On("GetWorkflowSpecList", ctx)} +} + +func (_c *ORM_GetWorkflowSpecList_Call) Run(run func(ctx context.Context)) *ORM_GetWorkflowSpecList_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context)) + }) + return _c +} + +func (_c *ORM_GetWorkflowSpecList_Call) Return(_a0 []*job.WorkflowSpec, _a1 error) *ORM_GetWorkflowSpecList_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *ORM_GetWorkflowSpecList_Call) RunAndReturn(run func(context.Context) ([]*job.WorkflowSpec, error)) *ORM_GetWorkflowSpecList_Call { + _c.Call.Return(run) + return _c +} + // UpsertWorkflowSpec provides a mock function with given fields: ctx, spec func (_m *ORM) UpsertWorkflowSpec(ctx context.Context, spec *job.WorkflowSpec) (int64, error) { ret := _m.Called(ctx, spec) diff --git a/core/web/resolver/testdata/config-empty-effective.toml b/core/web/resolver/testdata/config-empty-effective.toml index 2bf0926bf4a..0b5b7d4539d 100644 --- a/core/web/resolver/testdata/config-empty-effective.toml +++ b/core/web/resolver/testdata/config-empty-effective.toml @@ -367,6 +367,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +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 2499e984436..6eb0e9069d1 100644 --- a/core/web/resolver/testdata/config-full.toml +++ b/core/web/resolver/testdata/config-full.toml @@ -385,6 +385,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 = 'clp-cre-wf-zone-a-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 10bdee05eb2..316292cce52 100644 --- a/core/web/resolver/testdata/config-multi-chain-effective.toml +++ b/core/web/resolver/testdata/config-multi-chain-effective.toml @@ -367,6 +367,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200 diff --git a/deployment/go.mod b/deployment/go.mod index 889173e8005..b39e7d8e17d 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.20260715145851-8219609496a4 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260707105132-2da5d31f22fc github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 @@ -441,6 +441,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-20260710151514-27b5a126dabe // indirect github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 // 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 75e43de19f3..f3620bf625d 100644 --- a/deployment/go.sum +++ b/deployment/go.sum @@ -1390,8 +1390,8 @@ github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260715132147-59df1b138100 h github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260715132147-59df1b138100/go.mod h1:0v6RGdYa9NezVnBPIAVyxB3A9furKWwaSkLha+URYGk= 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.20260715145851-8219609496a4 h1:LzDSHQ5bM4pp8Wrr0mFjg60A4J47+IFPIuC/rHNF1qA= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4/go.mod h1:snfVBRRQTpC2x5O3bQHZe9SvJX5yv/SbG8oHkJTKLtE= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba h1:1sx4ZrmoGiBOa0obx8K7VTojaODpXB67KGTfqLObjss= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba/go.mod h1:tAsXbMsNrwIp7vCqeauNsT4f3gVATbTmIviCkZPDgdo= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 h1:DvfhsiIxB4JMuR+r1UciKV8jKiwhI/OZI/QhKawC4pM= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6/go.mod h1:6NefaCIMH4zFBN3T+cvsFGYoy3oTSPKDaxPAyBzlYTU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1436,6 +1436,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.202607011 github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.20260701185448-696c075849ea/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-20260710151514-27b5a126dabe h1:MDnY5wQbWTpFdDnMRicEnoMfSP5nM/KncARr4skP1ug= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 h1:/vkKPJoweLkRd56V4YHGRAtTG4+/JAlgklGEfvH6l4c= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16/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 4d9c685e59e..dffb01580fc 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -2588,6 +2588,77 @@ 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 = 'cre' # 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. This TOML section is the single authority for metering on the +core node; there is no environment-variable gate. For capability LOOP plugins +the values are forwarded unchanged over the plugin environment +(loop.EnvConfig), which is only a child-process transport produced from this +config, not a separate gate. Snapshots are emitted on a fixed internal +interval and are bucket-aligned (each snapshot timestamp is truncated to the +interval) so cross-node snapshot buckets agree. + +### 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 = 'cre' # 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 edb3e10f6c1..3e6a887a0bc 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.1.1-0.20260715132147-59df1b138100 - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260707105132-2da5d31f22fc @@ -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-20260710151514-27b5a126dabe github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 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 bb7cdc9ff07..ff97e988c77 100644 --- a/go.sum +++ b/go.sum @@ -1159,8 +1159,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.1.1-0.20260715132147-59df1b138100 h1:sH546tdm3VaYGaRLnZ6ZYBZiE25vEkMap76Xq/vQWx0= github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260715132147-59df1b138100/go.mod h1:0v6RGdYa9NezVnBPIAVyxB3A9furKWwaSkLha+URYGk= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4 h1:LzDSHQ5bM4pp8Wrr0mFjg60A4J47+IFPIuC/rHNF1qA= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4/go.mod h1:snfVBRRQTpC2x5O3bQHZe9SvJX5yv/SbG8oHkJTKLtE= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba h1:1sx4ZrmoGiBOa0obx8K7VTojaODpXB67KGTfqLObjss= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba/go.mod h1:tAsXbMsNrwIp7vCqeauNsT4f3gVATbTmIviCkZPDgdo= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 h1:DvfhsiIxB4JMuR+r1UciKV8jKiwhI/OZI/QhKawC4pM= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6/go.mod h1:6NefaCIMH4zFBN3T+cvsFGYoy3oTSPKDaxPAyBzlYTU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1201,6 +1201,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-20260710151514-27b5a126dabe h1:MDnY5wQbWTpFdDnMRicEnoMfSP5nM/KncARr4skP1ug= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 h1:/vkKPJoweLkRd56V4YHGRAtTG4+/JAlgklGEfvH6l4c= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16/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 7239fc1860d..a5f173b37a4 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.20260715145851-8219609496a4 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260715161014-611d8ac32364 @@ -60,6 +60,8 @@ require ( gopkg.in/guregu/null.v4 v4.0.0 ) +require github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe // indirect + require ( cloud.google.com/go/compute/metadata v0.9.0 // indirect cosmossdk.io/api v0.7.6 // indirect diff --git a/integration-tests/go.sum b/integration-tests/go.sum index 872f4b98dbf..59a7e2512ab 100644 --- a/integration-tests/go.sum +++ b/integration-tests/go.sum @@ -1377,8 +1377,8 @@ github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260715132147-59df1b138100 h github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260715132147-59df1b138100/go.mod h1:0v6RGdYa9NezVnBPIAVyxB3A9furKWwaSkLha+URYGk= 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.20260715145851-8219609496a4 h1:LzDSHQ5bM4pp8Wrr0mFjg60A4J47+IFPIuC/rHNF1qA= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4/go.mod h1:snfVBRRQTpC2x5O3bQHZe9SvJX5yv/SbG8oHkJTKLtE= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba h1:1sx4ZrmoGiBOa0obx8K7VTojaODpXB67KGTfqLObjss= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba/go.mod h1:tAsXbMsNrwIp7vCqeauNsT4f3gVATbTmIviCkZPDgdo= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 h1:DvfhsiIxB4JMuR+r1UciKV8jKiwhI/OZI/QhKawC4pM= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6/go.mod h1:6NefaCIMH4zFBN3T+cvsFGYoy3oTSPKDaxPAyBzlYTU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1423,6 +1423,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.202607011 github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.20260701185448-696c075849ea/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-20260710151514-27b5a126dabe h1:MDnY5wQbWTpFdDnMRicEnoMfSP5nM/KncARr4skP1ug= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 h1:/vkKPJoweLkRd56V4YHGRAtTG4+/JAlgklGEfvH6l4c= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16/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 172ca6a46a4..821221b306c 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.20260715145851-8219609496a4 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260715161014-611d8ac32364 github.com/smartcontractkit/chainlink-testing-framework/framework v0.16.6-0.20260708113039-95f97b2d25e9 @@ -39,6 +39,12 @@ require ( golang.org/x/sync v0.22.0 ) +require ( + github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 // indirect + github.com/smartcontractkit/chainlink-protos/orchestrator v0.11.0 // indirect + github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd // indirect +) + require ( cloud.google.com/go/auth v0.18.2 // indirect cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect @@ -503,10 +509,8 @@ require ( github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36 // indirect github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.20260701185448-696c075849ea // indirect github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 // indirect - github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 // indirect + github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe // indirect; indirect develop github.com/smartcontractkit/chainlink-protos/op-catalog v0.1.0 // indirect - github.com/smartcontractkit/chainlink-protos/orchestrator v0.11.0 // indirect - github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd // indirect github.com/smartcontractkit/chainlink-protos/rmn/v1.6/go v0.0.0-20250131130834-15e0d4cde2a6 // indirect github.com/smartcontractkit/chainlink-protos/storage-service v0.3.0 // indirect github.com/smartcontractkit/chainlink-protos/svr v1.2.0 // indirect diff --git a/integration-tests/load/go.sum b/integration-tests/load/go.sum index 5d4b7a4ae36..8260b79d0c0 100644 --- a/integration-tests/load/go.sum +++ b/integration-tests/load/go.sum @@ -1637,8 +1637,8 @@ github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260715132147-59df1b138100 h github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260715132147-59df1b138100/go.mod h1:0v6RGdYa9NezVnBPIAVyxB3A9furKWwaSkLha+URYGk= 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.20260715145851-8219609496a4 h1:LzDSHQ5bM4pp8Wrr0mFjg60A4J47+IFPIuC/rHNF1qA= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4/go.mod h1:snfVBRRQTpC2x5O3bQHZe9SvJX5yv/SbG8oHkJTKLtE= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba h1:1sx4ZrmoGiBOa0obx8K7VTojaODpXB67KGTfqLObjss= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba/go.mod h1:tAsXbMsNrwIp7vCqeauNsT4f3gVATbTmIviCkZPDgdo= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 h1:DvfhsiIxB4JMuR+r1UciKV8jKiwhI/OZI/QhKawC4pM= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6/go.mod h1:6NefaCIMH4zFBN3T+cvsFGYoy3oTSPKDaxPAyBzlYTU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1683,6 +1683,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.202607011 github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.20260701185448-696c075849ea/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-20260710151514-27b5a126dabe h1:MDnY5wQbWTpFdDnMRicEnoMfSP5nM/KncARr4skP1ug= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 h1:/vkKPJoweLkRd56V4YHGRAtTG4+/JAlgklGEfvH6l4c= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16/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 1239102eb2f..eb10022468c 100644 --- a/plugins/loop_registry_test.go +++ b/plugins/loop_registry_test.go @@ -98,6 +98,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 "clp-cre-wf-zone-a-1" } + type mockPrometheusBridge struct{} func (m mockPrometheusBridge) Enabled() bool { return true } @@ -191,6 +202,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 @@ -203,6 +215,7 @@ func TestLoopRegistry_Register(t *testing.T) { cfgMercury: mockCfgMercury, cfgTracing: mockCfgTracing, cfgTelemetry: mockCfgTelemetry, + cfgMetering: mockCfgMetering, } // Test case 1: Register new loop @@ -256,6 +269,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, "clp-cre-wf-zone-a-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 d5989c7f081..24d0bfa8eff 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: "792fd475a9c294b61fa90b9dd23388da2827f093" + 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: "2cf965790a71171e1336f6d1d4f5162f2bd39bd1" + installPath: "." + evm: + - moduleURI: "github.com/smartcontractkit/capabilities/chain_capabilities/evm" + gitRef: "d7162a2b15aea2ce861052ebaa2f5ca716a9b60a" + 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: "4cac911b36d589a67689e3ab1ca26ea263eea440" diff --git a/system-tests/lib/go.mod b/system-tests/lib/go.mod index bcee8d79808..1cc500dbfe1 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.106 github.com/smartcontractkit/chainlink-aptos v0.0.0-20260708114855-e953eeb028a7 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260715161014-611d8ac32364 @@ -474,6 +474,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-20260710151514-27b5a126dabe // indirect github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 // 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 d452336e6fb..07fb9d503a0 100644 --- a/system-tests/lib/go.sum +++ b/system-tests/lib/go.sum @@ -1551,8 +1551,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.1.1-0.20260715132147-59df1b138100 h1:sH546tdm3VaYGaRLnZ6ZYBZiE25vEkMap76Xq/vQWx0= github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260715132147-59df1b138100/go.mod h1:0v6RGdYa9NezVnBPIAVyxB3A9furKWwaSkLha+URYGk= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4 h1:LzDSHQ5bM4pp8Wrr0mFjg60A4J47+IFPIuC/rHNF1qA= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4/go.mod h1:snfVBRRQTpC2x5O3bQHZe9SvJX5yv/SbG8oHkJTKLtE= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba h1:1sx4ZrmoGiBOa0obx8K7VTojaODpXB67KGTfqLObjss= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba/go.mod h1:tAsXbMsNrwIp7vCqeauNsT4f3gVATbTmIviCkZPDgdo= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 h1:DvfhsiIxB4JMuR+r1UciKV8jKiwhI/OZI/QhKawC4pM= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6/go.mod h1:6NefaCIMH4zFBN3T+cvsFGYoy3oTSPKDaxPAyBzlYTU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1597,6 +1597,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.202607011 github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.20260701185448-696c075849ea/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-20260710151514-27b5a126dabe h1:MDnY5wQbWTpFdDnMRicEnoMfSP5nM/KncARr4skP1ug= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 h1:/vkKPJoweLkRd56V4YHGRAtTG4+/JAlgklGEfvH6l4c= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16/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 70bb420ec0a..e0624853060 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.106 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260624154507-ea7ff77a0ddb - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 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 @@ -245,6 +245,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.20.1-0.20260701185448-696c075849ea // indirect + github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe // indirect github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 // indirect github.com/smartcontractkit/chainlink-solana/contracts v0.0.0-20260513123719-d347eaf314e1 // indirect github.com/smartcontractkit/chainlink-sui/codec v0.0.0-20260714120433-7667cad5ff5c // indirect diff --git a/system-tests/tests/go.sum b/system-tests/tests/go.sum index 12c85759577..52fbe06cc21 100644 --- a/system-tests/tests/go.sum +++ b/system-tests/tests/go.sum @@ -1756,8 +1756,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.1.1-0.20260715132147-59df1b138100 h1:sH546tdm3VaYGaRLnZ6ZYBZiE25vEkMap76Xq/vQWx0= github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260715132147-59df1b138100/go.mod h1:0v6RGdYa9NezVnBPIAVyxB3A9furKWwaSkLha+URYGk= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4 h1:LzDSHQ5bM4pp8Wrr0mFjg60A4J47+IFPIuC/rHNF1qA= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715145851-8219609496a4/go.mod h1:snfVBRRQTpC2x5O3bQHZe9SvJX5yv/SbG8oHkJTKLtE= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba h1:1sx4ZrmoGiBOa0obx8K7VTojaODpXB67KGTfqLObjss= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260715163833-347adf48a9ba/go.mod h1:tAsXbMsNrwIp7vCqeauNsT4f3gVATbTmIviCkZPDgdo= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6 h1:DvfhsiIxB4JMuR+r1UciKV8jKiwhI/OZI/QhKawC4pM= github.com/smartcontractkit/chainlink-common/keystore v1.2.1-0.20260623104656-f39eba3e2bc6/go.mod h1:6NefaCIMH4zFBN3T+cvsFGYoy3oTSPKDaxPAyBzlYTU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= @@ -1802,6 +1802,8 @@ github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.202607011 github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.20260701185448-696c075849ea/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-20260710151514-27b5a126dabe h1:MDnY5wQbWTpFdDnMRicEnoMfSP5nM/KncARr4skP1ug= +github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe/go.mod h1:z7lx7wI3XZ4u9kmUtAVdwn1BCC9T8aieWSDcuDgPTdQ= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 h1:/vkKPJoweLkRd56V4YHGRAtTG4+/JAlgklGEfvH6l4c= github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16/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 59d1312b0e3..50bcc8437ff 100644 --- a/testdata/scripts/config/merge_raw_configs.txtar +++ b/testdata/scripts/config/merge_raw_configs.txtar @@ -514,6 +514,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +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 e23f12d629d..db93dc6a3f2 100644 --- a/testdata/scripts/node/validate/default.txtar +++ b/testdata/scripts/node/validate/default.txtar @@ -379,6 +379,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +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 071cf3b1e9f..18cd6eca212 100644 --- a/testdata/scripts/node/validate/defaults-override.txtar +++ b/testdata/scripts/node/validate/defaults-override.txtar @@ -440,6 +440,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +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 634c721c983..9ac7f44e307 100644 --- a/testdata/scripts/node/validate/disk-based-logging-disabled.txtar +++ b/testdata/scripts/node/validate/disk-based-logging-disabled.txtar @@ -423,6 +423,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +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 f8c0faf3203..768401a6a0a 100644 --- a/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar +++ b/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar @@ -423,6 +423,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +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 8faa53ac7fc..3425522c13f 100644 --- a/testdata/scripts/node/validate/disk-based-logging.txtar +++ b/testdata/scripts/node/validate/disk-based-logging.txtar @@ -423,6 +423,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +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 01159b25ba0..add4fae96de 100644 --- a/testdata/scripts/node/validate/fallback-override.txtar +++ b/testdata/scripts/node/validate/fallback-override.txtar @@ -525,6 +525,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +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 0336bde785d..8c9eaea4b80 100644 --- a/testdata/scripts/node/validate/invalid-ocr-p2p.txtar +++ b/testdata/scripts/node/validate/invalid-ocr-p2p.txtar @@ -408,6 +408,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +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 9dd2d8e2720..1f60b2bbbd4 100644 --- a/testdata/scripts/node/validate/invalid.txtar +++ b/testdata/scripts/node/validate/invalid.txtar @@ -419,6 +419,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +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 6131678aca3..48e41fe4bb6 100644 --- a/testdata/scripts/node/validate/valid.txtar +++ b/testdata/scripts/node/validate/valid.txtar @@ -420,6 +420,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +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 df728c63b0d..6b300e2ce06 100644 --- a/testdata/scripts/node/validate/warnings.txtar +++ b/testdata/scripts/node/validate/warnings.txtar @@ -402,6 +402,16 @@ LogMaxQueueSize = 2048 Enabled = false Prefixes = ['go_'] +[Metering] +MeterRecordsEnabled = false +MeterSnapshotsEnabled = false +Product = 'cre' +Tenant = '' +NumericTenantID = '' +Environment = '' +Zone = '' +NodeID = '' + [Workflows] [Workflows.Limits] Global = 200