diff --git a/packages/core-bridge/sdk-core b/packages/core-bridge/sdk-core index 2872b5363e..5440336797 160000 --- a/packages/core-bridge/sdk-core +++ b/packages/core-bridge/sdk-core @@ -1 +1 @@ -Subproject commit 2872b5363e1b745cfb90313ebc7a507c5d25c398 +Subproject commit 5440336797f0557c3e0d176cc1c8c3ce9a274cbd diff --git a/packages/test/src/mock-native-worker.ts b/packages/test/src/mock-native-worker.ts index 9feaa947b2..62795598f8 100644 --- a/packages/test/src/mock-native-worker.ts +++ b/packages/test/src/mock-native-worker.ts @@ -12,6 +12,7 @@ import { MetricMeterWithComposedTags } from '@temporalio/common/lib/metrics'; import type { CompiledWorkerOptions, WorkerOptions } from '@temporalio/worker/lib/worker-options'; import { compileWorkerOptions } from '@temporalio/worker/lib/worker-options'; import type { WorkflowCreator } from '@temporalio/worker/lib/workflow/interface'; +import { WorkflowMetricsTracker } from '@temporalio/worker/lib/workflow/metrics-tracker'; import * as activities from './activities'; const sleep = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); @@ -182,7 +183,17 @@ export class Worker extends RealWorker { taskQueue: opts.taskQueue, }); const nativeWorker = new MockNativeWorker(); - super(runtime, nativeWorker, workflowCreator, opts, logger, runtime.metricMeter, opts.plugins ?? []); + const metricsTracker = new WorkflowMetricsTracker(runtime.metricMeter); + super( + runtime, + nativeWorker, + workflowCreator, + opts, + logger, + runtime.metricMeter, + metricsTracker, + opts.plugins ?? [] + ); } public runWorkflows(...args: Parameters): Promise { @@ -207,6 +218,7 @@ export function isolateFreeWorker(options: WorkerOptions = defaultOptions): Work namespace: options.namespace ?? 'default', taskQueue: options.taskQueue ?? 'default', }); + const metricsTracker = new WorkflowMetricsTracker(metricMeter); return new Worker( { async createWorkflow() { @@ -216,6 +228,6 @@ export function isolateFreeWorker(options: WorkerOptions = defaultOptions): Work /* Nothing to destroy */ }, }, - compileWorkerOptions(options, logger, metricMeter) + compileWorkerOptions(options, logger, metricsTracker) ); } diff --git a/packages/test/src/test-metrics-custom.ts b/packages/test/src/test-metrics-custom.ts index e628a5cc75..2152a9572e 100644 --- a/packages/test/src/test-metrics-custom.ts +++ b/packages/test/src/test-metrics-custom.ts @@ -89,6 +89,13 @@ test('Custom Metrics - Bridge supports works properly (no tags)', async (t) => { await assertMetricReported(t, /my_float_histogram_bucket{le="0.05"} 1/); await assertMetricReported(t, /my_float_histogram_bucket{le="0.1"} 2/); await assertMetricReported(t, /my_float_histogram_bucket{le="1"} 3/); + + // UpDownCounter + const upDown = meter.createUpDownCounter!('my-up-down-counter', 'my-up-down-counter-unit', 'my-up-down-counter-description'); + upDown.add(1); + upDown.add(5); + upDown.add(-3); // 1 + 5 - 3 = 3 + await assertMetricReported(t, /my_up_down_counter 3/); }); /** @@ -127,6 +134,11 @@ test('Custom Metrics - Tags composition works properly', async (t) => { t, /my_float_histogram_bucket{labelA="value-a",labelB="true",labelC="123",labelD="123.456",le="0.5"} 1/ ); + + // UpDownCounter + const upDown = meter.createUpDownCounter!('my-up-down-counter', 'my-up-down-counter-unit', 'my-up-down-counter-description'); + upDown.add(2, { labelA: 'value-a', labelB: true, labelC: 123, labelD: 123.456 }); + await assertMetricReported(t, /my_up_down_counter{labelA="value-a",labelB="true",labelC="123",labelD="123.456"} 2/); }); export async function metricWorksWorkflow(): Promise { @@ -161,12 +173,18 @@ export async function metricWorksWorkflow(): Promise { 'workflow-float-gauge-unit', 'workflow-float-gauge-description' ); + const myUpDownCounterMetric = metricMeter.createUpDownCounter!( + 'workflow-up-down-counter', + 'workflow-up-down-counter-unit', + 'workflow-up-down-counter-description' + ); myCounterMetric.add(1); myHistogramMetric.record(1); myFloatHistogramMetric.record(0.01); myGaugeMetric.set(1); myFloatGaugeMetric.set(0.1); + myUpDownCounterMetric.add(2); // Pause here, so that we can force replay to a distinct worker let signalReceived = false; @@ -181,6 +199,7 @@ export async function metricWorksWorkflow(): Promise { myFloatHistogramMetric.record(0.03); myGaugeMetric.set(3); myFloatGaugeMetric.set(0.3); + myUpDownCounterMetric.add(-1); } test('Metric in Workflow works and are not replayed', async (t) => { @@ -212,6 +231,7 @@ test('Metric in Workflow works and are not replayed', async (t) => { await assertMetricReported(t, /workflow_float_histogram_bucket{[^}]+?,le="0.05"} 2/); await assertMetricReported(t, /workflow_gauge{[^}]+} 1/); await assertMetricReported(t, /workflow_float_gauge{[^}]+} 0.1/); + await assertMetricReported(t, /workflow_up_down_counter{[^}]+} 4/); const worker2 = await createWorker(); await worker2.runUntil(async () => { @@ -224,6 +244,7 @@ test('Metric in Workflow works and are not replayed', async (t) => { await assertMetricReported(t, /workflow_float_histogram_bucket{[^}]+?,le="0.05"} 4/); await assertMetricReported(t, /workflow_gauge{[^}]+} 3/); await assertMetricReported(t, /workflow_float_gauge{[^}]+} 0.3/); + await assertMetricReported(t, /workflow_up_down_counter{[^}]+} 2/); }); export async function MetricTagsWorkflow(): Promise { diff --git a/packages/test/src/test-up-down-counter-replay.ts b/packages/test/src/test-up-down-counter-replay.ts new file mode 100644 index 0000000000..ba20197a0f --- /dev/null +++ b/packages/test/src/test-up-down-counter-replay.ts @@ -0,0 +1,220 @@ +import test from 'ava'; +import type { + MetricMeter, + MetricTags, + MetricUpDownCounter, + MetricCounter, + MetricHistogram, + MetricGauge, + NumericMetricValueType, +} from '@temporalio/common'; +import { WorkflowMetricsTracker } from '@temporalio/worker/lib/workflow/metrics-tracker'; + +interface RecordedAdd { + name: string; + value: number; + tags: MetricTags; +} + +interface RecordedInstrument { + name: string; + unit: string | undefined; + description: string | undefined; +} + +class FakeUpDownCounter implements MetricUpDownCounter { + public readonly kind = 'up-down-counter' as const; + public readonly valueType = 'int' as const; + constructor( + public readonly name: string, + public readonly unit: string | undefined, + public readonly description: string | undefined, + private readonly recorded: RecordedAdd[] + ) {} + add(value: number, tags: MetricTags = {}): void { + this.recorded.push({ name: this.name, value, tags }); + } + withTags(_tags: MetricTags): MetricUpDownCounter { + throw new Error('not used in this test'); + } +} + +class FakeMetricMeter implements MetricMeter { + public readonly recorded: RecordedAdd[] = []; + public readonly instruments: RecordedInstrument[] = []; + createCounter(_n: string, _u?: string, _d?: string): MetricCounter { + throw new Error('unused'); + } + createHistogram(_n: string, _v?: NumericMetricValueType, _u?: string, _d?: string): MetricHistogram { + throw new Error('unused'); + } + createGauge(_n: string, _v?: NumericMetricValueType, _u?: string, _d?: string): MetricGauge { + throw new Error('unused'); + } + createUpDownCounter(name: string, unit?: string, description?: string): MetricUpDownCounter { + this.instruments.push({ name, unit, description }); + return new FakeUpDownCounter(name, unit, description, this.recorded); + } + withTags(_t: MetricTags): MetricMeter { + throw new Error('unused'); + } +} + +function callUpDownCounterSink( + tracker: WorkflowMetricsTracker, + runId: string, + name: string, + netValue: number, + attrs: MetricTags = {} +) { + const sinks = tracker.getInjectedSinks(); + const fn = sinks.__temporal_metrics.addMetricUpDownCounterValue.fn; + void fn({ runId } as any, name, undefined, undefined, netValue, attrs); +} + +test('first emission applies full delta', (t) => { + const meter = new FakeMetricMeter(); + const tracker = new WorkflowMetricsTracker(meter); + callUpDownCounterSink(tracker, 'run-1', 'inflight', 5); + t.deepEqual(meter.recorded, [{ name: 'inflight', value: 5, tags: {} }]); +}); + +test('idempotent re-emission with same net value applies no delta', (t) => { + const meter = new FakeMetricMeter(); + const tracker = new WorkflowMetricsTracker(meter); + callUpDownCounterSink(tracker, 'run-1', 'inflight', 5); + callUpDownCounterSink(tracker, 'run-1', 'inflight', 5); + t.deepEqual(meter.recorded, [{ name: 'inflight', value: 5, tags: {} }]); +}); + +test('different tag combinations are tracked independently', (t) => { + const meter = new FakeMetricMeter(); + const tracker = new WorkflowMetricsTracker(meter); + callUpDownCounterSink(tracker, 'run-1', 'inflight', 1, { region: 'us' }); + callUpDownCounterSink(tracker, 'run-1', 'inflight', 1, { region: 'eu' }); + callUpDownCounterSink(tracker, 'run-1', 'inflight', 3, { region: 'us' }); + t.deepEqual(meter.recorded, [ + { name: 'inflight', value: 1, tags: { region: 'us' } }, + { name: 'inflight', value: 1, tags: { region: 'eu' } }, + { name: 'inflight', value: 2, tags: { region: 'us' } }, + ]); +}); + +test('tag keys do not collide when tag values contain separators', (t) => { + const meter = new FakeMetricMeter(); + const tracker = new WorkflowMetricsTracker(meter); + callUpDownCounterSink(tracker, 'run-1', 'inflight', 1, { a: 'b,c=d' }); + callUpDownCounterSink(tracker, 'run-1', 'inflight', 1, { a: 'b', c: 'd' }); + t.deepEqual(meter.recorded, [ + { name: 'inflight', value: 1, tags: { a: 'b,c=d' } }, + { name: 'inflight', value: 1, tags: { a: 'b', c: 'd' } }, + ]); +}); + +test('same metric name with different descriptors creates independent instruments', (t) => { + const meter = new FakeMetricMeter(); + const tracker = new WorkflowMetricsTracker(meter); + const sinks = tracker.getInjectedSinks(); + const fn = sinks.__temporal_metrics.addMetricUpDownCounterValue.fn; + + void fn({ runId: 'run-1' } as any, 'inflight', 'items', 'Item count', 1, {}); + void fn({ runId: 'run-1' } as any, 'inflight', 'bytes', 'Byte count', 2, {}); + + t.deepEqual(meter.instruments, [ + { name: 'inflight', unit: 'items', description: 'Item count' }, + { name: 'inflight', unit: 'bytes', description: 'Byte count' }, + ]); + t.deepEqual(meter.recorded, [ + { name: 'inflight', value: 1, tags: {} }, + { name: 'inflight', value: 2, tags: {} }, + ]); +}); + +test('negative net values produce negative deltas', (t) => { + const meter = new FakeMetricMeter(); + const tracker = new WorkflowMetricsTracker(meter); + callUpDownCounterSink(tracker, 'run-1', 'inflight', 1); + callUpDownCounterSink(tracker, 'run-1', 'inflight', -2); + t.deepEqual(meter.recorded, [ + { name: 'inflight', value: 1, tags: {} }, + { name: 'inflight', value: -3, tags: {} }, + ]); +}); + +test('notifyWorkflowEvicted undoes contributions for that runId only', (t) => { + const meter = new FakeMetricMeter(); + const tracker = new WorkflowMetricsTracker(meter); + callUpDownCounterSink(tracker, 'run-A', 'inflight', 4); + callUpDownCounterSink(tracker, 'run-B', 'inflight', 7); + tracker.notifyWorkflowEvicted('run-A'); + t.deepEqual(meter.recorded, [ + { name: 'inflight', value: 4, tags: {} }, + { name: 'inflight', value: 7, tags: {} }, + { name: 'inflight', value: -4, tags: {} }, + ]); +}); + +test('notifyWorkflowEvicted with no contributions is a no-op', (t) => { + const meter = new FakeMetricMeter(); + const tracker = new WorkflowMetricsTracker(meter); + callUpDownCounterSink(tracker, 'run-A', 'inflight', 0); + tracker.notifyWorkflowEvicted('run-A'); + // The first emission (netValue=0) results in delta=0 → no native add. + // Eviction has nothing to undo. + t.deepEqual(meter.recorded, []); +}); + +test('full lifecycle: emit, evict, re-emit reaches steady state', (t) => { + const meter = new FakeMetricMeter(); + const tracker = new WorkflowMetricsTracker(meter); + callUpDownCounterSink(tracker, 'run-A', 'inflight', 1); + tracker.notifyWorkflowEvicted('run-A'); + callUpDownCounterSink(tracker, 'run-A', 'inflight', 1); + // Sum of recorded values: 1 + (-1) + 1 = 1 + const sum = meter.recorded.reduce((acc, r) => acc + r.value, 0); + t.is(sum, 1); +}); + +test('Counter sink applies values to the underlying meter', (t) => { + const counterAdds: RecordedAdd[] = []; + + class FakeCounter { + public readonly kind = 'counter' as const; + public readonly valueType = 'int' as const; + constructor( + public readonly name: string, + public readonly unit?: string, + public readonly description?: string + ) {} + add(value: number, tags: MetricTags = {}) { + counterAdds.push({ name: this.name, value, tags }); + } + withTags(): any { + throw new Error('unused'); + } + } + + const meter: MetricMeter = { + createCounter: (name, unit, description) => new FakeCounter(name, unit, description), + createHistogram: () => { + throw new Error('unused'); + }, + createGauge: () => { + throw new Error('unused'); + }, + createUpDownCounter: () => { + throw new Error('unused'); + }, + withTags: () => { + throw new Error('unused'); + }, + }; + + const tracker = new WorkflowMetricsTracker(meter); + const sinks = tracker.getInjectedSinks(); + sinks.__temporal_metrics.addMetricCounterValue.fn({ runId: 'r' } as any, 'requests', undefined, undefined, 3, { + region: 'us', + }); + t.deepEqual(counterAdds, [{ name: 'requests', value: 3, tags: { region: 'us' } }]); + t.is(sinks.__temporal_metrics.addMetricCounterValue.callDuringReplay, false); +}); diff --git a/packages/worker/src/worker-options.ts b/packages/worker/src/worker-options.ts index 60c4393531..1916289645 100644 --- a/packages/worker/src/worker-options.ts +++ b/packages/worker/src/worker-options.ts @@ -6,7 +6,6 @@ import type { ActivityFunction, DataConverter, LoadedDataConverter, - MetricMeter, VersioningBehavior, WorkerDeploymentVersion, } from '@temporalio/common'; @@ -22,7 +21,7 @@ import type { NativeConnection } from './connection'; import type { CompiledWorkerInterceptors, WorkerInterceptors } from './interceptors'; import type { Logger } from './logger'; import { initLoggerSink } from './workflow/logger'; -import { initMetricSink } from './workflow/metrics'; +import type { WorkflowMetricsTracker } from './workflow/metrics-tracker'; import { Runtime } from './runtime'; import type { InjectedSinks } from './sinks'; import { MiB } from './utils'; @@ -920,7 +919,7 @@ export type CompiledWorkerOptionsWithBuildId = CompiledWorkerOptions & { function addDefaultWorkerOptions( options: WorkerOptions, logger: Logger, - metricMeter: MetricMeter + metricsTracker: WorkflowMetricsTracker ): WorkerOptionsWithDefaults { const { buildId, @@ -1045,7 +1044,7 @@ function addDefaultWorkerOptions( nonStickyToStickyPollRatio: nonStickyToStickyPollRatio ?? 0.2, sinks: { ...initLoggerSink(logger), - ...initMetricSink(metricMeter), + ...metricsTracker.getInjectedSinks(), // Fix deprecated registration of the 'defaultWorkerLogger' sink ...(sinks?.defaultWorkerLogger ? { __temporal_logger: sinks.defaultWorkerLogger } : {}), ...sinks, @@ -1059,7 +1058,7 @@ function addDefaultWorkerOptions( export function compileWorkerOptions( rawOpts: WorkerOptions, logger: Logger, - metricMeter: MetricMeter + metricsTracker: WorkflowMetricsTracker ): CompiledWorkerOptions { // Validate sink names to ensure they don't use reserved prefixes/names if (rawOpts.sinks) { @@ -1068,7 +1067,7 @@ export function compileWorkerOptions( } } - const opts = addDefaultWorkerOptions(rawOpts, logger, metricMeter); + const opts = addDefaultWorkerOptions(rawOpts, logger, metricsTracker); if (opts.maxCachedWorkflows !== 0 && opts.maxCachedWorkflows < 2) { logger.warn('maxCachedWorkflows must be either 0 (ie. cache is disabled) or greater than 1. Defaulting to 2.'); opts.maxCachedWorkflows = 2; diff --git a/packages/worker/src/worker.ts b/packages/worker/src/worker.ts index 493ad74d35..cb315f8b39 100644 --- a/packages/worker/src/worker.ts +++ b/packages/worker/src/worker.ts @@ -79,6 +79,7 @@ import { compileWorkerOptions, isCodeBundleOption, isPathBundleOption, toNativeW import { WorkflowCodecRunner } from './workflow-codec-runner'; import { defaultWorkflowInterceptorModules, WorkflowCodeBundler } from './workflow/bundler'; import type { Workflow, WorkflowCreator } from './workflow/interface'; +import { WorkflowMetricsTracker } from './workflow/metrics-tracker'; import { ReusableVMWorkflowCreator } from './workflow/reusable-vm'; import { ThreadedVMWorkflowCreator } from './workflow/threaded-vm'; import { VMWorkflowCreator } from './workflow/vm'; @@ -504,8 +505,9 @@ export class Worker { namespace: options.namespace ?? 'default', taskQueue: options.taskQueue ?? 'default', }); + const metricsTracker = new WorkflowMetricsTracker(metricMeter); const nativeWorkerCtor: NativeWorkerConstructor = this.nativeWorkerCtor; - const compiledOptions = compileWorkerOptions(options, logger, metricMeter); + const compiledOptions = compileWorkerOptions(options, logger, metricsTracker); logger.debug('Creating worker', { options: { ...compiledOptions, @@ -546,6 +548,7 @@ export class Worker { compiledOptionsWithBuildId, logger, metricMeter, + metricsTracker, options.plugins ?? [], connection ); @@ -715,7 +718,8 @@ export class Worker { namespace: 'default', taskQueue: fixedUpOptions.taskQueue, }); - const compiledOptions = compileWorkerOptions(fixedUpOptions, logger, metricMeter); + const metricsTracker = new WorkflowMetricsTracker(metricMeter); + const compiledOptions = compileWorkerOptions(fixedUpOptions, logger, metricsTracker); const bundle = await this.getOrCreateBundle(compiledOptions, logger); if (!bundle) { throw new TypeError('ReplayWorkerOptions must contain workflowsPath or workflowBundle'); @@ -733,6 +737,7 @@ export class Worker { compiledOptions, logger, metricMeter, + metricsTracker, plugins, undefined, true @@ -806,10 +811,15 @@ export class Worker { /** Logger bound to 'sdkComponent: worker' */ protected readonly logger: Logger, protected readonly metricMeter: MetricMeter, + protected readonly metricsTracker: WorkflowMetricsTracker, protected readonly plugins: WorkerPlugin[], protected _connection?: NativeConnection, protected readonly isReplayWorker: boolean = false - ) {} + ) { + this.evictionsEmitter.on('eviction', (eviction: EvictionWithRunID) => { + this.metricsTracker.notifyWorkflowEvicted(eviction.runId); + }); + } /** * An Observable which emits each time the number of in flight activations changes diff --git a/packages/worker/src/workflow/metrics-tracker.ts b/packages/worker/src/workflow/metrics-tracker.ts new file mode 100644 index 0000000000..473f98b601 --- /dev/null +++ b/packages/worker/src/workflow/metrics-tracker.ts @@ -0,0 +1,128 @@ +import { type Metric, type MetricMeter, type MetricTags, type MetricUpDownCounter } from '@temporalio/common'; +import type { MetricSinks } from '@temporalio/workflow/lib/metrics'; +import type { InjectedSinks } from '../sinks'; + +interface TrackedContribution { + runId: string; + metricName: string; + unit: string | undefined; + description: string | undefined; + tags: MetricTags; + netValue: number; +} + +export function stableTagsKey(tags: MetricTags): string { + const keys = Object.keys(tags).sort(); + if (keys.length === 0) return ''; + return JSON.stringify(keys.map((k) => [k, tags[k]])); +} + +function metricDescriptorKey(name: string, unit: string | undefined, description: string | undefined): string { + return JSON.stringify([name, unit ?? null, description ?? null]); +} + +export class WorkflowMetricsTracker { + private readonly perWorkflowUpDownCounters = new Map(); + private readonly upDownCounterCache = new Map(); + + constructor(private readonly metricMeter: MetricMeter) {} + + getInjectedSinks(): InjectedSinks { + // Per-instrument cache so we don't recreate instruments on every emit. Uses + // WeakRef so unused instruments can be garbage-collected. + const cache = new Map>(); + const getOrCreate = (key: string, create: () => T): T => { + let value = cache.get(key)?.deref(); + if (value === undefined) { + value = create(); + cache.set(key, new WeakRef(value)); + } + return value as T; + }; + + return { + __temporal_metrics: { + addMetricCounterValue: { + fn: (_, metricName, unit, description, value, attrs) => { + const key = `${metricName}:counter`; + getOrCreate(key, () => this.metricMeter.createCounter(metricName, unit, description)).add(value, attrs); + }, + callDuringReplay: false, + }, + recordMetricHistogramValue: { + fn: (_, metricName, valueType, unit, description, value, attrs) => { + const key = `histogram:${valueType}:${metricName}`; + getOrCreate(key, () => this.metricMeter.createHistogram(metricName, valueType, unit, description)).record( + value, + attrs + ); + }, + callDuringReplay: false, + }, + setMetricGaugeValue: { + fn: (_, metricName, valueType, unit, description, value, attrs) => { + const key = `gauge:${valueType}:${metricName}`; + getOrCreate(key, () => this.metricMeter.createGauge(metricName, valueType, unit, description)).set( + value, + attrs + ); + }, + callDuringReplay: false, + }, + addMetricUpDownCounterValue: { + fn: (workflowInfo, metricName, unit, description, netValue, attrs) => { + const key = JSON.stringify([ + workflowInfo.runId, + metricName, + unit ?? null, + description ?? null, + stableTagsKey(attrs), + ]); + const existing = this.perWorkflowUpDownCounters.get(key); + const oldNet = existing?.netValue ?? 0; + const delta = netValue - oldNet; + this.perWorkflowUpDownCounters.set(key, { + runId: workflowInfo.runId, + metricName, + unit, + description, + tags: attrs, + netValue, + }); + if (delta !== 0) { + this.getUpDownCounter(metricName, unit, description).add(delta, attrs); + } + }, + callDuringReplay: true, + }, + }, + }; + } + + notifyWorkflowEvicted(runId: string): void { + for (const [key, contribution] of this.perWorkflowUpDownCounters) { + if (contribution.runId !== runId) continue; + if (contribution.netValue !== 0) { + this.getUpDownCounter(contribution.metricName, contribution.unit, contribution.description).add( + -contribution.netValue, + contribution.tags + ); + } + this.perWorkflowUpDownCounters.delete(key); + } + } + + private getUpDownCounter( + name: string, + unit: string | undefined, + description: string | undefined + ): MetricUpDownCounter { + const key = metricDescriptorKey(name, unit, description); + let counter = this.upDownCounterCache.get(key); + if (counter === undefined) { + counter = this.metricMeter.createUpDownCounter!(name, unit, description); + this.upDownCounterCache.set(key, counter); + } + return counter; + } +} diff --git a/packages/worker/src/workflow/metrics.ts b/packages/worker/src/workflow/metrics.ts deleted file mode 100644 index 849753be07..0000000000 --- a/packages/worker/src/workflow/metrics.ts +++ /dev/null @@ -1,73 +0,0 @@ -import type { NumericMetricValueType, Metric } from '@temporalio/common'; -import { type MetricMeter, type MetricTags } from '@temporalio/common'; -import type { MetricSinks } from '@temporalio/workflow/lib/metrics'; -import type { InjectedSinks } from '../sinks'; - -export function initMetricSink(metricMeter: MetricMeter): InjectedSinks { - // Creation of a new metric object isn't quite cheap, requiring a call down the bridge to the - // actual Metric Meter. Unfortunately, the workflow sandbox execution model doesn't allow to - // reuse metric objects from the caller side. We therefore maintain local caches of metric - // objects to avoid creating a new one for every single metric value being emitted. - const cache = new Map>(); - - function getOrCreate(key: string, create: () => T): T { - let value = cache.get(key)?.deref(); - if (value === undefined) { - value = create(); - cache.set(key, new WeakRef(value)); - } - return value as T; - } - - return { - __temporal_metrics: { - addMetricCounterValue: { - fn( - _, - metricName: string, - unit: string | undefined, - description: string | undefined, - value: number, - attrs: MetricTags - ) { - const cacheKey = `${metricName}:counter`; - const createFn = () => metricMeter.createCounter(metricName, unit, description); - getOrCreate(cacheKey, createFn).add(value, attrs); - }, - callDuringReplay: false, - }, - recordMetricHistogramValue: { - fn( - _, - metricName: string, - valueType: NumericMetricValueType, - unit: string | undefined, - description: string | undefined, - value: number, - attrs: MetricTags - ) { - const cacheKey = `histogram:${valueType}:${metricName}`; - const createFn = () => metricMeter.createHistogram(metricName, valueType, unit, description); - getOrCreate(cacheKey, createFn).record(value, attrs); - }, - callDuringReplay: false, - }, - setMetricGaugeValue: { - fn( - _, - metricName: string, - valueType: NumericMetricValueType, - unit: string | undefined, - description: string | undefined, - value: number, - attrs: MetricTags - ) { - const cacheKey = `gauge:${valueType}:${metricName}`; - const createFn = () => metricMeter.createGauge(metricName, valueType, unit, description); - getOrCreate(cacheKey, createFn).set(value, attrs); - }, - callDuringReplay: false, - }, - }, - }; -} diff --git a/packages/workflow/src/metrics.ts b/packages/workflow/src/metrics.ts index 7a4fbb5919..af63c4dfdf 100644 --- a/packages/workflow/src/metrics.ts +++ b/packages/workflow/src/metrics.ts @@ -4,6 +4,7 @@ import type { MetricHistogram, MetricMeter, MetricTags, + MetricUpDownCounter, NumericMetricValueType, } from '@temporalio/common'; import { MetricMeterWithComposedTags } from '@temporalio/common'; @@ -12,6 +13,43 @@ import type { Sink, Sinks } from './sinks'; import { proxySinks } from './sinks'; import { workflowInfo } from './workflow'; import { assertInWorkflowContext } from './global-attributes'; +import type { Activator } from './internals'; + +function stableTagsKey(tags: MetricTags): string { + const keys = Object.keys(tags).sort(); + if (keys.length === 0) return ''; + return JSON.stringify(keys.map((k) => [k, tags[k]])); +} + +function metricDescriptorKey(name: string, unit: string | undefined, description: string | undefined): string { + return JSON.stringify([name, unit ?? null, description ?? null]); +} + +// Per-workflow cache keyed by Activator. Each workflow execution gets its +// own singleton instances per metric descriptor. Required so that `add` calls +// against `metricMeter.createUpDownCounter('foo')` from multiple call +// sites accumulate against one canonical net-value map. +const upDownCounterCaches = new WeakMap>(); + +function getOrCreateUpDownCounter( + activator: Activator, + name: string, + unit: string | undefined, + description: string | undefined +): WorkflowMetricUpDownCounter { + let cache = upDownCounterCaches.get(activator); + if (cache === undefined) { + cache = new Map(); + upDownCounterCaches.set(activator, cache); + } + const key = metricDescriptorKey(name, unit, description); + let counter = cache.get(key); + if (counter === undefined) { + counter = new WorkflowMetricUpDownCounter(name, unit, description); + cache.set(key, counter); + } + return counter; +} class WorkflowMetricMeterImpl implements MetricMeter { constructor() {} @@ -41,6 +79,11 @@ class WorkflowMetricMeterImpl implements MetricMeter { return new WorkflowMetricGauge(name, valueType, unit, description); } + createUpDownCounter(name: string, unit?: string, description?: string): MetricUpDownCounter { + const activator = assertInWorkflowContext("Workflow's `metricMeter` can only be used while in Workflow Context"); + return getOrCreateUpDownCounter(activator, name, unit, description); + } + withTags(_tags: MetricTags): MetricMeter { assertInWorkflowContext("Workflow's `metricMeter` can only be used while in Workflow Context"); // Tags composition is handled by a MetricMeterWithComposedTags wrapper over this one @@ -123,6 +166,37 @@ class WorkflowMetricGauge implements MetricGauge { } } +class WorkflowMetricUpDownCounter implements MetricUpDownCounter { + public readonly kind = 'up-down-counter'; + public readonly valueType = 'int'; + + // Cumulative net value per stable tag-key. Replay rebuilds this map by + // re-executing the workflow, so on a fresh sandbox the values match what + // they were on the previous worker — emitting the same absolute net. + private readonly netValues = new Map(); + + constructor( + public readonly name: string, + public readonly unit: string | undefined, + public readonly description: string | undefined + ) {} + + add(value: number, tags?: MetricTags): void { + const resolvedTags = tags ?? {}; + const key = stableTagsKey(resolvedTags); + const newNet = (this.netValues.get(key) ?? 0) + value; + this.netValues.set(key, newNet); + // Always emit — including during replay. The worker sink applies only + // the delta from the previously tracked net, so idempotent replays + // produce delta=0 and no native call. + metricSink.addMetricUpDownCounterValue(this.name, this.unit, this.description, newNet, resolvedTags); + } + + withTags(_tags: MetricTags): MetricUpDownCounter { + throw new Error('withTags is not supported directly on WorkflowMetricUpDownCounter'); + } +} + //////////////////////////////////////////////////////////////////////////////////////////////////// // Note: given that forwarding metrics outside of the sanbox can be quite chatty and add non @@ -176,6 +250,14 @@ export interface WorkflowMetricMeter extends Sink { value: number, attrs: MetricTags ): void; + + addMetricUpDownCounterValue( + metricName: string, + unit: string | undefined, + description: string | undefined, + netValue: number, + attrs: MetricTags + ): void; } /**