feat(worker): Custom Metrics (#1705)

Author: mjameswh

packages/activity/src/index.ts

@@ -70,7 +70,7 @@
  */
 
 import { AsyncLocalStorage } from 'node:async_hooks';
-import { Logger, Duration, LogLevel, LogMetadata, Priority } from '@temporalio/common';
+import { Logger, Duration, LogLevel, LogMetadata, MetricMeter, Priority } from '@temporalio/common';
 import { msToNumber } from '@temporalio/common/lib/time';
 import { SymbolBasedInstanceOfError } from '@temporalio/common/lib/type-helpers';
 
@@ -281,6 +281,14 @@ export class Context {
    */
   public log: Logger;
 
+  /**
+   * Get the metric meter for this activity with activity-specific tags.
+   *
+   * To add custom tags, register a {@link ActivityOutboundCallsInterceptor} that
+   * intercepts the `getMetricTags()` method.
+   */
+  public readonly metricMeter: MetricMeter;
+
   /**
    * **Not** meant to instantiated by Activity code, used by the worker.
    *
@@ -291,13 +299,15 @@ export class Context {
     cancelled: Promise<never>,
     cancellationSignal: AbortSignal,
     heartbeat: (details?: any) => void,
-    log: Logger
+    log: Logger,
+    metricMeter: MetricMeter
   ) {
     this.info = info;
     this.cancelled = cancelled;
     this.cancellationSignal = cancellationSignal;
     this.heartbeatFn = heartbeat;
     this.log = log;
+    this.metricMeter = metricMeter;
   }
 
   /**
@@ -434,3 +444,26 @@ export function cancelled(): Promise<never> {
 export function cancellationSignal(): AbortSignal {
   return Context.current().cancellationSignal;
 }
+
+/**
+ * Get the metric meter for the current activity, with activity-specific tags.
+ *
+ * To add custom tags, register a {@link ActivityOutboundCallsInterceptor} that
+ * intercepts the `getMetricTags()` method.
+ *
+ * This is a shortcut for `Context.current().metricMeter` (see {@link Context.metricMeter}).
+ */
+export const metricMeter: MetricMeter = {
+  createCounter(name, unit, description) {
+    return Context.current().metricMeter.createCounter(name, unit, description);
+  },
+  createHistogram(name, valueType = 'int', unit, description) {
+    return Context.current().metricMeter.createHistogram(name, valueType, unit, description);
+  },
+  createGauge(name, valueType = 'int', unit, description) {
+    return Context.current().metricMeter.createGauge(name, valueType, unit, description);
+  },
+  withTags(tags) {
+    return Context.current().metricMeter.withTags(tags);
+  },
+};

packages/client/src/async-completion-client.ts

@@ -1,10 +1,7 @@
 import { status as grpcStatus } from '@grpc/grpc-js';
 import { ensureTemporalFailure } from '@temporalio/common';
-import {
-  encodeErrorToFailure,
-  encodeToPayloads,
-  filterNullAndUndefined,
-} from '@temporalio/common/lib/internal-non-workflow';
+import { encodeErrorToFailure, encodeToPayloads } from '@temporalio/common/lib/internal-non-workflow';
+import { filterNullAndUndefined } from '@temporalio/common/lib/internal-workflow';
 import { SymbolBasedInstanceOfError } from '@temporalio/common/lib/type-helpers';
 import {
   BaseClient,

packages/client/src/client.ts

@@ -1,4 +1,4 @@
-import { filterNullAndUndefined } from '@temporalio/common/lib/internal-non-workflow';
+import { filterNullAndUndefined } from '@temporalio/common/lib/internal-workflow';
 import { AsyncCompletionClient } from './async-completion-client';
 import { BaseClient, BaseClientOptions, defaultBaseClientOptions, LoadedWithDefaults } from './base-client';
 import { ClientInterceptors } from './interceptors';

packages/client/src/connection.ts

@@ -2,11 +2,11 @@ import { AsyncLocalStorage } from 'node:async_hooks';
 import * as grpc from '@grpc/grpc-js';
 import type * as proto from 'protobufjs';
 import {
-  filterNullAndUndefined,
   normalizeTlsConfig,
   TLSConfig,
   normalizeGrpcEndpointAddress,
 } from '@temporalio/common/lib/internal-non-workflow';
+import { filterNullAndUndefined } from '@temporalio/common/lib/internal-workflow';
 import { Duration, msOptionalToNumber } from '@temporalio/common/lib/time';
 import { type temporal } from '@temporalio/proto';
 import { isGrpcServiceError, ServiceError } from './errors';

packages/client/src/schedule-client.ts

@@ -7,11 +7,8 @@ import {
   encodeUnifiedSearchAttributes,
 } from '@temporalio/common/lib/converter/payload-search-attributes';
 import { composeInterceptors, Headers } from '@temporalio/common/lib/interceptors';
-import {
-  encodeMapToPayloads,
-  decodeMapFromPayloads,
-  filterNullAndUndefined,
-} from '@temporalio/common/lib/internal-non-workflow';
+import { encodeMapToPayloads, decodeMapFromPayloads } from '@temporalio/common/lib/internal-non-workflow';
+import { filterNullAndUndefined } from '@temporalio/common/lib/internal-workflow';
 import { temporal } from '@temporalio/proto';
 import {
   optionalDateToTs,

packages/client/src/task-queue-client.ts

@@ -1,7 +1,6 @@
 import { status } from '@grpc/grpc-js';
-import { filterNullAndUndefined } from '@temporalio/common/lib/internal-non-workflow';
 import { assertNever, SymbolBasedInstanceOfError, RequireAtLeastOne } from '@temporalio/common/lib/type-helpers';
-import { makeProtoEnumConverters } from '@temporalio/common/lib/internal-workflow';
+import { filterNullAndUndefined, makeProtoEnumConverters } from '@temporalio/common/lib/internal-workflow';
 import { temporal } from '@temporalio/proto';
 import { BaseClient, BaseClientOptions, defaultBaseClientOptions, LoadedWithDefaults } from './base-client';
 import { WorkflowService } from './types';

packages/client/src/workflow-client.ts

@@ -34,8 +34,8 @@ import {
   decodeOptionalFailureToOptionalError,
   encodeMapToPayloads,
   encodeToPayloads,
-  filterNullAndUndefined,
 } from '@temporalio/common/lib/internal-non-workflow';
+import { filterNullAndUndefined } from '@temporalio/common/lib/internal-workflow';
 import { temporal } from '@temporalio/proto';
 import {
   ServiceError,

packages/cloud/src/cloud-operations-client.ts

@@ -2,11 +2,11 @@ import { AsyncLocalStorage } from 'node:async_hooks';
 import * as grpc from '@grpc/grpc-js';
 import type { RPCImpl } from 'protobufjs';
 import {
-  filterNullAndUndefined,
   normalizeTlsConfig,
   TLSConfig,
   normalizeGrpcEndpointAddress,
 } from '@temporalio/common/lib/internal-non-workflow';
+import { filterNullAndUndefined } from '@temporalio/common/lib/internal-workflow';
 import { Duration, msOptionalToNumber } from '@temporalio/common/lib/time';
 import {
   CallContext,

packages/common/src/index.ts

@@ -20,6 +20,7 @@ export { Headers, Next } from './interceptors';
 export * from './interfaces';
 export * from './logger';
 export * from './priority';
+export * from './metrics';
 export * from './retry-policy';
 export type { Timestamp, Duration, StringValue } from './time';
 export * from './worker-deployments';

packages/common/src/internal-non-workflow/index.ts

@@ -9,4 +9,3 @@ export * from './data-converter-helpers';
 export * from './parse-host-uri';
 export * from './proxy-config';
 export * from './tls-config';
-export * from './utils';