feat(activity): Add Activity Context logger (#1138)

mjameswh · web-flow · commit 38b81d37d84c · 2023-06-21T18:40:54.000-04:00
diff --git a/packages/activity/src/index.ts b/packages/activity/src/index.ts
@@ -71,7 +71,8 @@
 
 import 'abort-controller/polyfill'; // eslint-disable-line import/no-unassigned-import
 import { AsyncLocalStorage } from 'node:async_hooks';
-import { Duration, msToNumber } from '@temporalio/common/lib/time';
+import { Logger, Duration } from '@temporalio/common';
+import { msToNumber } from '@temporalio/common/lib/time';
 
 export {
   ActivityFunction,
@@ -243,6 +244,19 @@ export class Context {
    * The heartbeat implementation, injected via the constructor.
    */
   protected readonly heartbeatFn: (details?: any) => void;
+  /**
+   * The logger for this Activity.
+   *
+   * This defaults to the `Runtime`'s Logger (see {@link Runtime.logger}). If the {@link ActivityInboundLogInterceptor}
+   * is installed (by default, it is; see {@link WorkerOptions.interceptors}), then various attributes from the current
+   * Activity context will automatically be included as metadata on every log entries, and some key events of the
+   * Activity's life cycle will automatically be logged (at 'DEBUG' level for most messages; 'WARN' for failures).
+   *
+   * To use a different Logger, either overwrite this property from an Activity Interceptor, or explicitly register the
+   * `ActivityInboundLogInterceptor` with your custom Logger. You may also subclass `ActivityInboundLogInterceptor` to
+   * customize attributes that are emitted as metadata.
+   */
+  public logger: Logger;
 
   /**
    * **Not** meant to instantiated by Activity code, used by the worker.
@@ -253,12 +267,14 @@ export class Context {
     info: Info,
     cancelled: Promise<never>,
     cancellationSignal: AbortSignal,
-    heartbeat: (details?: any) => void
+    heartbeat: (details?: any) => void,
+    logger: Logger
   ) {
     this.info = info;
     this.cancelled = cancelled;
     this.cancellationSignal = cancellationSignal;
     this.heartbeatFn = heartbeat;
+    this.logger = logger;
   }
 
   /**
diff --git a/packages/common/src/index.ts b/packages/common/src/index.ts
@@ -18,6 +18,7 @@ export * from './errors';
 export * from './failure';
 export { Headers, Next } from './interceptors';
 export * from './interfaces';
+export * from './logger';
 export * from './retry-policy';
 export { type Timestamp, Duration, StringValue } from './time';
 export * from './workflow-handle';
diff --git a/packages/common/src/logger.ts b/packages/common/src/logger.ts
@@ -0,0 +1,15 @@
+export type LogLevel = 'TRACE' | 'DEBUG' | 'INFO' | 'WARN' | 'ERROR';
+
+export type LogMetadata = Record<string | symbol, any>;
+
+/**
+ * Implement this interface in order to customize worker logging
+ */
+export interface Logger {
+  log(level: LogLevel, message: string, meta?: LogMetadata): any;
+  trace(message: string, meta?: LogMetadata): any;
+  debug(message: string, meta?: LogMetadata): any;
+  info(message: string, meta?: LogMetadata): any;
+  warn(message: string, meta?: LogMetadata): any;
+  error(message: string, meta?: LogMetadata): any;
+}
diff --git a/packages/core-bridge/src/conversions.rs b/packages/core-bridge/src/conversions.rs
@@ -323,10 +323,10 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
 
         let max_worker_activities_per_second =
             js_optional_getter!(cx, self, "maxActivitiesPerSecond", JsNumber)
-                .map(|num| num.value(cx) as f64);
+                .map(|num| num.value(cx));
         let max_task_queue_activities_per_second =
             js_optional_getter!(cx, self, "maxTaskQueueActivitiesPerSecond", JsNumber)
-                .map(|num| num.value(cx) as f64);
+                .map(|num| num.value(cx));
 
         let graceful_shutdown_period =
             js_optional_getter!(cx, self, "shutdownGraceTimeMs", JsNumber)
diff --git a/packages/core-bridge/ts/index.ts b/packages/core-bridge/ts/index.ts
@@ -1,9 +1,12 @@
 import { SpanContext } from '@opentelemetry/api';
-import { Duration } from '@temporalio/common';
+import { LogLevel, Duration } from '@temporalio/common';
 import type { TLSConfig } from '@temporalio/common/lib/internal-non-workflow';
 
 export { TLSConfig };
 
+/** @deprecated Import from @temporalio/common instead */
+export { LogLevel };
+
 type Shadow<Base, New> = Base extends object
   ? New extends object
     ? {
@@ -334,9 +337,6 @@ export interface WorkerOptions {
   maxActivitiesPerSecond?: number;
 }
 
-/** Log level - must match rust log level names */
-export type LogLevel = 'TRACE' | 'DEBUG' | 'INFO' | 'WARN' | 'ERROR';
-
 export interface LogEntry {
   /** Log message */
   message: string;
diff --git a/packages/test/src/mock-native-worker.ts b/packages/test/src/mock-native-worker.ts
@@ -4,7 +4,7 @@ import { lastValueFrom } from 'rxjs';
 import { defaultPayloadConverter, fromPayloadsAtIndex } from '@temporalio/common';
 import { msToTs } from '@temporalio/common/lib/time';
 import { coresdk } from '@temporalio/proto';
-import { DefaultLogger, ShutdownError } from '@temporalio/worker';
+import { DefaultLogger, Runtime, ShutdownError } from '@temporalio/worker';
 import { byteArrayToBuffer } from '@temporalio/worker/lib/utils';
 import { NativeReplayHandle, NativeWorkerLike, Worker as RealWorker } from '@temporalio/worker/lib/worker';
 import {
@@ -164,6 +164,9 @@ export class Worker extends RealWorker {
   }
 
   public constructor(workflowCreator: WorkflowCreator, opts: CompiledWorkerOptions) {
+    // Worker.create() accesses Runtime.instance(), which has some side effects that would not happen (or that would
+    // happen too late) when creating a MockWorker. Force the singleton to be created now, if it doesn't already exist.
+    Runtime.instance();
     const nativeWorker = new MockNativeWorker();
     super(nativeWorker, workflowCreator, opts);
   }
diff --git a/packages/test/src/test-activity-log-interceptor.ts b/packages/test/src/test-activity-log-interceptor.ts
@@ -1,17 +1,69 @@
 import test from 'ava';
-import { ActivityInboundLogInterceptor, activityLogAttributes, DefaultLogger, LogEntry } from '@temporalio/worker';
+import {
+  ActivityInboundLogInterceptor,
+  activityLogAttributes,
+  DefaultLogger,
+  LogEntry,
+  Runtime,
+} from '@temporalio/worker';
 import { MockActivityEnvironment, defaultActivityInfo } from '@temporalio/testing';
 import { composeInterceptors } from '@temporalio/common/lib/interceptors';
 import * as activity from '@temporalio/activity';
 import { withZeroesHTTPServer } from './zeroes-http-server';
 import { cancellableFetch } from './activities/cancellable-fetch';
 
+interface MyTestActivityContext extends activity.Context {
+  logs: Array<LogEntry>;
+}
+
+test.before(() => {
+  const mockLogger = new DefaultLogger('DEBUG', (entry) => {
+    (activity.Context.current() as MyTestActivityContext).logs ??= [];
+    (activity.Context.current() as MyTestActivityContext).logs.push(entry);
+  });
+  Runtime.install({
+    logger: mockLogger,
+  });
+});
+
+test("Activity Context's logger defaults to Runtime's Logger", async (t) => {
+  const env = new MockActivityEnvironment();
+  const logs: LogEntry[] = await env.run(async () => {
+    const ctx = activity.Context.current();
+    ctx.logger.debug('log message from activity');
+    return (ctx as MyTestActivityContext).logs;
+  });
+  const activityLogEntry = logs.find((entry) => entry.message === 'log message from activity');
+  t.not(activityLogEntry, undefined);
+  t.is(activityLogEntry?.level, 'DEBUG');
+});
+
+test("Activity Log Interceptor dont override Context's logger by default", async (t) => {
+  const env = new MockActivityEnvironment();
+  const logs: LogEntry[] = await env.run(async () => {
+    const ctx = activity.Context.current();
+    const interceptor = new ActivityInboundLogInterceptor(ctx);
+    const execute = composeInterceptors([interceptor], 'execute', async () => {
+      ctx.logger.debug('log message from activity');
+    });
+    try {
+      await execute({ args: [], headers: {} });
+    } catch {
+      // ignore
+    }
+    return (ctx as MyTestActivityContext).logs;
+  });
+  const activityLogEntry = logs.find((entry) => entry.message === 'log message from activity');
+  t.not(activityLogEntry, undefined);
+  t.is(activityLogEntry?.level, 'DEBUG');
+});
+
 async function runActivity(
   fn: activity.ActivityFunction,
   heartbeatCallback = (_env: MockActivityEnvironment) => {
     // not an empty body eslint
   }
-): Promise<[LogEntry, LogEntry]> {
+): Promise<LogEntry[]> {
   const logs = Array<LogEntry>();
   const env = new MockActivityEnvironment();
   env.on('heartbeat', () => heartbeatCallback(env));
@@ -30,26 +82,38 @@ async function runActivity(
       // ignore
     }
   });
-  if (logs.length !== 2) {
-    throw new Error('Expected exactly 2 log entries');
-  }
-  return logs as [LogEntry, LogEntry];
+  return logs;
 }
 
+test('ActivityInboundLogInterceptor overrides Context logger if specified', async (t) => {
+  const logs = await runActivity(async () => {
+    activity.Context.current().logger.debug('log message from activity');
+  });
+  t.is(logs.length, 3);
+  const [_, midLog] = logs;
+  t.is(midLog.level, 'DEBUG');
+  t.is(midLog.message, 'log message from activity');
+  t.deepEqual(midLog.meta, activityLogAttributes(defaultActivityInfo));
+});
+
 test('ActivityInboundLogInterceptor logs when activity starts', async (t) => {
-  const [startLog] = await runActivity(async () => {
+  const logs = await runActivity(async () => {
     // not an empty body eslint
   });
+  t.is(logs.length, 2);
+  const [startLog] = logs;
   t.is(startLog.level, 'DEBUG');
   t.is(startLog.message, 'Activity started');
   t.deepEqual(startLog.meta, activityLogAttributes(defaultActivityInfo));
 });
 
 test('ActivityInboundLogInterceptor logs warning when activity fails', async (t) => {
   const err = new Error('Failed for test');
-  const [_, endLog] = await runActivity(async () => {
+  const logs = await runActivity(async () => {
     throw err;
   });
+  t.is(logs.length, 2);
+  const [_, endLog] = logs;
   t.is(endLog.level, 'WARN');
   t.is(endLog.message, 'Activity failed');
   const { durationMs, error, ...rest } = endLog.meta ?? {};
@@ -59,9 +123,11 @@ test('ActivityInboundLogInterceptor logs warning when activity fails', async (t)
 });
 
 test('ActivityInboundLogInterceptor logs when activity completes async', async (t) => {
-  const [_, endLog] = await runActivity(async () => {
+  const logs = await runActivity(async () => {
     throw new activity.CompleteAsyncError();
   });
+  t.is(logs.length, 2);
+  const [_, endLog] = logs;
   t.is(endLog.level, 'DEBUG');
   t.is(endLog.message, 'Activity will complete asynchronously');
   const { durationMs, ...rest } = endLog.meta ?? {};
@@ -70,7 +136,7 @@ test('ActivityInboundLogInterceptor logs when activity completes async', async (
 });
 
 test('ActivityInboundLogInterceptor logs when activity is cancelled with promise', async (t) => {
-  const [_, endLog] = await runActivity(
+  const logs = await runActivity(
     async () => {
       activity.Context.current().heartbeat();
       await activity.Context.current().cancelled;
@@ -79,6 +145,8 @@ test('ActivityInboundLogInterceptor logs when activity is cancelled with promise
       env.cancel();
     }
   );
+  t.is(logs.length, 2);
+  const [_, endLog] = logs;
   t.is(endLog.level, 'DEBUG');
   t.is(endLog.message, 'Activity completed as cancelled');
   const { durationMs, ...rest } = endLog.meta ?? {};
@@ -87,7 +155,7 @@ test('ActivityInboundLogInterceptor logs when activity is cancelled with promise
 });
 
 test('ActivityInboundLogInterceptor logs when activity is cancelled with signal', async (t) => {
-  const [_, endLog] = await runActivity(
+  const logs = await runActivity(
     async () => {
       await withZeroesHTTPServer(async (port) => {
         await cancellableFetch(`http:127.0.0.1:${port}`);
@@ -97,6 +165,8 @@ test('ActivityInboundLogInterceptor logs when activity is cancelled with signal'
       env.cancel();
     }
   );
+  t.is(logs.length, 2);
+  const [_, endLog] = logs;
   t.is(endLog.level, 'DEBUG');
   t.is(endLog.message, 'Activity completed as cancelled');
   const { durationMs, ...rest } = endLog.meta ?? {};
diff --git a/packages/testing/src/index.ts b/packages/testing/src/index.ts
@@ -409,7 +409,8 @@ export class MockActivityEnvironment extends events.EventEmitter {
       { ...defaultActivityInfo, ...info },
       promise,
       abortController.signal,
-      heartbeatCallback
+      heartbeatCallback,
+      Runtime.instance().logger
     );
     promise.catch(() => {
       /* avoid unhandled rejection */
diff --git a/packages/worker/src/activity-log-interceptor.ts b/packages/worker/src/activity-log-interceptor.ts
@@ -25,7 +25,33 @@ export function activityLogAttributes(info: Info): Record<string, unknown> {
 
 /** Logs Activity execution starts and their completions */
 export class ActivityInboundLogInterceptor implements ActivityInboundCallsInterceptor {
-  constructor(protected readonly ctx: Context, protected readonly logger: Logger) {}
+  /**
+   * @deprecated Use `Context.current().logger` instead
+   */
+  protected readonly logger: Logger;
+
+  constructor(protected readonly ctx: Context, logger?: Logger | undefined) {
+    // If a parent logger was explicitly provided on this interceptor, then use it.
+    // Otherwise, use the logger that is already set on the activity context.
+    // By default, that will be Runtime.logger, but another interceptor might have overriden it,
+    // in which case we would want to use that one as our parent logger.
+    const parentLogger = logger ?? ctx.logger;
+    this.logger = parentLogger; // eslint-disable-this-line deprecation/deprecation
+
+    this.ctx.logger = Object.fromEntries(
+      (['trace', 'debug', 'info', 'warn', 'error'] as const).map((level) => {
+        return [
+          level,
+          (message: string, attrs: Record<string, unknown>) => {
+            return parentLogger[level](message, {
+              ...this.logAttributes(),
+              ...attrs,
+            });
+          },
+        ];
+      })
+    ) as any;
+  }
 
   protected logAttributes(): Record<string, unknown> {
     return activityLogAttributes(this.ctx.info);
@@ -34,7 +60,7 @@ export class ActivityInboundLogInterceptor implements ActivityInboundCallsInterc
   async execute(input: ActivityExecuteInput, next: Next<ActivityInboundCallsInterceptor, 'execute'>): Promise<unknown> {
     let error: any = UNINITIALIZED; // In case someone decides to throw undefined...
     const startTime = process.hrtime.bigint();
-    this.logger.debug('Activity started', this.logAttributes());
+    this.ctx.logger.debug('Activity started');
     try {
       return await next(input);
     } catch (err: any) {
@@ -45,18 +71,18 @@ export class ActivityInboundLogInterceptor implements ActivityInboundCallsInterc
       const durationMs = Number(durationNanos / 1_000_000n);
 
       if (error === UNINITIALIZED) {
-        this.logger.debug('Activity completed', { durationMs, ...this.logAttributes() });
+        this.ctx.logger.debug('Activity completed', { durationMs });
       } else if (
         typeof error === 'object' &&
         error != null &&
         (CancelledFailure.is(error) || error.name === 'AbortError') &&
         this.ctx.cancellationSignal.aborted
       ) {
-        this.logger.debug('Activity completed as cancelled', { durationMs, ...this.logAttributes() });
+        this.ctx.logger.debug('Activity completed as cancelled', { durationMs });
       } else if (CompleteAsyncError.is(error)) {
-        this.logger.debug('Activity will complete asynchronously', { durationMs, ...this.logAttributes() });
+        this.ctx.logger.debug('Activity will complete asynchronously', { durationMs });
       } else {
-        this.logger.warn('Activity failed', { error, durationMs, ...this.logAttributes() });
+        this.ctx.logger.warn('Activity failed', { error, durationMs });
       }
     }
   }
diff --git a/packages/worker/src/activity.ts b/packages/worker/src/activity.ts
@@ -16,6 +16,7 @@ import {
   ActivityInboundCallsInterceptor,
   ActivityInboundCallsInterceptorFactory,
 } from './interceptors';
+import { Runtime } from './runtime';
 
 export type CancelReason =
   | keyof typeof coresdk.activity_task.ActivityCancelReason
@@ -47,7 +48,13 @@ export class Activity {
         reject(new CancelledFailure(reason));
       };
     });
-    this.context = new Context(info, promise, this.abortController.signal, this.heartbeatCallback);
+    this.context = new Context(
+      info,
+      promise,
+      this.abortController.signal,
+      this.heartbeatCallback,
+      Runtime.instance().logger
+    );
     // Prevent unhandled rejection
     promise.catch(() => undefined);
     this.interceptors = {
diff --git a/packages/worker/src/logger.ts b/packages/worker/src/logger.ts
diff --git a/packages/worker/src/worker-options.ts b/packages/worker/src/worker-options.ts
