adding resource manager for workflows

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),
 	}
 

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

core/config/docs/core.toml

@@ -929,6 +929,34 @@ 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. 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.
+[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

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
+}

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,70 @@ func (b *Telemetry) ValidateConfig() (err error) {
 	return err
 }
 
+// Metering configures durable resource metering emission and the coarse
+// deployment/node identity dimensions stamped on emitted MeterRecords and
+// MeterSnapshots. These are passed via loop.EnvConfig to every LOOP plugin.
+type Metering struct {
+	// MeterRecordsEnabled enables durable MeterRecord emission for LOOP plugins.
+	MeterRecordsEnabled *bool
+	// MeterSnapshotsEnabled enables durable MeterSnapshot emission. Requires
+	// MeterRecordsEnabled to be true.
+	MeterSnapshotsEnabled *bool
+	// Product is the deployment product identity dimension, e.g. "cre".
+	Product *string
+	// Tenant is the human-readable tenant name, e.g. "mainline".
+	Tenant *string
+	// NumericTenantID is the numbered tenant identifier as a string.
+	NumericTenantID *string
+	// Environment is the deployment environment dimension, e.g. "production".
+	Environment *string
+	// Zone is the deployment zone dimension, e.g. "wf-zone-a".
+	Zone *string
+	// NodeID is the node's logical name, e.g. "clp-cre-wf-zone-a-1" (NOT the CSA
+	// public key). The billing service uses it to look up the node's CSA key in
+	// the workflow registry; the CSA key itself is attached to events as the
+	// node_csa_key attribute.
+	NodeID *string
+}
+
+func (b *Metering) setFrom(f *Metering) {
+	if v := f.MeterRecordsEnabled; v != nil {
+		b.MeterRecordsEnabled = v
+	}
+	if v := f.MeterSnapshotsEnabled; v != nil {
+		b.MeterSnapshotsEnabled = v
+	}
+	if v := f.Product; v != nil {
+		b.Product = v
+	}
+	if v := f.Tenant; v != nil {
+		b.Tenant = v
+	}
+	if v := f.NumericTenantID; v != nil {
+		b.NumericTenantID = v
+	}
+	if v := f.Environment; v != nil {
+		b.Environment = v
+	}
+	if v := f.Zone; v != nil {
+		b.Zone = v
+	}
+	if v := f.NodeID; v != nil {
+		b.NodeID = v
+	}
+}
+
+func (b *Metering) ValidateConfig() (err error) {
+	if b.MeterSnapshotsEnabled != nil && *b.MeterSnapshotsEnabled && (b.MeterRecordsEnabled == nil || !*b.MeterRecordsEnabled) {
+		err = errors.Join(err, configutils.ErrInvalid{
+			Name:  "MeterSnapshotsEnabled",
+			Value: true,
+			Msg:   "requires MeterRecordsEnabled to be true",
+		})
+	}
+	return err
+}
+
 type PrometheusBridge struct {
 	Enabled  *bool
 	Prefixes []string

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

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,

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}
 }

core/services/chainlink/config_metering.go

@@ -0,0 +1,68 @@
+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 "cre" (resourcemanager.DefaultMeteringProduct) when unset,
+// so the syncer's metering identity matches the capability plugins' fallback and
+// metering can never be enabled with an empty product dimension.
+func (b *meteringConfig) Product() string {
+	if b.s.Product == nil {
+		return "cre"
+	}
+	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
+}

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, "cre", 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())
+	})
+}

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),