adding resource manager for workflows

core/cmd/shell_local_test.go

@@ -74,7 +74,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

@@ -960,6 +960,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

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

@@ -64,6 +64,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"`
@@ -112,6 +113,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)
@@ -3163,6 +3165,75 @@ 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",
+		})
+	}
+	if b.MeterRecordsEnabled != nil && *b.MeterRecordsEnabled && (b.NodeID == nil || *b.NodeID == "") {
+		err = errors.Join(err, configutils.ErrInvalid{
+			Name:  "NodeID",
+			Value: "",
+			Msg:   "must be non-empty when MeterRecordsEnabled is true (an empty NodeID collapses per-node snapshot dedup scope DON-wide)",
+		})
+	}
+	return err
+}
+
 type PrometheusBridge struct {
 	Enabled  *bool
 	Prefixes []string

core/config/toml/types_test.go

@@ -860,3 +860,67 @@ func durationPtr(d time.Duration) *commonconfig.Duration {
 	cd := *commonconfig.MustNewDuration(d)
 	return &cd
 }
+
+func TestMetering_ValidateConfig(t *testing.T) {
+	t.Parallel()
+	testCases := []struct {
+		name        string
+		config      *Metering
+		expectError bool
+		errorMsg    string
+	}{
+		{
+			name:        "disabled with all nil fields",
+			config:      &Metering{},
+			expectError: false,
+		},
+		{
+			name: "records enabled with non-empty NodeID",
+			config: &Metering{
+				MeterRecordsEnabled: new(true),
+				NodeID:              new("clp-cre-wf-zone-a-1"),
+			},
+			expectError: false,
+		},
+		{
+			name: "records enabled with nil NodeID",
+			config: &Metering{
+				MeterRecordsEnabled: new(true),
+				NodeID:              nil,
+			},
+			expectError: true,
+			errorMsg:    "NodeID",
+		},
+		{
+			name: "records enabled with empty NodeID",
+			config: &Metering{
+				MeterRecordsEnabled: new(true),
+				NodeID:              new(""),
+			},
+			expectError: true,
+			errorMsg:    "NodeID",
+		},
+		{
+			name: "snapshots enabled without records enabled",
+			config: &Metering{
+				MeterSnapshotsEnabled: new(true),
+				NodeID:                new("clp-cre-wf-zone-a-1"),
+			},
+			expectError: true,
+			errorMsg:    "requires MeterRecordsEnabled to be true",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			err := tc.config.ValidateConfig()
+			if tc.expectError {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tc.errorMsg)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

core/scripts/cre/environment/configs/examples/workflow-don-overrides.toml

@@ -64,28 +64,16 @@
       #docker_file = "core/chainlink.Dockerfile"
       image = "chainlink-tmp:latest"
       user_config_overrides = """
+      # [Telemetry], [Billing], and [Metering] are framework-managed and are
+      # REJECTED here (see validateUserConfigOverrides in system-tests/lib):
+      # the framework generates them itself, pointing telemetry at the
+      # chip-router so downstream subscribers (e.g. the CI test sink) see all
+      # events. To enable metering, set enable_metering = true on the nodeset.
       [Log]
       Level = 'debug'
       JSONConsole = true
 
-      [Telemetry]
-      Enabled = true
-      # assumes that CTF's observability stack is running on the local host
-      # 4317 is OTEL ingress port of the stack (LGTM)
-      # command: ctf obs up
-      Endpoint = 'host.docker.internal:4317'
-      # default service-name and port used by locally running Chip Ingress
-      # command: go run . env chip-ingress-stack start
-      ChipIngressEndpoint = 'chip-ingress:50051'
-      InsecureConnection = true
-      TraceSampleRatio = 1
-      HeartbeatInterval = '30s'
-
       # Remove this bit to allow downloading workflows from remote sources
       [CRE.WorkflowFetcher]
       URL = "file:///home/chainlink/workflows"
-
-      [Billing]
-      URL = "host.docker.internal:2223"
-      TLSEnabled = false
       """

core/scripts/cre/environment/configs/workflow-gateway-capabilities-don.toml

@@ -40,6 +40,7 @@
   name = "workflow"
   don_family = "test-don-family"
   don_types = ["workflow"]
+  enable_metering = true
   override_mode = "all"
   http_port_range_start = 10100
 
@@ -71,6 +72,7 @@
   name = "capabilities"
   don_family = "test-don-family"
   don_types = ["capabilities"]
+  enable_metering = true
   exposes_remote_capabilities = true
   override_mode = "all"
   http_port_range_start = 10200

core/scripts/cre/environment/environment/chip_ingress_stack.go

@@ -77,10 +77,14 @@ func schemaCommitRefFromGoMod(ctx context.Context, repoRoot, targetModule string
 // getSchemaSetFromGoMod resolves SchemaSets from chainlink-protos commits pinned in go.mod:
 //   - workflows (chip-cre.json) for CRE/workflow telemetry
 //   - node-platform (chip-schemas.json) for PluginRelayerConfigEmitter / common.v1.ChainPluginConfig
+//   - metering (chip-cll.meter.json) for durable resource metering (MeterRecord/MeterSnapshot on
+//     the cll.meter domain); without this, ChIP Ingress rejects those events at pre-publish encode
+//     time with "Subject 'cll-meter-metering.v1.MeterRecord' not found" and drops them silently.
 func getSchemaSetFromGoMod(ctx context.Context) ([]chipingressset.SchemaSet, error) {
 	const (
 		workflowsModule    = "github.com/smartcontractkit/chainlink-protos/workflows/go"
 		nodePlatformModule = "github.com/smartcontractkit/chainlink-protos/node-platform"
+		meteringModule     = "github.com/smartcontractkit/chainlink-protos/metering/go"
 	)
 
 	repoRoot, err := filepath.Abs(relativePathToRepoRoot)
@@ -100,6 +104,12 @@ func getSchemaSetFromGoMod(ctx context.Context) ([]chipingressset.SchemaSet, err
 	}
 	framework.L.Info().Msgf("Extracted commit ref for %s: %s (from version: %s)", nodePlatformModule, npRef, npVer)
 
+	meteringRef, meteringVer, err := schemaCommitRefFromGoMod(ctx, repoRoot, meteringModule)
+	if err != nil {
+		return nil, err
+	}
+	framework.L.Info().Msgf("Extracted commit ref for %s: %s (from version: %s)", meteringModule, meteringRef, meteringVer)
+
 	return []chipingressset.SchemaSet{
 		{
 			URI:        chainlinkProtosGitURI,
@@ -113,6 +123,12 @@ func getSchemaSetFromGoMod(ctx context.Context) ([]chipingressset.SchemaSet, err
 			SchemaDir:  "node-platform",
 			ConfigFile: "chip-schemas.json",
 		},
+		{
+			URI:        chainlinkProtosGitURI,
+			Ref:        meteringRef,
+			SchemaDir:  "metering",
+			ConfigFile: "chip-cll.meter.json",
+		},
 	}, nil
 }
 
@@ -1045,6 +1061,6 @@ func fetchAndRegisterProtosCmd() *cobra.Command {
 			})
 		},
 	}
-	cmd.Flags().StringVarP(&chipIngressGRPCURL, "chip-ingress-grpc-url", "h", "localhost:"+chipingressset.DEFAULT_CHIP_INGRESS_GRPC_PORT, "Chip Ingress GRPC URL")
+	cmd.Flags().StringVarP(&chipIngressGRPCURL, "chip-ingress-grpc-url", "u", "localhost:"+chipingressset.DEFAULT_CHIP_INGRESS_GRPC_PORT, "Chip Ingress GRPC URL")
 	return cmd
 }

core/scripts/go.mod

@@ -511,7 +511,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/metering/go v0.0.0-20260729184203-90b4cdd48536 // 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

@@ -286,7 +286,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

@@ -587,6 +587,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}
 }