Helpers for Resource Managers#2252
Draft
patrickhuie19 wants to merge 6 commits into
Draft
Conversation
…pers Reshape the resourcemanager metering API around the delta-based contract and add the shared helpers that kill the per-producer triplication. API (breaking, intentional): - Replace EmitMeterRecord with intent-named EmitDelta (METER_ACTION_UPDATE, signed delta) and EmitUsage (METER_ACTION_USAGE, one-off consumption) so producers cannot emit the deprecated RESERVE/RELEASE actions. Both fail-open. - event_id is generated by the manager (uuid.NewString) per emission and stamped on every emitted Utilization for both records and snapshots; remove EventID from UtilizationFields so producers cannot supply it. - NewUtilizationInt documents/accepts negative values. Snapshot loop hardening (resourcemanager.go): - Align the first tick to the next wall-clock interval boundary and truncate the snapshot timestamp to SnapshotInterval so cross-node buckets agree; record timestamps stay raw edges. - Run each registrant under a per-tick context deadline (SnapshotInterval/4, floor 5s); a slow registrant is logged and skipped, never stalling the tick. - Rewrite package docs to the contract framing (records + snapshots are both load-bearing); drop RESERVE/RELEASE pairing prose. Shared helpers: - resourcemanager.ConfigFromEnv(*loop.EnvConfig) Config + Config bundling the loop-env -> ResourceManagerConfig + DeploymentIdentity mapping. - resourcemanager.NewBaseIdentity(dep, capDONID, service, resourcePool) and (ResourceIdentity).WithWorkflowDonFallback(workflowDonID) encapsulating the product-fallback, DON-ID formatting, nil-Don, and CapDONID-authoritative rules. - orgresolver.NewCaching(inner, refreshInterval) background-refreshing resolver for the no-network snapshot path, and orgresolver.ResolveOrEmpty(...) canonical fail-open resolution helper. Tests updated for the new API; config_test.go MeterNodeID fixture is now a logical node name. go.mod adds a local replace to the sibling chainlink-protos metering module for parallel development.
decodeSnapshots already filtered snapshot emit calls by entity; also assert the beholder_domain attribute is cll-meter so the snapshot surface is explicitly verified alongside the existing MeterRecord domain assertion.
ConfigFromEnv now wires the process-global durable emitter (durableemitter.GetGlobalEmitter()) instead of the standard beholder emitter (beholder.GetEmitter()). Metering records/snapshots are counter-semantic billing events, so they must ride the durable queue for at-least-once delivery; the consumer dedups by event_id. The durable emitter is initialized once per process by durableemitter.Setup (the LOOP server does this at startup). If it has not been initialized, GetGlobalEmitter returns a nil *DurableEmitter; ConfigFromEnv assigns it through a local Emitter so an unconfigured emitter yields a true nil interface (not a non-nil interface wrapping a nil pointer). A nil emitter makes the ResourceManager a no-op, consistent with the fail-open posture. Also picks up the merged chainlink-protos/metering pin (go.mod pin bump, local replace dropped) and the corresponding go.sum entry so the module builds against the published module.
Records: event_id is a required, producer-supplied parameter on EmitDelta and
EmitUsage, derived deterministically from the triggering request so every DON
node fielding the same logical delta emits the identical value (the consumer's
cross-node dedup/aggregation key). The RM no longer generates a per-node id for
records — that made each node emit a different event_id for the same delta, so
the billing consumer could not dedup/aggregate across a multi-node DON.
No fallback: an empty event_id is logged, increments the emit-failure counter
(reason="missing_event_id"), and SKIPS the emit; a random id is never generated.
Fail-open for the resource operation is preserved.
Signatures (breaking):
EmitDelta(ctx, identity, eventID string, delta int64, fields UtilizationFields)
EmitUsage(ctx, identity, eventID string, quantity int64, fields UtilizationFields)
Snapshots: event_id is RM-derived, a deterministic per-node/per-bucket key:
snapshot:{node_id}:{service}:{resource_pool}:{resource_type}:{resource_id}:{bucket_unix}
Adds shared helpers EventID(namespace, parts...) and SnapshotEventID(...). event_id
is an opaque string with no format guarantee and no validation beyond
empty-vs-nonempty. Tests: identical event_id across nodes for the same request,
distinct across requests, empty event_id skips the emit, snapshot key stability.
Contributor
|
patrickhuie19
force-pushed
the
feat/metering-v2-cll-meter
branch
from
9eb1823 to
0ed829b
Compare
jmank88
reviewed
Jul 14, 2026
Comment on lines
+243
to
258
| // Align the first tick to the next wall-clock multiple of the interval, | ||
| // then run on the interval boundary thereafter. For DonTime-synced | ||
| // clocks this makes cross-node snapshot buckets agree structurally: the | ||
| // emitted snapshot timestamp is the tick time truncated to the interval. | ||
| now := rm.clock.Now() | ||
| firstTick := now.Truncate(rm.snapshotInterval).Add(rm.snapshotInterval) | ||
| timer := rm.clock.NewTimer(firstTick.Sub(now)) | ||
| select { | ||
| case <-ctx.Done(): | ||
| timer.Stop() | ||
| return | ||
| case <-timer.Chan(): | ||
| } | ||
| rm.emitSnapshots(ctx, firstTick) | ||
|
|
||
| ticker := rm.clock.NewTicker(rm.snapshotInterval) |
Contributor
There was a problem hiding this comment.
Could you use a services.TickerConfig? Or I guess there is a fake clock involved here too?
Contributor
Author
There was a problem hiding this comment.
Or I guess there is a fake clock involved here too?
yes, exactly. so the services.TickerConfig wouldn't propagate to tests appropriately ifiuc
jmank88
reviewed
Jul 14, 2026
Comment on lines
+320
to
+330
| func(m Meterable) { | ||
| rctx, cancel := context.WithTimeout(ctx, timeout) | ||
| defer cancel() | ||
| rm.emitSnapshot(rctx, m, tickTime) | ||
| if rctx.Err() != nil && ctx.Err() == nil { | ||
| rm.lggr.Warnw("snapshot registrant exceeded its per-tick deadline; skipped", | ||
| "timeout", timeout, | ||
| "err", rctx.Err(), | ||
| ) | ||
| } | ||
| }(m) |
Contributor
There was a problem hiding this comment.
If you add a context param then you can use normal naming within the func:
Suggested change
| func(m Meterable) { | |
| rctx, cancel := context.WithTimeout(ctx, timeout) | |
| defer cancel() | |
| rm.emitSnapshot(rctx, m, tickTime) | |
| if rctx.Err() != nil && ctx.Err() == nil { | |
| rm.lggr.Warnw("snapshot registrant exceeded its per-tick deadline; skipped", | |
| "timeout", timeout, | |
| "err", rctx.Err(), | |
| ) | |
| } | |
| }(m) | |
| func(ctx context.Context, m Meterable) { | |
| var cancel context.CancelFn | |
| ctx, cancel = context.WithTimeout(ctx, timeout) | |
| defer cancel() | |
| rm.emitSnapshot(ctx, m, tickTime) | |
| if ctx.Err() == nil { | |
| rm.lggr.Warnw("snapshot registrant exceeded its per-tick deadline; skipped", | |
| "timeout", timeout, | |
| "err", ctx.Err(), | |
| ) | |
| } | |
| }(ctx, m) |
No need to check both errors either. The parent will propogate.
jmank88
reviewed
Jul 14, 2026
|
|
||
| import ( | ||
| "github.com/smartcontractkit/chainlink-common/pkg/durableemitter" | ||
| "github.com/smartcontractkit/chainlink-common/pkg/loop" |
Contributor
There was a problem hiding this comment.
It looks like this convenience func is the only reason for this import - can we try to avoid it? Or even just reversed might be better (e.g. move it to a method on loop.EnvConfig)
jmank88
reviewed
Jul 14, 2026
| // durable emitter yields a true nil interface (not a non-nil interface | ||
| // wrapping a nil pointer), which the ResourceManager treats as no-op. | ||
| var emitter Emitter | ||
| if de := durableemitter.GetGlobalEmitter(); de != nil { |
Contributor
There was a problem hiding this comment.
Can we inject this to avoid calling a global func from such a low level helper?
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
ResourceManagers share needs around populating deployment + resource identities. There is also shared code like generating event_ids for snapshots or deltas.