Add workflow interceptor support

Author: mjameswh

packages/test/src/test-metrics-custom.ts

@@ -1,6 +1,8 @@
 import { ExecutionContext } from 'ava';
-import { Runtime } from '@temporalio/worker';
+import { ActivityInboundCallsInterceptor, ActivityOutboundCallsInterceptor, Runtime } from '@temporalio/worker';
 import * as workflow from '@temporalio/workflow';
+import { MetricTags } from '@temporalio/common';
+import { Context as ActivityContext, metricMeter as activityMetricMeter } from '@temporalio/activity';
 import { Context as BaseContext, helpers, makeTestFunction } from './helpers-integration';
 import { getRandomPort } from './helpers';
 
@@ -261,7 +263,7 @@ export async function MetricTagsWorkflow(): Promise<void> {
 
 test('Metric tags in Workflow works', async (t) => {
   const { createWorker, executeWorkflow, taskQueue } = helpers(t);
-  const tags = `labelA="value-a",labelB="value-b2",labelC="value-c",labelD="value-d",labelX="value-x",labelY="value-y",namespace="default",taskQueue="${taskQueue}"`;
+  const tags = `labelA="value-a",labelB="value-b2",labelC="value-c",labelD="value-d",labelX="value-x",labelY="value-y",namespace="default",taskQueue="${taskQueue}",workflowType="MetricTagsWorkflow"`;
 
   const worker = await createWorker();
   await worker.runUntil(executeWorkflow(MetricTagsWorkflow));
@@ -270,3 +272,103 @@ test('Metric tags in Workflow works', async (t) => {
   await assertMetricReported(t, new RegExp(`workflow2_histogram_bucket{${tags},le="50"} 1`));
   await assertMetricReported(t, new RegExp(`workflow2_gauge{${tags}} 2`));
 });
+
+// Define workflow interceptor for metrics
+export const interceptors = (): workflow.WorkflowInterceptors => ({
+  outbound: [
+    {
+      getMetricTags(tags: MetricTags): MetricTags {
+        if (!workflow.workflowInfo().workflowType.includes('Interceptor')) return tags;
+        return {
+          ...tags,
+          intercepted: 'workflow-interceptor',
+        };
+      },
+    },
+  ],
+});
+
+// Define activity interceptor for metrics
+export function activityInterceptorFactory(_ctx: ActivityContext): {
+  inbound?: ActivityInboundCallsInterceptor;
+  outbound?: ActivityOutboundCallsInterceptor;
+} {
+  return {
+    outbound: {
+      getMetricTags(tags: MetricTags): MetricTags {
+        return {
+          ...tags,
+          intercepted: 'activity-interceptor',
+        };
+      },
+    },
+  };
+}
+
+// Activity that uses metrics
+export async function metricActivity(): Promise<void> {
+  const { metricMeter } = ActivityContext.current();
+
+  const counter = metricMeter.createCounter('activity-counter');
+  counter.add(5);
+
+  const histogram = metricMeter.createHistogram('activity-histogram');
+  histogram.record(10);
+
+  // Use the `metricMeter` exported from the top level of the activity module rather than the one in the context
+  const gauge = activityMetricMeter.createGauge('activity-gauge');
+  gauge.set(15);
+}
+
+// Workflow that uses metrics and calls the activity
+export async function metricsInterceptorWorkflow(): Promise<void> {
+  const metricMeter = workflow.metricMeter;
+
+  // Use workflow metrics
+  const counter = metricMeter.createCounter('intercepted-workflow-counter');
+  counter.add(3);
+
+  const histogram = metricMeter.createHistogram('intercepted-workflow-histogram');
+  histogram.record(6);
+
+  const gauge = metricMeter.createGauge('intercepted-workflow-gauge');
+  gauge.set(9);
+
+  // Call activity with metrics
+  await workflow
+    .proxyActivities({
+      startToCloseTimeout: '1 minute',
+    })
+    .metricActivity();
+}
+
+// Test for workflow metrics interceptor
+test('Workflow and Activity Context metrics interceptors add tags', async (t) => {
+  const { createWorker, executeWorkflow, taskQueue } = helpers(t);
+
+  const worker = await createWorker({
+    taskQueue,
+    workflowsPath: __filename,
+    activities: {
+      metricActivity,
+    },
+    interceptors: {
+      activity: [activityInterceptorFactory],
+    },
+  });
+
+  await worker.runUntil(executeWorkflow(metricsInterceptorWorkflow));
+
+  // Verify workflow metrics have interceptor tag
+  await assertMetricReported(t, /intercepted_workflow_counter{[^}]*intercepted="workflow-interceptor"[^}]*} 3/);
+  await assertMetricReported(
+    t,
+    /intercepted_workflow_histogram_bucket{[^}]*intercepted="workflow-interceptor"[^}]*} \d+/
+  );
+  await assertMetricReported(t, /intercepted_workflow_gauge{[^}]*intercepted="workflow-interceptor"[^}]*} 9/);
+
+  // Verify activity metrics have interceptor tag
+  await assertMetricReported(t, /activity_counter{[^}]*intercepted="activity-interceptor"[^}]*} 5/);
+  await assertMetricReported(t, /activity_histogram_bucket{[^}]*intercepted="activity-interceptor"[^}]*} \d+/);
+  await assertMetricReported(t, /activity_gauge{[^}]*intercepted="activity-interceptor"[^}]*} 15/);
+});

packages/test/src/test-sinks.ts

@@ -94,6 +94,7 @@ if (RUN_INTEGRATION_TESTS) {
     });
 
     // Capture volatile values that are hard to predict
+    // eslint-disable-next-line deprecation/deprecation
     const { historySize, startTime, runStartTime, currentBuildId } = recordedCalls[0].info;
     t.true(historySize > 300);
 

packages/workflow/src/metrics.ts

@@ -55,6 +55,9 @@ class WorkflowMetricCounter implements MetricCounter {
   ) {}
 
   add(value: number, extraTags: MetricTags = {}): void {
+    if (value < 0) {
+      throw new Error(`MetricCounter value must be non-negative (got ${value})`);
+    }
     if (!workflowInfo().unsafe.isReplaying) {
       metricSink.addMetricCounterValue(this.name, this.unit, this.description, value, extraTags);
     }
@@ -75,6 +78,9 @@ class WorkflowMetricHistogram implements MetricHistogram {
   ) {}
 
   record(value: number, extraTags: MetricTags = {}): void {
+    if (value < 0) {
+      throw new Error(`MetricHistogram value must be non-negative (got ${value})`);
+    }
     if (!workflowInfo().unsafe.isReplaying) {
       metricSink.recordMetricHistogramValue(this.name, this.valueType, this.unit, this.description, value, extraTags);
     }
@@ -95,6 +101,9 @@ class WorkflowMetricGauge implements MetricGauge {
   ) {}
 
   set(value: number, tags?: MetricTags): void {
+    if (value < 0) {
+      throw new Error(`MetricGauge value must be non-negative (got ${value})`);
+    }
     if (!workflowInfo().unsafe.isReplaying) {
       metricSink.setMetricGaugeValue(this.name, this.valueType, this.unit, this.description, value, tags ?? {});
     }
@@ -168,6 +177,16 @@ export interface WorkflowMetricMeter extends Sink {
  */
 export const metricMeter: MetricMeter = MetricMeterWithComposedTags.compose(
   new WorkflowMetricMeterImpl(),
-  () => ({}), // FIXME: Add base tags + call getTags interceptor
+  () => {
+    const activator = assertInWorkflowContext('Workflow.metricMeter may only be used from workflow context.');
+    const getMetricTags = composeInterceptors(activator.interceptors.outbound, 'getMetricTags', (a) => a);
+
+    const info = activator.info;
+    return getMetricTags({
+      namespace: info.namespace,
+      taskQueue: info.taskQueue,
+      workflowType: info.workflowType,
+    });
+  },
   true
 );