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

@@ -922,6 +922,28 @@ Enabled = false # Default
 # By default, we only forward the go runtime metrics. Empty means forward everything.
 Prefixes = ["go_"] # Default
 
+# Metering configures durable resource metering emission and the coarse
+# deployment/node identity dimensions stamped on emitted MeterRecords and
+# MeterSnapshots. These are plumbed to LOOP plugins via env.
+[Metering]
+# MeterRecordsEnabled enables durable MeterRecord emission for LOOP plugins.
+MeterRecordsEnabled = false # Default
+# MeterSnapshotsEnabled enables durable MeterSnapshot emission for LOOP plugins.
+# Requires MeterRecordsEnabled = true.
+MeterSnapshotsEnabled = false # Default
+# Product is the deployment product identity dimension, e.g. 'cre'.
+Product = '' # Default
+# Tenant is the human-readable tenant name, e.g. 'mainline'.
+Tenant = '' # Default
+# NumericTenantID is the numbered tenant identifier represented as a string.
+NumericTenantID = '' # Default
+# Environment is the deployment environment identity dimension, e.g. 'production'.
+Environment = '' # Default
+# Zone is the deployment zone identity dimension, e.g. 'wf-zone-a'.
+Zone = '' # Default
+# NodeID is the node's logical name, e.g. 'clp-cre-wf-zone-a-1' (not the CSA public key).
+NodeID = '' # Default
+
 [CRE.Streams]
 # WsURL is the websockets url for the streams sdk config
 WsURL = "streams.url" # Example

core/config/env/env.go

@@ -14,6 +14,10 @@ var (
 	DatabaseAllowSimplePasswords = Var("CL_DATABASE_ALLOW_SIMPLE_PASSWORDS")
 	IgnorePrereleaseVersionCheck = Var("CL_IGNORE_PRE_RELEASE_VERSION_CHECK")
 	SkipAppVersionCheck          = Var("CL_SKIP_APP_VERSION_CHECK")
+	// MeterRecordsEnabled gates emission of metering.v1.MeterRecord events for
+	// durable CRE resources. Accepts strconv.ParseBool values; default false.
+	// Temporary deploy gate; promotion to TOML config is tracked by SHARED-2718.
+	MeterRecordsEnabled = Var("CL_METER_RECORDS_ENABLED")
 
 	DatabaseURL         = Secret("CL_DATABASE_URL")
 	DatabaseBackupURL   = Secret("CL_DATABASE_BACKUP_URL")

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

core/scripts/go.mod

@@ -47,7 +47,7 @@ require (
 	github.com/smartcontractkit/chain-selectors v1.0.104
 	github.com/smartcontractkit/chainlink-automation v0.8.1
 	github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260624154507-ea7ff77a0ddb
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260701091216-9264d4444ce0
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6
 	github.com/smartcontractkit/chainlink-common/keystore v1.2.0
 	github.com/smartcontractkit/chainlink-data-streams v0.1.15-0.20260522094612-5f9f748bd87a
 	github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54
@@ -507,6 +507,7 @@ require (
 	github.com/smartcontractkit/chainlink-protos/chainlink-ccv/verifier v0.0.0-20251211142334-5c3421fe2c8d // indirect
 	github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260501174546-2e8846986b36 // indirect
 	github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20260512230622-65f10f4cd305 // indirect
+	github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019 // indirect
 	github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260512230622-65f10f4cd305 // indirect
 	github.com/smartcontractkit/chainlink-protos/orchestrator v0.11.0 // indirect
 	github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd // indirect

core/services/chainlink/application.go

@@ -287,7 +287,7 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
 	}
 	loopRegistry := plugins.NewLoopRegistry(globalLogger, cfg.AppID().String(), cfg.Feature().LogPoller(),
 		cfg.Database(), cfg.Mercury(), cfg.Pyroscope(), cfg.AutoPprof(), cfg.Tracing(), cfg.Telemetry(),
-		beholderAuthHeaders, csaPubKeyHex, cfg.LOOPP())
+		cfg.Metering(), beholderAuthHeaders, csaPubKeyHex, cfg.LOOPP())
 
 	relayerFactory := RelayerFactory{
 		Logger:                opts.Logger,
@@ -676,6 +676,16 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
 		atomicSettings,
 		creServices.OCRConfigService,
 		cfg.Capabilities().Local(),
+		// Host-injected deployment/node metering identity for spawned capability
+		// LOOPs. Sourced once here from node config: product constant, env/zone
+		// from [Telemetry.ResourceAttributes], node_id = the same CSA pubkey hex
+		// the node uses for beholder auth and the engine's node_id.
+		standardcapabilities.NodeIdentity{
+			Product:     "cre",
+			Environment: cfg.Telemetry().ResourceAttributes()["env"],
+			Zone:        cfg.Telemetry().ResourceAttributes()["zone"],
+			NodeID:      csaPubKeyHex,
+		},
 	)
 	delegates[job.StandardCapabilities] = stdcapDelegate
 	if creServices.SetDelegatesDeps != nil {

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,65 @@
+package chainlink
+
+import (
+	"github.com/smartcontractkit/chainlink/v2/core/config/toml"
+)
+
+type meteringConfig struct {
+	s toml.Metering
+}
+
+func (b *meteringConfig) MeterRecordsEnabled() bool {
+	if b.s.MeterRecordsEnabled == nil {
+		return false
+	}
+	return *b.s.MeterRecordsEnabled
+}
+
+func (b *meteringConfig) MeterSnapshotsEnabled() bool {
+	if b.s.MeterSnapshotsEnabled == nil {
+		return false
+	}
+	return *b.s.MeterSnapshotsEnabled
+}
+
+func (b *meteringConfig) Product() string {
+	if b.s.Product == nil {
+		return ""
+	}
+	return *b.s.Product
+}
+
+func (b *meteringConfig) Tenant() string {
+	if b.s.Tenant == nil {
+		return ""
+	}
+	return *b.s.Tenant
+}
+
+func (b *meteringConfig) NumericTenantID() string {
+	if b.s.NumericTenantID == nil {
+		return ""
+	}
+	return *b.s.NumericTenantID
+}
+
+func (b *meteringConfig) Environment() string {
+	if b.s.Environment == nil {
+		return ""
+	}
+	return *b.s.Environment
+}
+
+func (b *meteringConfig) Zone() string {
+	if b.s.Zone == nil {
+		return ""
+	}
+	return *b.s.Zone
+}
+
+func (b *meteringConfig) NodeID() string {
+	if b.s.NodeID == nil {
+		return ""
+	}
+	return *b.s.NodeID
+}

core/services/chainlink/config_metering_test.go

@@ -0,0 +1,44 @@
+package chainlink
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/smartcontractkit/chainlink/v2/core/config/toml"
+)
+
+func TestMeteringConfig(t *testing.T) {
+	t.Run("defaults", func(t *testing.T) {
+		mc := meteringConfig{s: toml.Metering{}}
+		assert.False(t, mc.MeterRecordsEnabled())
+		assert.False(t, mc.MeterSnapshotsEnabled())
+		assert.Empty(t, mc.Product())
+		assert.Empty(t, mc.Tenant())
+		assert.Empty(t, mc.NumericTenantID())
+		assert.Empty(t, mc.Environment())
+		assert.Empty(t, mc.Zone())
+		assert.Empty(t, mc.NodeID())
+	})
+
+	t.Run("explicit values", func(t *testing.T) {
+		mc := meteringConfig{s: toml.Metering{
+			MeterRecordsEnabled:   ptr(true),
+			MeterSnapshotsEnabled: ptr(true),
+			Product:               ptr("cre"),
+			Tenant:                ptr("mainline"),
+			NumericTenantID:       ptr("42"),
+			Environment:           ptr("production"),
+			Zone:                  ptr("wf-zone-a"),
+			NodeID:                ptr("csa-pubkey-1"),
+		}}
+		assert.True(t, mc.MeterRecordsEnabled())
+		assert.True(t, mc.MeterSnapshotsEnabled())
+		assert.Equal(t, "cre", mc.Product())
+		assert.Equal(t, "mainline", mc.Tenant())
+		assert.Equal(t, "42", mc.NumericTenantID())
+		assert.Equal(t, "production", mc.Environment())
+		assert.Equal(t, "wf-zone-a", mc.Zone())
+		assert.Equal(t, "csa-pubkey-1", mc.NodeID())
+	})
+}

core/services/chainlink/config_test.go

@@ -578,6 +578,16 @@ func TestConfig_Marshal(t *testing.T) {
 			Prefixes: []string{"ocr_"},
 		},
 	}
+	full.Metering = toml.Metering{
+		MeterRecordsEnabled:   ptr(true),
+		MeterSnapshotsEnabled: ptr(true),
+		Product:               ptr("cre"),
+		Tenant:                ptr("mainline"),
+		NumericTenantID:       ptr("42"),
+		Environment:           ptr("production"),
+		Zone:                  ptr("wf-zone-a"),
+		NodeID:                ptr("csa-pubkey-1"),
+	}
 	full.CRE = toml.CreConfig{
 		UseLocalTimeProvider: ptr(true),
 		EnableDKGRecipient:   ptr(false),