diff --git a/.changeset/compute-usage-meter-records.md b/.changeset/compute-usage-meter-records.md new file mode 100644 index 00000000000..3f0f92cd20d --- /dev/null +++ b/.changeset/compute-usage-meter-records.md @@ -0,0 +1,5 @@ +--- +"chainlink": patch +--- + +Emit a `cre:workflow:compute` usage MeterRecord per workflow execution when `[Metering].MeterRecordsEnabled` is on. The compute duration is reported by the module that ran the workflow (the guest run for WASM, the enclave-measured run for confidential workflows). #internal diff --git a/core/services/cre/cre.go b/core/services/cre/cre.go index 378e8e6a5fd..9eb36e49d87 100644 --- a/core/services/cre/cre.go +++ b/core/services/cre/cre.go @@ -24,6 +24,7 @@ import ( commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" "github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry" "github.com/smartcontractkit/chainlink-common/pkg/custmsg" + "github.com/smartcontractkit/chainlink-common/pkg/durableemitter" "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/loop" commonmetrics "github.com/smartcontractkit/chainlink-common/pkg/metrics" @@ -715,6 +716,22 @@ func newBillingClient(lggr logger.Logger, cfg Config, opts Opts) (metering.Billi // registry 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.NewSpecMeter. +// newDeploymentIdentity returns the node's [Metering] deployment dimensions. +func newDeploymentIdentity(cfg Config) resourcemanager.DeploymentIdentity { + m := cfg.Metering() + if m == nil { + return resourcemanager.DeploymentIdentity{} + } + return resourcemanager.DeploymentIdentity{ + Product: m.Product(), + Tenant: m.Tenant(), + NumericTenantID: m.NumericTenantID(), + Environment: m.Environment(), + Zone: m.Zone(), + NodeID: m.NodeID(), + } +} + func newSyncerMeterIdentity(cfg Config) resourcemanager.ResourceIdentity { m := cfg.Metering() if m == nil { @@ -982,6 +999,15 @@ func newWorkflowRegistrySyncerV2( handlerOpts = append(handlerOpts, syncerV2.WithSpecMeter(specMeter)) } + // Compute usage records share the [Metering].MeterRecordsEnabled gate. + if meterRecordsEnabled { + if d := durableemitter.GetGlobalEmitter(); d != nil { + handlerOpts = append(handlerOpts, syncerV2.WithUsageMetering(newDeploymentIdentity(cfg), d)) + } else { + lggr.Warnw("Compute usage metering enabled without a durable emitter; compute usage records disabled") + } + } + mc := capCfg.WorkflowRegistry().ModuleCache() cacheEnabled := mc.Enabled() diskMonitorEnabled := mc.DiskMonitorEnabled() || cacheEnabled diff --git a/core/services/workflows/syncer/v2/handler.go b/core/services/workflows/syncer/v2/handler.go index 46191f7a5ce..eff4120dcff 100644 --- a/core/services/workflows/syncer/v2/handler.go +++ b/core/services/workflows/syncer/v2/handler.go @@ -108,6 +108,11 @@ type eventHandler struct { // disabled: its handler-facing methods are nil-receiver safe no-ops. specMeter *SpecMeter + // meterRecordsEnabled gates the engines' compute usage records; meteringIdentity is the node's [Metering] dimensions. + meterRecordsEnabled bool + meteringIdentity resourcemanager.DeploymentIdentity + usageEmitter resourcemanager.Emitter + orgResolver orgresolver.OrgResolver secretsFetcher v2.SecretsFetcher // localSecretOverrides is keyed by owner address; values are secret id -> secret value @@ -193,6 +198,16 @@ func WithBillingClient(client metering.BillingClient) func(*eventHandler) { } } +// WithUsageMetering enables compute usage records on the engines, delivered +// through emitter; dep is the node's [Metering] dimensions. +func WithUsageMetering(dep resourcemanager.DeploymentIdentity, emitter resourcemanager.Emitter) func(*eventHandler) { + return func(e *eventHandler) { + e.meterRecordsEnabled = true + e.meteringIdentity = dep + e.usageEmitter = emitter + } +} + // WithSpecMeter supplies the SpecMeter that emits metering.v1.MeterRecord // events for the workflow_specs_v2 storage lifecycle. The handler runs it as a // sub-service and reports storage transitions through EmitSpecDelta; all other @@ -1505,6 +1520,9 @@ func (h *eventHandler) newV2EngineConfig( WorkflowRegistryAddress: h.workflowRegistryAddress, WorkflowRegistryChainSelector: h.workflowRegistryChainSelector, OrgResolver: h.orgResolver, + MeterRecordsEnabled: h.meterRecordsEnabled, + MeteringIdentity: h.meteringIdentity, + UsageEmitter: h.usageEmitter, SecretsFetcher: h.secretsFetcher, OverrideFetcher: h.overrideFetcherForOwner(owner), DebugMode: h.debugMode, diff --git a/core/services/workflows/v2/base_engine.go b/core/services/workflows/v2/base_engine.go index 3f64b735f55..4908a3fb675 100644 --- a/core/services/workflows/v2/base_engine.go +++ b/core/services/workflows/v2/base_engine.go @@ -27,6 +27,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/custmsg" "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/metrics" + "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/cresettings" @@ -111,7 +112,9 @@ type baseEngine struct { // tracer is the OTel tracer for this engine. It's a noop tracer when DebugMode is false. tracer trace.Tracer - orgID string + usageMeter *resourcemanager.ResourceManager + usageIdentity resourcemanager.ResourceIdentity + orgID string // orgIDMissingReason records why orgID is empty ("resolver_nil", // "resolver_error", or "empty_response"). It is set once in startWith() and // read by startExecution() to label the org_id_missing counter. It is only @@ -186,6 +189,11 @@ func newBaseEngine(cfg *EngineConfig, engineMode string) (*baseEngine, logger.Su engine.setLogger(beholderLogger) engine.meterReports = metering.NewReports(cfg.BillingClient, cfg.WorkflowOwner, cfg.WorkflowID, beholderLogger, labelsMap, metricsLabeler, cfg.WorkflowRegistryAddress, cfg.WorkflowRegistryChainSelector, metering.EngineVersionV2) engine.metrics = metricsLabeler + // Disabled metering is a no-op inside the resource manager. + engine.usageMeter = resourcemanager.NewResourceManager(cfg.Lggr, resourcemanager.ResourceManagerConfig{MeterRecordsEnabled: cfg.MeterRecordsEnabled, Emitter: cfg.UsageEmitter}) + engine.usageIdentity = resourcemanager.WithWorkflowUsagePool( + resourcemanager.NewBaseIdentity(cfg.MeteringIdentity, resourcemanager.EmittingServiceWorkflowEngine, ""), + resourcemanager.ResourceTypeWorkflowCompute) engine.loggerLabels.Store(&labelsMap) engine.localNode.Store(&localNode) @@ -732,13 +740,6 @@ func (e *baseEngine) startExecution(ctx context.Context, event triggers.Coordina } } - // Track time the guest spends blocked in DON-time host calls so it can be - // excluded from metered compute (see compute_metering.go). The wrapper is - // always installed; whether the recorded wait is subtracted is gated by - // e.cfg.MeterComputeExcludingHostWait below. - suspension := &suspensionTracker{} - timeProvider = newMeasuredTimeProvider(timeProvider, e.cfg.Clock, suspension) - // Limit is always usable even on a read failure; err is advisory. moduleExecuteMaxResponseSizeBytes, err := e.cfg.LocalLimiters.ExecutionResponse.Limit(ctx) if err != nil { @@ -765,7 +766,6 @@ func (e *baseEngine) startExecution(ctx context.Context, event triggers.Coordina base: e, WorkflowExecutionID: executionID, ExecutionTimestamp: executionTimestamp, UserLogChan: userLogChan, TimeProvider: timeProvider, SecretsFetcher: e.secretsFetcher(executionID), executionProfile: newExecutionProfileCollector(), - suspension: suspension, } execHelper.initLimiters(e.cfg.LocalLimiters) @@ -787,25 +787,14 @@ func (e *baseEngine) startExecution(ctx context.Context, event triggers.Coordina endTime := e.cfg.Clock.Now() executionDuration := endTime.Sub(startTime) - // computeDuration is what we meter as RESOURCE_TYPE_COMPUTE. We start from the - // end-to-end wall-clock and subtract time the guest spent blocked in DON-time - // host calls: that wait is not compute and is non-deterministic across nodes - // (a node whose DON-time request lands in the current consensus round returns - // in microseconds; one whose request slips to the next round blocks ~a full - // round). Operational latency metrics below intentionally keep using - // executionDuration. + // Metered compute: the module's reported duration, else wall-clock. computeDuration := executionDuration - if hostWait := suspension.total(); hostWait > 0 { - computeDuration -= hostWait - if computeDuration < 0 { - computeDuration = 0 - } - executionLogger.Debugw("Excluding DON-time wait from metered compute", - "wallClockMs", executionDuration.Milliseconds(), - "hostWaitMs", hostWait.Milliseconds(), - "computeMs", computeDuration.Milliseconds()) + if result != nil && result.ComputeDurationMs != nil { + computeDuration = time.Duration(result.GetComputeDurationMs()) * time.Millisecond } + e.emitComputeUsage(ctx, executionLogger, executionID, computeDuration) + if isMetering { computeUnit := billing.ResourceType_name[int32(billing.ResourceType_RESOURCE_TYPE_COMPUTE)] mrErr := meteringReport.Settle(computeUnit, @@ -1150,3 +1139,36 @@ func resolveOrgID(ctx context.Context, resolver orgresolver.OrgResolver, workflo } return resolvedOrg{ID: orgID} } + +// emitComputeUsage emits the cre:workflow:compute usage MeterRecord for one +// execution and logs the emission. The log line is a contract consumed by the +// billing reconciler (fields: executionID, eventID, resourceType, value, orgID) +// and must stay stable. Fail-open: never returns an error to the execution. +func (e *baseEngine) emitComputeUsage(ctx context.Context, lggr logger.Logger, executionID string, computeDuration time.Duration) { + if !e.cfg.MeterRecordsEnabled { + return + } + resourceID, err := resourcemanager.WorkflowUsageResourceID(e.cfg.WorkflowID, executionID) + if err != nil { + lggr.Errorw("Compute usage meter record not emitted", "err", err) + return + } + identity := e.usageIdentity + if node := e.localNode.Load(); node != nil { + identity = identity.WithDonID(strconv.FormatUint(uint64(node.WorkflowDON.ID), 10)) + } + valueMs := computeDuration.Milliseconds() + // The capability event id for compute is the execution id itself: one + // compute record per execution, identical on every node of the DON. + e.usageMeter.EmitUsage(ctx, identity, executionID, valueMs, resourcemanager.UtilizationFields{ + ResourceType: resourcemanager.ResourceTypeWorkflowCompute, + ResourceID: resourceID, + OrgID: e.orgID, + }) + lggr.Infow("Emitted capability usage meter record", + "eventID", executionID, + "resourceType", resourcemanager.ResourceTypeWorkflowCompute, + "value", valueMs, + "orgID", e.orgID, + ) +} diff --git a/core/services/workflows/v2/capability_executor.go b/core/services/workflows/v2/capability_executor.go index d5b40fc2ada..0236e1e4c02 100644 --- a/core/services/workflows/v2/capability_executor.go +++ b/core/services/workflows/v2/capability_executor.go @@ -46,10 +46,6 @@ type ExecutionHelper struct { callCounts map[limits.Limiter[int]]int executionProfile *executionProfileCollector - - // suspension accumulates wall-clock time the guest spent blocked in DON-time - // host calls, so it can be excluded from metered compute. See compute_metering.go. - suspension *suspensionTracker } func (c *ExecutionHelper) initLimiters(limiters *EngineLimiters) { diff --git a/core/services/workflows/v2/compute_metering.go b/core/services/workflows/v2/compute_metering.go deleted file mode 100644 index 389f8b4650e..00000000000 --- a/core/services/workflows/v2/compute_metering.go +++ /dev/null @@ -1,76 +0,0 @@ -package v2 - -import ( - "sync/atomic" - "time" - - "github.com/jonboulle/clockwork" -) - -// suspensionTracker accumulates the wall-clock time during which the WASM guest -// is suspended inside a blocking DON-time host call (GetDONTime). -// -// Why this exists: -// Compute (RESOURCE_TYPE_COMPUTE) is metered as the end-to-end wall-clock around -// Module.Execute. That wall-clock also includes time the guest spends blocked in -// GetDONTime, which can stall for a full DON-time consensus round. That wait is -// not compute: it is non-deterministic across nodes (a node whose request lands -// in the current round returns in microseconds; a node whose request slips to -// the next round blocks for ~a round), and it has produced >100x divergence in -// per-node metered compute - and therefore consumed credits - for -// otherwise-identical executions on the same DON. -// -// Why summing is correct: -// The WASM guest is single-threaded, so its GetDONTime calls happen one at a -// time and never overlap with guest execution. Summing the individual wait -// intervals therefore yields the total non-compute DON-time wait, and -// -// computeTime = wallClock - totalDONTimeWait -// -// Scope: this only covers the DON-time wait, which is the one blocking host call -// the engine observes (via TimeProvider). It is NOT a general "guest suspended" -// accountant - see the package note in the PR for the capability/secrets-await -// gap, which is only measurable inside chainlink-common's host runtime. -type suspensionTracker struct { - totalNanos atomic.Int64 -} - -// add records a single guest-suspension interval. Safe for concurrent use and -// nil-safe so callers never have to guard the pointer. -func (s *suspensionTracker) add(d time.Duration) { - if s == nil || d <= 0 { - return - } - s.totalNanos.Add(int64(d)) -} - -// total returns the cumulative guest-suspended duration recorded so far. -func (s *suspensionTracker) total() time.Duration { - if s == nil { - return 0 - } - return time.Duration(s.totalNanos.Load()) -} - -// measuredTimeProvider decorates a TimeProvider, recording the wall-clock spent -// in each blocking time call into the supplied suspension tracker. -// -// GetDONTime can block on DON-time consensus; that wait must not be billed as -// compute. GetNodeTime is served locally and is effectively instant, so it is -// inherited unchanged from the embedded TimeProvider (no measurement needed). -type measuredTimeProvider struct { - TimeProvider - clock clockwork.Clock - tracker *suspensionTracker -} - -func newMeasuredTimeProvider(tp TimeProvider, clock clockwork.Clock, tracker *suspensionTracker) measuredTimeProvider { - return measuredTimeProvider{TimeProvider: tp, clock: clock, tracker: tracker} -} - -func (m measuredTimeProvider) GetDONTime() (time.Time, error) { - start := m.clock.Now() - t, err := m.TimeProvider.GetDONTime() - m.tracker.add(m.clock.Now().Sub(start)) - return t, err -} diff --git a/core/services/workflows/v2/compute_metering_test.go b/core/services/workflows/v2/compute_metering_test.go deleted file mode 100644 index a48ed9f6552..00000000000 --- a/core/services/workflows/v2/compute_metering_test.go +++ /dev/null @@ -1,74 +0,0 @@ -package v2 - -import ( - "testing" - "time" - - "github.com/jonboulle/clockwork" - "github.com/stretchr/testify/require" -) - -// stubTimeProvider simulates a TimeProvider whose DON-time fetch blocks on -// consensus by advancing a fake clock for the duration of the call. -type stubTimeProvider struct { - clock *clockwork.FakeClock - donBlock time.Duration - donErr error -} - -func (s *stubTimeProvider) GetNodeTime() time.Time { return s.clock.Now() } - -func (s *stubTimeProvider) GetDONTime() (time.Time, error) { - s.clock.Advance(s.donBlock) // simulate blocking on the DON-time consensus round - return s.clock.Now(), s.donErr -} - -func TestSuspensionTracker_SumsSequentialIntervals(t *testing.T) { - t.Parallel() - - var tr suspensionTracker - tr.add(100 * time.Millisecond) - tr.add(0) // ignored - tr.add(-5 * time.Millisecond) // ignored - tr.add(250 * time.Millisecond) - require.Equal(t, 350*time.Millisecond, tr.total()) -} - -func TestSuspensionTracker_NilSafe(t *testing.T) { - t.Parallel() - - var tr *suspensionTracker - tr.add(time.Second) // must not panic - require.Equal(t, time.Duration(0), tr.total()) -} - -func TestMeasuredTimeProvider_RecordsDONTimeWait(t *testing.T) { - t.Parallel() - - clock := clockwork.NewFakeClock() - tracker := &suspensionTracker{} - stub := &stubTimeProvider{clock: clock, donBlock: 10 * time.Second} - mtp := newMeasuredTimeProvider(stub, clock, tracker) - - _, err := mtp.GetDONTime() - require.NoError(t, err) - require.Equal(t, 10*time.Second, tracker.total(), "DON-time wait should be recorded as guest suspension") - - // A second blocking call accumulates on top of the first. - stub.donBlock = 2 * time.Second - _, err = mtp.GetDONTime() - require.NoError(t, err) - require.Equal(t, 12*time.Second, tracker.total()) -} - -func TestMeasuredTimeProvider_NodeTimeNotMeasured(t *testing.T) { - t.Parallel() - - clock := clockwork.NewFakeClock() - tracker := &suspensionTracker{} - stub := &stubTimeProvider{clock: clock} - mtp := newMeasuredTimeProvider(stub, clock, tracker) - - _ = mtp.GetNodeTime() // local, instant; not a guest suspension - require.Equal(t, time.Duration(0), tracker.total()) -} diff --git a/core/services/workflows/v2/compute_usage_test.go b/core/services/workflows/v2/compute_usage_test.go new file mode 100644 index 00000000000..ea27fd0d621 --- /dev/null +++ b/core/services/workflows/v2/compute_usage_test.go @@ -0,0 +1,297 @@ +package v2_test + +import ( + "context" + "errors" + "strconv" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "go.uber.org/zap/zapcore" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + confworkflowtypes "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/actions/confidentialworkflow" + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/resourcemanager" + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" + regmocks "github.com/smartcontractkit/chainlink-common/pkg/types/core/mocks" + "github.com/smartcontractkit/chainlink-common/pkg/workflows/host" + modulemocks "github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host/mocks" + sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" + valuespb "github.com/smartcontractkit/chainlink-protos/cre/go/values/pb" + meteringpb "github.com/smartcontractkit/chainlink-protos/metering/go" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/confidentialrelay" + capmocks "github.com/smartcontractkit/chainlink/v2/core/capabilities/mocks" + v2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2" + "github.com/smartcontractkit/chainlink/v2/core/utils/matches" +) + +// recordingEmitter captures MeterRecords emitted by a ResourceManager. +type recordingEmitter struct { + mu sync.Mutex + records []*meteringpb.MeterRecord +} + +func (r *recordingEmitter) Emit(_ context.Context, body []byte, _ ...any) error { + var rec meteringpb.MeterRecord + if err := proto.Unmarshal(body, &rec); err != nil { + return err + } + r.mu.Lock() + defer r.mu.Unlock() + r.records = append(r.records, &rec) + return nil +} + +func (r *recordingEmitter) all() []*meteringpb.MeterRecord { + r.mu.Lock() + defer r.mu.Unlock() + return append([]*meteringpb.MeterRecord(nil), r.records...) +} + +// moduleRun is what the mocked module does for the one trigger execution. +type moduleRun struct { + result *sdkpb.ExecutionResult + err error + onExecute func() +} + +// runOneExecution starts an engine with cfg, fires one trigger event, and +// returns the execution id. +func runOneExecution(t *testing.T, cfg *v2.EngineConfig, run moduleRun) string { + module := modulemocks.NewModuleV2(t) + module.EXPECT().Start() + module.EXPECT().Close() + capreg := regmocks.NewCapabilitiesRegistry(t) + capreg.EXPECT().LocalNode(matches.AnyContext).Return(newNode(t), nil) + + initDoneCh := make(chan error) + subscribedCh := make(chan []string, 1) + finishedCh := make(chan string) + cfg.Module = module + cfg.CapRegistry = capreg + cfg.BillingClient = setupMockBillingClient(t) + cfg.OrgResolver = &mockOrgResolver{orgID: "org-123"} + cfg.Hooks = v2.LifecycleHooks{ + OnInitialized: func(err error) { initDoneCh <- err }, + OnSubscribedToTriggers: func(ids []string) { subscribedCh <- ids }, + OnExecutionFinished: func(id string, _ string) { finishedCh <- id }, + } + + engine, err := v2.NewEngine(cfg) + require.NoError(t, err) + + module.EXPECT().Execute(matches.AnyContext, mock.Anything, mock.Anything).Return(newTriggerSubs(1), nil).Once() + trigger := capmocks.NewTriggerCapability(t) + capreg.EXPECT().GetTrigger(matches.AnyContext, "id_0").Return(trigger, nil).Once() + eventCh := make(chan capabilities.TriggerResponse) + trigger.EXPECT().RegisterTrigger(matches.AnyContext, mock.Anything).Return(eventCh, nil).Once() + trigger.EXPECT().UnregisterTrigger(matches.AnyContext, mock.Anything).Return(nil).Once() + trigger.EXPECT().AckEvent(matches.AnyContext, mock.Anything, mock.Anything, mock.Anything).Return(nil) + + require.NoError(t, engine.Start(t.Context())) + require.NoError(t, <-initDoneCh) + require.Equal(t, []string{"id_0"}, <-subscribedCh) + + module.EXPECT().Execute(matches.AnyContext, mock.Anything, mock.Anything).Run(func(context.Context, *sdkpb.ExecuteRequest, host.ExecutionHelper) { + if run.onExecute != nil { + run.onExecute() + } + }).Return(run.result, run.err).Once() + eventCh <- capabilities.TriggerResponse{Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "usage_event"}} + executionID := <-finishedCh + require.NoError(t, engine.Close()) + return executionID +} + +func TestEngine_ComputeUsageMeterRecord(t *testing.T) { + t.Parallel() + + t.Run("emits one compute record per execution and logs the contract line", func(t *testing.T) { + t.Parallel() + lggr, obs := logger.TestObserved(t, zapcore.InfoLevel) + cfg := defaultTestConfig(t, nil) + cfg.Lggr = lggr + emitter := &recordingEmitter{} + cfg.MeterRecordsEnabled = true + cfg.UsageEmitter = emitter + cfg.MeteringIdentity = resourcemanager.DeploymentIdentity{Product: "cre"} + + executionID := runOneExecution(t, cfg, moduleRun{}) + + records := emitter.all() + require.Len(t, records, 1) + rec := records[0] + require.Equal(t, meteringpb.MeterAction_METER_ACTION_USAGE, rec.GetAction()) + require.Equal(t, resourcemanager.EmittingServiceWorkflowEngine, rec.GetIdentity().GetService()) + require.Equal(t, "cre:workflow:compute", rec.GetIdentity().GetResourcePool()) + require.Equal(t, "cre:workflow:compute", rec.GetIdentity().GetResourcePoolId()) + require.NotEmpty(t, rec.GetIdentity().GetDon().GetDonId(), "don id must be stamped from the local node") + require.Len(t, rec.GetUtilizations(), 1) + u := rec.GetUtilizations()[0] + require.Equal(t, resourcemanager.ResourceTypeWorkflowCompute, u.GetResourceType()) + require.Equal(t, cfg.WorkflowID+":"+executionID, u.GetResourceId()) + require.Equal(t, executionID, u.GetEventId(), "compute capability event id is the execution id") + require.Equal(t, "org-123", u.GetOrgId()) + require.NotEmpty(t, u.GetValue()) + + logs := obs.FilterMessage("Emitted capability usage meter record").All() + require.Len(t, logs, 1) + fields := logs[0].ContextMap() + require.Equal(t, executionID, fields["eventID"]) + require.Equal(t, resourcemanager.ResourceTypeWorkflowCompute, fields["resourceType"]) + require.Equal(t, "org-123", fields["orgID"]) + require.Contains(t, fields, "value") + }) + + t.Run("emits for failed executions too", func(t *testing.T) { + t.Parallel() + cfg := defaultTestConfig(t, nil) + emitter := &recordingEmitter{} + cfg.MeterRecordsEnabled = true + cfg.UsageEmitter = emitter + cfg.MeteringIdentity = resourcemanager.DeploymentIdentity{Product: "cre"} + + runOneExecution(t, cfg, moduleRun{err: errors.New("module failed")}) + + require.Len(t, emitter.all(), 1) + }) + + t.Run("bills the compute duration the module reports", func(t *testing.T) { + t.Parallel() + cfg := defaultTestConfig(t, nil) + emitter := &recordingEmitter{} + cfg.MeterRecordsEnabled = true + cfg.UsageEmitter = emitter + cfg.MeteringIdentity = resourcemanager.DeploymentIdentity{Product: "cre"} + runOneExecution(t, cfg, moduleRun{result: &sdkpb.ExecutionResult{ComputeDurationMs: proto.Int64(1234)}}) + + records := emitter.all() + require.Len(t, records, 1) + require.Equal(t, "1234", records[0].GetUtilizations()[0].GetValue()) + }) + + t.Run("falls back to wall-clock when the module reports no duration", func(t *testing.T) { + t.Parallel() + cfg := defaultTestConfig(t, nil) + emitter := &recordingEmitter{} + cfg.MeterRecordsEnabled = true + cfg.UsageEmitter = emitter + cfg.MeteringIdentity = resourcemanager.DeploymentIdentity{Product: "cre"} + + runOneExecution(t, cfg, moduleRun{onExecute: func() { time.Sleep(20 * time.Millisecond) }}) + + records := emitter.all() + require.Len(t, records, 1) + ms, err := strconv.ParseInt(records[0].GetUtilizations()[0].GetValue(), 10, 64) + require.NoError(t, err) + require.GreaterOrEqual(t, ms, int64(20)) + }) + + t.Run("emits nothing when metering is disabled", func(t *testing.T) { + t.Parallel() + lggr, obs := logger.TestObserved(t, zapcore.InfoLevel) + cfg := defaultTestConfig(t, nil) + cfg.Lggr = lggr + + runOneExecution(t, cfg, moduleRun{}) + + require.Empty(t, obs.FilterMessage("Emitted capability usage meter record").All()) + }) +} + +// A TEE-routed trigger is metered from the duration the enclave set on the result. +func TestEngine_ComputeUsageMeterRecord_ConfidentialRun(t *testing.T) { + t.Parallel() + const capID = "confidential-workflows@1.0.0-alpha" + lggr := logger.Test(t) + cfg := defaultTestConfig(t, nil) + emitter := &recordingEmitter{} + cfg.MeterRecordsEnabled = true + cfg.UsageEmitter = emitter + cfg.MeteringIdentity = resourcemanager.DeploymentIdentity{Product: "cre"} + + capreg := regmocks.NewCapabilitiesRegistry(t) + capreg.EXPECT().LocalNode(matches.AnyContext).Return(newNode(t), nil) + execCap := capmocks.NewExecutableCapability(t) + capreg.EXPECT().GetExecutable(matches.AnyContext, capID).Return(execCap, nil) + + tees, err := anypb.New(&confworkflowtypes.ProvidedTeesResponse{Tee: []*sdkpb.TeeTypeAndRegions{ + {Type: sdkpb.TeeType_TEE_TYPE_AWS_NITRO, Regions: []string{"us-east-1"}}, + }}) + require.NoError(t, err) + execCap.EXPECT().Execute(matches.AnyContext, mock.MatchedBy(func(req capabilities.CapabilityRequest) bool { + return req.Method == "ProvidedTees" + })).Return(capabilities.CapabilityResponse{Payload: tees}, nil).Once() + // The enclave runs the same WASM host, which sets compute_duration_ms. + enclaveResp, err := anypb.New(&confworkflowtypes.ConfidentialWorkflowResponse{SdkExecutionResult: &sdkpb.ExecutionResult{ + Result: &sdkpb.ExecutionResult_Value{Value: valuespb.NewStringValue("enclave-output")}, + ComputeDurationMs: proto.Int64(1234), + }}) + require.NoError(t, err) + execCap.EXPECT().Execute(matches.AnyContext, mock.MatchedBy(func(req capabilities.CapabilityRequest) bool { + return req.Method == "Execute" + })).Run(func(context.Context, capabilities.CapabilityRequest) { + time.Sleep(20 * time.Millisecond) // round trip the engine must not bill + }).Return(capabilities.CapabilityResponse{Payload: enclaveResp}, nil).Once() + + conf, err := v2.NewConfidentialModule(capreg, &confidentialrelay.ExecutionHandlers{}, "https://example.com/binary.wasm", []byte("hash"), + cfg.WorkflowID, cfg.WorkflowOwner, cfg.WorkflowName.String(), cfg.WorkflowTag, + func(context.Context, string) (string, error) { return "org-123", nil }, + limits.NewGateLimiter(true), cfg.LocalLimiters.Settings, lggr) + require.NoError(t, err) + + local := modulemocks.NewModuleV2(t) + local.EXPECT().Start().Maybe() + local.EXPECT().Close().Maybe() + teeSubs := newTriggerSubs(1) + teeSubs.GetTriggerSubscriptions().Subscriptions[0].Requirements = &sdkpb.Requirements{Tee: &sdkpb.Tee{ + Item: &sdkpb.Tee_AnyRegions{AnyRegions: &sdkpb.Regions{Regions: []string{"us-east-1"}}}, + }} + // Subscribe runs locally; the trigger is routed to the enclave. + local.EXPECT().Execute(matches.AnyContext, mock.Anything, mock.Anything).Return(teeSubs, nil).Once() + cfg.Module = host.NewRequirementSelectingModule(host.ModuleAndHandler{Module: local}, []host.ModuleAndHandler{{ + Module: conf, + RequirementsHandler: host.RequirementsHandler{Tee: conf.Tee}, + }}) + cfg.CapRegistry = capreg + cfg.BillingClient = setupMockBillingClient(t) + cfg.OrgResolver = &mockOrgResolver{orgID: "org-123"} + + initDoneCh := make(chan error) + subscribedCh := make(chan []string, 1) + finishedCh := make(chan string) + cfg.Hooks = v2.LifecycleHooks{ + OnInitialized: func(err error) { initDoneCh <- err }, + OnSubscribedToTriggers: func(ids []string) { subscribedCh <- ids }, + OnExecutionFinished: func(id string, _ string) { finishedCh <- id }, + } + engine, err := v2.NewEngine(cfg) + require.NoError(t, err) + + trigger := capmocks.NewTriggerCapability(t) + capreg.EXPECT().GetTrigger(matches.AnyContext, "id_0").Return(trigger, nil).Once() + eventCh := make(chan capabilities.TriggerResponse) + trigger.EXPECT().RegisterTrigger(matches.AnyContext, mock.Anything).Return(eventCh, nil).Once() + trigger.EXPECT().UnregisterTrigger(matches.AnyContext, mock.Anything).Return(nil).Once() + trigger.EXPECT().AckEvent(matches.AnyContext, mock.Anything, mock.Anything, mock.Anything).Return(nil) + + require.NoError(t, engine.Start(t.Context())) + require.NoError(t, <-initDoneCh) + require.Equal(t, []string{"id_0"}, <-subscribedCh) + + eventCh <- capabilities.TriggerResponse{Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "confidential_event"}} + executionID := <-finishedCh + require.NoError(t, engine.Close()) + + records := emitter.all() + require.Len(t, records, 1) + u := records[0].GetUtilizations()[0] + require.Equal(t, cfg.WorkflowID+":"+executionID, u.GetResourceId()) + require.Equal(t, "1234", u.GetValue(), "enclave-measured duration, not the round trip") +} diff --git a/core/services/workflows/v2/config.go b/core/services/workflows/v2/config.go index a82fa864232..c56d687ce5f 100644 --- a/core/services/workflows/v2/config.go +++ b/core/services/workflows/v2/config.go @@ -15,6 +15,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" @@ -70,6 +71,13 @@ type EngineConfig struct { // OrgResolver is used to resolve organization IDs from workflow owners OrgResolver orgresolver.OrgResolver + // MeterRecordsEnabled emits a cre:workflow:compute usage record per execution. + MeterRecordsEnabled bool + // MeteringIdentity is the node's [Metering] deployment dimensions; the engine adds service, pool and DON. + MeteringIdentity resourcemanager.DeploymentIdentity + // UsageEmitter delivers compute usage records; required when MeterRecordsEnabled. + UsageEmitter resourcemanager.Emitter + // includes additional logging of events internal to user workflows DebugMode bool @@ -503,6 +511,9 @@ func (c *EngineConfig) Validate() error { if c.Clock == nil { c.Clock = clockwork.NewRealClock() } + if c.MeterRecordsEnabled && c.UsageEmitter == nil { + return errors.New("usage emitter not set") + } _, err := types.WorkflowIDFromHex(c.WorkflowID) if err != nil { diff --git a/plugins/plugins.public.yaml b/plugins/plugins.public.yaml index 5f87830bfa8..3a6ded6d6bf 100644 --- a/plugins/plugins.public.yaml +++ b/plugins/plugins.public.yaml @@ -84,12 +84,12 @@ plugins: capability-evm: - moduleURI: "github.com/smartcontractkit/capabilities/chain_capabilities/evm" - gitRef: "e6f512fbf3bdc3adaa96bfdd1d497c4507eab05d" + gitRef: "2951f2cf06c3c0d52fe8b738764218cdb571635c" installPath: "." capability-solana: - moduleURI: "github.com/smartcontractkit/capabilities/chain_capabilities/solana" - gitRef: "7ae963289733be8764d0b49e8e4c6233bd34680b" + gitRef: "2951f2cf06c3c0d52fe8b738764218cdb571635c" installPath: "." capability-aptos: diff --git a/system-tests/tests/canaries_sentinels/proof-of-reserve/cron-based/go.mod b/system-tests/tests/canaries_sentinels/proof-of-reserve/cron-based/go.mod index 2dc528037c6..5900cbd99ce 100644 --- a/system-tests/tests/canaries_sentinels/proof-of-reserve/cron-based/go.mod +++ b/system-tests/tests/canaries_sentinels/proof-of-reserve/cron-based/go.mod @@ -5,8 +5,8 @@ go 1.27.1 require ( github.com/ethereum/go-ethereum v1.17.4 github.com/smartcontractkit/chain-selectors v1.0.107 - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260923161343-170ea96c3269 - github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260804191526-b7a850ae7648 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015152047-f1f2b6d3bd57 + github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261014151355-5868490cb3e3 github.com/smartcontractkit/cre-sdk-go v1.5.0 github.com/smartcontractkit/cre-sdk-go/capabilities/blockchain/evm v0.10.0 github.com/smartcontractkit/cre-sdk-go/capabilities/networking/http v1.3.0 @@ -53,7 +53,7 @@ require ( github.com/prometheus/procfs v0.20.1 // indirect github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 // indirect github.com/shopspring/decimal v1.4.0 // indirect - github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260915184316-2730f1867c92 // indirect + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 // indirect github.com/smartcontractkit/libocr v0.0.0-20260810200708-618b5bf7f342 // indirect github.com/stretchr/testify v1.12.1 // indirect github.com/wk8/go-ordered-map/v2 v2.1.8 // indirect diff --git a/system-tests/tests/canaries_sentinels/proof-of-reserve/cron-based/go.sum b/system-tests/tests/canaries_sentinels/proof-of-reserve/cron-based/go.sum index e0b9666a7d9..88eab15e419 100644 --- a/system-tests/tests/canaries_sentinels/proof-of-reserve/cron-based/go.sum +++ b/system-tests/tests/canaries_sentinels/proof-of-reserve/cron-based/go.sum @@ -100,12 +100,12 @@ github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= github.com/smartcontractkit/chain-selectors v1.0.107 h1:LlQI72DqaFrJXWRmcv78H48MjUNhFzKm+xBOt+5ZXbg= github.com/smartcontractkit/chain-selectors v1.0.107/go.mod h1:qy7whtgG5g+7z0jt0nRyii9bLND9m15NZTzuQPkMZ5w= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260923161343-170ea96c3269 h1:Y67MFgyBUo08lAFSVGMRY0smRJ9SD+csX44BHCGOPU4= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260923161343-170ea96c3269/go.mod h1:BniOJQOh52lgDDNyfOQJEmQUVQ68txTSeT0Qxwt0zQI= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260915184316-2730f1867c92 h1:Z1OIGukDhJD+Gknr5G07Vtml24E7BwKaIQoWzV/SdqA= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260915184316-2730f1867c92/go.mod h1:Scyk7O80fde5rJF3Ty3fEE3xiS6j84eWbu8/2gl3pIY= -github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260804191526-b7a850ae7648 h1:WEUMkKQPAgcNMRgES6CBWrRUiII+HKEWQjulKQBSuMA= -github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260804191526-b7a850ae7648/go.mod h1:/i8hjTPFdVWHiY+QjeSiVS2Z3GB3WAZznGgXHstC02E= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015152047-f1f2b6d3bd57 h1:71QCghDEu9+vSDAjhrUrIzL01MlrIOWOW8gABXuBtMg= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015152047-f1f2b6d3bd57/go.mod h1:QeHe1mlcVzzg+awYAkRfl/8mDLRAhsl18IIOg4MVZec= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 h1:J4vYRJxkBO8Yf//17wScKjYWmS45VZaFWx0zt+4ETEU= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549/go.mod h1:Scyk7O80fde5rJF3Ty3fEE3xiS6j84eWbu8/2gl3pIY= +github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261014151355-5868490cb3e3 h1:rUUx0roqc0jAWDpdqL+oMvZiUORyYrGEGji0PJaXwjI= +github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261014151355-5868490cb3e3/go.mod h1:/i8hjTPFdVWHiY+QjeSiVS2Z3GB3WAZznGgXHstC02E= github.com/smartcontractkit/cre-sdk-go v1.5.0 h1:kepW3QDKARrOOHjXwWAZ9j5KLk6bxLzvi6OMrLsFwVo= github.com/smartcontractkit/cre-sdk-go v1.5.0/go.mod h1:yYrQFz1UH7hhRbPO0q4fgo1tfsJNd4yXnI3oCZE0RzM= github.com/smartcontractkit/cre-sdk-go/capabilities/blockchain/evm v0.10.0 h1:G0w0cLzHy/5m74IzSGz1Ynjffym4ZxLeUrRLp8EFP5w=