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

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

@@ -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)
@@ -3164,6 +3166,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)
+			}
+		})
+	}
+}