resourcemanager

Documentation

Overview

Package resourcemanager emits metering.v1 events for durable billable resources such as trigger registrations, workflow specs, and log filters.

It emits two kinds of records, each covering exactly one resource identified by its ResourceIdentity:

• MeterRecord lifecycle edges, inline, via EmitMeterRecord at the point a resource is reserved, released, or its utilization changes; and
• MeterSnapshot records, on a timer, one per resource a registered Meterable reports as currently active. Snapshots are the liveness/utilization-over-time signal that pure RESERVE/RELEASE cannot provide (a node panic would otherwise leak a reservation forever).

The ResourceManager is the single owner of the snapshot tick: each producer starts the manager as a sub-service and only Registers itself; producers never run their own snapshot loop.

Emission is fail-open by design: EmitMeterRecord and the snapshot loop return no error, and a metering failure must never gate, delay, or retry the resource operation being metered. Failures surface via error-level logs and the resource_manager_*_failure_total counters; billing correctness is recovered downstream through idempotency keys and reconciliation.

Index

• Constants
• func NewUtilization(value string, fields UtilizationFields) *meteringpb.Utilization
• func NewUtilizationBig(value *big.Int, fields UtilizationFields) *meteringpb.Utilization
• func NewUtilizationFloat(value float64, fields UtilizationFields) *meteringpb.Utilization
• func NewUtilizationInt(value int64, fields UtilizationFields) *meteringpb.Utilization
• type DeploymentIdentity
• type DonIdentity
• type Emitter
• type Meterable
• type ResourceIdentity
  • func (r ResourceIdentity) DonID() string
  • func (r ResourceIdentity) NodeID() string
• type ResourceManager
  • func NewResourceManager(lggr logger.Logger, cfg ResourceManagerConfig) *ResourceManager
  • func (rm *ResourceManager) EmitMeterRecord(ctx context.Context, identity ResourceIdentity, action meteringpb.MeterAction, ...)
  • func (rm *ResourceManager) Register(m Meterable) (unregister func())
• type ResourceManagerConfig
• type SnapshotEntry
• type UtilizationFields

Constants

const DefaultSnapshotInterval = 60 * time.Second

DefaultSnapshotInterval is the recommended snapshot period. It is NOT applied implicitly: a zero ResourceManagerConfig.SnapshotInterval disables snapshots. Callers that want snapshots pass this (or their own value) explicitly.

Functions

func NewUtilization

func NewUtilization(value string, fields UtilizationFields) *meteringpb.Utilization

NewUtilization builds a Utilization from a pre-formatted numeric string value.

func NewUtilizationBig

func NewUtilizationBig(value *big.Int, fields UtilizationFields) *meteringpb.Utilization

NewUtilizationBig builds a Utilization from an arbitrary-precision integer.

func NewUtilizationFloat

func NewUtilizationFloat(value float64, fields UtilizationFields) *meteringpb.Utilization

NewUtilizationFloat builds a Utilization from a floating-point value.

func NewUtilizationInt

func NewUtilizationInt(value int64, fields UtilizationFields) *meteringpb.Utilization

NewUtilizationInt builds a Utilization with int64 quantity encoded as a decimal string value.

Types

type DeploymentIdentity

type DeploymentIdentity struct {
	// Product is the deployment product, e.g. "cre".
	Product string
	// Tenant is the human-readable deployment tenant name, e.g. "mainline" or
	// "enterprise".
	Tenant string
	// NumericTenantID is the numbered tenant identifier as a string.
	NumericTenantID string
	// Environment is the deployment environment, e.g. "production", "staging".
	Environment string
	// Zone is the deployment zone, e.g. "wf-zone-a".
	Zone string
	// NodeID is the node's logical name, e.g. "clp-cre-wf-zone-a-1". It is NOT
	// the CSA public key; it is a stable name the billing service can use to
	// look up the node's CSA key in the workflow registry. The CSA key itself is
	// carried separately as the node_csa_key event attribute.
	NodeID string
}

DeploymentIdentity carries the static deployment + node identity dimensions that are fixed for a LOOP plugin process. They are resolved once from node config by the host and delivered to every LOOP plugin over the environment (loop.EnvConfig), not the standard-capabilities boundary, so any LOOP plugin can populate the coarse metering rollup dimensions. Any field may be empty if the host did not provide it.

type DonIdentity

type DonIdentity struct {
	// DonID is the DON ID the emitting service belongs to.
	DonID string
	// NodeID is the node's logical name within the scope of the DON, e.g.
	// "clp-cre-wf-zone-a-1". It is a human-readable ID, NOT the CSA
	// public key. The prefix can be redundant with other fully-qualified
	// dimensions, but helps readability. The CSA key is emitted separately via
	// the node_csa_key attribute.
	NodeID string
}

DonIdentity captures DON-specific identity dimensions as one unit.

type Emitter

type Emitter interface {
	Emit(ctx context.Context, body []byte, attrKVs ...any) error
}

Emitter delivers an encoded metering message with its routing attributes. beholder.Emitter satisfies it, so production wiring is beholder.GetEmitter(); tests substitute a fake.

type Meterable

type Meterable interface {
	// ResourceIdentity returns the producer's base identity: the coarse
	// dimensions (product, tenant, numeric_tenant_id, environment, zone, don, service) plus
	// the service-level resource_pool / resource_pool_id. Per-resource billing
	// fields (resource_type/resource_id/org_id/event_id/value) are carried by
	// Utilizations on MeterRecord and MeterSnapshot.
	ResourceIdentity() ResourceIdentity

	// GetUtilization returns the current level of the producer's currently
	// active resources, one SnapshotEntry per resource. The manager emits one
	// MeterSnapshot per entry.
	//
	// It is called on the snapshot tick and MUST be a cheap, non-blocking
	// read-snapshot of in-memory state: no network, no disk, no lock held
	// across I/O. It must tolerate ctx cancellation (returning promptly, and
	// nil/empty is acceptable) and tolerate concurrent registration of new
	// resources. An empty or nil return is valid and means nothing is currently
	// active: no snapshots are emitted, and billing zeroes the resource out by
	// its absence from subsequent snapshots.
	GetUtilization(ctx context.Context) []SnapshotEntry
}

Meterable is implemented by producers that manage durable billable resources (trigger registrations, workflow specs, log filters). A producer registers itself with a ResourceManager (see ResourceManager.Register) so it is polled once per snapshot tick for the absolute state of its currently active resources, in addition to emitting lifecycle edges inline via EmitMeterRecord.

type ResourceIdentity

type ResourceIdentity struct {
	// Product is the deployment product, e.g. "cre".
	Product string

	// Tenant is the human-readable deployment tenant name, e.g. "mainline" or
	// "enterprise".
	Tenant string

	// NumericTenantID is the numbered tenant identifier as a string.
	NumericTenantID string

	// Environment is the deployment environment, e.g. "production",
	// "staging".
	Environment string

	// Zone is the deployment zone, e.g. "wf-zone-a".
	Zone string

	// Don groups DON-specific identity dimensions so consumers can
	// branch on one struct instead of handling don/node permutations.
	Don *DonIdentity

	// Service is the stable service constant identifying the emitting service,
	// e.g. "cron-trigger", "http-trigger", "evm-log-trigger",
	// "workflow-syncer-v2". It must not
	// encode deployment environment or zone.
	Service string

	// ResourcePool is the service-level resource pool the record applies to,
	// e.g. "trigger_registrations", "log_filters", "workflow_storage".
	ResourcePool string

	// ResourcePoolID optionally scopes identity further within the resource pool.
	ResourcePoolID string
}

ResourceIdentity is the structured, first-class identity of a durable resource. Its fields map 1:1 to metering.v1.ResourceIdentity so every emitted record carries each dimension as a discrete column rather than a parsed dotted string or out-of-band telemetry attribute.

func (ResourceIdentity) DonID

func (r ResourceIdentity) DonID() string

DonID returns the DON ID when present.

func (ResourceIdentity) NodeID

func (r ResourceIdentity) NodeID() string

NodeID returns the node ID when present.

type ResourceManager

type ResourceManager struct {
	services.Service
	// contains filtered or unexported fields
}

ResourceManager emits MeterRecords and periodic MeterSnapshots for durable resources. It is a services.Service: callers start it (typically as a sub-service of the producer) and Register Meterables to be snapshotted. It is safe for concurrent use.

func NewResourceManager

func NewResourceManager(lggr logger.Logger, cfg ResourceManagerConfig) *ResourceManager

NewResourceManager returns a ResourceManager. A failure to create a metric instrument is logged and that instrument is skipped; it never prevents construction. The manager must be Started before its snapshot loop runs; EmitMeterRecord works regardless of Start.

func (*ResourceManager) EmitMeterRecord

func (rm *ResourceManager) EmitMeterRecord(ctx context.Context, identity ResourceIdentity, action meteringpb.MeterAction, utilizations []*meteringpb.Utilization)

EmitMeterRecord emits a metering.v1.MeterRecord, timestamped now, for action on the one resource described by identity.

EmitMeterRecord is fail-open and returns no error: when the manager is disabled or has no emitter it does nothing, and marshal or emit failures are recorded only via error-level logs and the failure counter. Callers must never gate resource allocation or deallocation on emission.

func (*ResourceManager) Register

func (rm *ResourceManager) Register(m Meterable) (unregister func())

Register adds m to the snapshot registry and returns an idempotent function that removes it. Calling the returned function more than once is a no-op. The returned closure is safe for concurrent use.

type ResourceManagerConfig

type ResourceManagerConfig struct {
	// MeterRecordsEnabled is the meter-record rollout gate. When false (the
	// default), EmitMeterRecord is a no-op.
	MeterRecordsEnabled bool

	// MeterSnapshotsEnabled gates snapshot emission. Snapshots are only emitted
	// when this is true AND MeterRecordsEnabled is true.
	MeterSnapshotsEnabled bool

	// Emitter delivers encoded records, typically beholder.GetEmitter(). A nil
	// Emitter makes EmitMeterRecord a no-op even when Enabled is true and keeps
	// the snapshot loop from starting.
	Emitter Emitter

	// SnapshotInterval is the period between snapshots. Zero (the default)
	// DISABLES the snapshot loop; the manager still starts as a service and
	// EmitMeterRecord still works. Callers that want snapshots set a positive
	// value, e.g. DefaultSnapshotInterval. The default is not substituted for
	// zero — zero means off.
	SnapshotInterval time.Duration

	// Clock drives snapshot tick timing and record timestamps. Nil selects the
	// real clock.
	Clock clockwork.Clock
}

ResourceManagerConfig configures a ResourceManager.

type SnapshotEntry

type SnapshotEntry struct {
	Identity     ResourceIdentity
	Utilizations []*meteringpb.Utilization
}

SnapshotEntry is the current level of one active resource at a snapshot tick. Identity is the base resource identity, and Utilizations carries one or more billed dimensions for that resource (resource_type/resource_id/org_id/event_id/value).

type UtilizationFields

type UtilizationFields struct {
	ResourceType string
	ResourceID   string
	EventID      string
	OrgID        string
}

UtilizationFields identifies one billed utilization dimension.