Add sdkComponent and taskQueue metadata on log messages (#1401)

Author: mjameswh

.github/workflows/ci.yml

@@ -17,7 +17,7 @@ jobs:
       matrix:
         reuse-v8-context: [true]
         node: [14, 16, 18, 20]
-        os: [ubuntu-latest, macos-latest, windows-latest]
+        os: [ubuntu-latest, macos-12, windows-latest]
         include:
           - os: ubuntu-latest
             node: 20
@@ -148,9 +148,9 @@ jobs:
           - runsOn: buildjet-2vcpu-ubuntu-1804
             target: aarch64-unknown-linux-gnu
             rustflags: '-C linker=aarch64-linux-gnu-gcc'
-          - runsOn: macos-latest
+          - runsOn: macos-12
             target: x86_64-apple-darwin
-          - runsOn: macos-latest
+          - runsOn: macos-12
             target: aarch64-apple-darwin
           - runsOn: windows-latest
             target: x86_64-pc-windows-msvc
@@ -216,7 +216,7 @@ jobs:
             node: 20
             server: local
             sample: fetch-esm
-          - os: macos-latest
+          - os: macos-12
             target: x86_64-apple-darwin
             node: 20
             server: cloud

packages/activity/src/index.ts

@@ -254,8 +254,8 @@ export class Context {
    * The logger for this Activity.
    *
    * This defaults to the `Runtime`'s Logger (see {@link Runtime.logger}). Attributes from the current Activity context
-   * will automatically be included as metadata on every log entries, and some key events of the Activity's lifecycle
-   * will automatically be logged (at 'DEBUG' level for most messages; 'WARN' for failures).
+   * are automatically included as metadata on every log entries. An extra `sdkComponent` metadata attribute is also
+   * added, with value `activity`; this can be used for fine-grained filtering of log entries further downstream.
    *
    * To customize log attributes, register a {@link ActivityOutboundCallsInterceptor} that intercepts the
    * `getLogAttributes()` method.

packages/common/src/logger.ts

@@ -13,3 +13,43 @@ export interface Logger {
   warn(message: string, meta?: LogMetadata): any;
   error(message: string, meta?: LogMetadata): any;
 }
+
+/**
+ * Possible values of the `sdkComponent` meta attributes on log messages. This
+ * attribute indicates which subsystem emitted the log message; this may for
+ * example be used to implement fine-grained filtering of log messages.
+ *
+ * Note that there is no guarantee that this list will remain stable in the
+ * future; values may be added or removed, and messages that are currently
+ * emitted with some `sdkComponent` value may use a different value in the future.
+ */
+export enum SdkComponent {
+  /**
+   * Component name for messages emited from Workflow code, using the {@link Workflow context logger|workflow.log}.
+   * The SDK itself never publishes messages with this component name.
+   */
+  workflow = 'workflow',
+
+  /**
+   * Component name for messages emited from an activity, using the {@link activity context logger|Context.log}.
+   * The SDK itself never publishes messages with this component name.
+   */
+  activity = 'activity',
+
+  /**
+   * Component name for messages emited from a Temporal Worker instance.
+   *
+   * This notably includes:
+   * - Issues with Worker or runtime configuration, or the JS execution environment;
+   * - Worker's, Activity's, and Workflow's lifecycle events;
+   * - Workflow Activation and Activity Task processing events;
+   * - Workflow bundling messages;
+   * - Sink processing issues.
+   */
+  worker = 'worker',
+
+  /**
+   * Component name for all messages emitted by the Rust Core SDK library.
+   */
+  core = 'core',
+}

packages/test/src/mock-native-worker.ts

@@ -1,17 +1,13 @@
 /* eslint-disable @typescript-eslint/no-non-null-assertion */
 import { lastValueFrom } from 'rxjs';
-import { defaultPayloadConverter, fromPayloadsAtIndex } from '@temporalio/common';
+import { SdkComponent, defaultPayloadConverter, fromPayloadsAtIndex } from '@temporalio/common';
 import { msToTs } from '@temporalio/common/lib/time';
 import { coresdk } from '@temporalio/proto';
 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 {
-  addDefaultWorkerOptions,
-  CompiledWorkerOptions,
-  compileWorkerOptions,
-  WorkerOptions,
-} from '@temporalio/worker/lib/worker-options';
+import { withMetadata } from '@temporalio/worker/lib/logger';
+import { CompiledWorkerOptions, compileWorkerOptions, WorkerOptions } from '@temporalio/worker/lib/worker-options';
 import type { WorkflowCreator } from '@temporalio/worker/lib/workflow/interface';
 import * as activities from './activities';
 
@@ -163,11 +159,12 @@ 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 logger = withMetadata(Runtime.instance().logger, {
+      sdkComponent: SdkComponent.worker,
+      taskQueue: opts.taskQueue,
+    });
     const nativeWorker = new MockNativeWorker();
-    super(nativeWorker, workflowCreator, opts);
+    super(nativeWorker, workflowCreator, opts, logger);
   }
 
   public runWorkflows(...args: Parameters<Worker['workflow$']>): Promise<void> {
@@ -183,6 +180,10 @@ export const defaultOptions: WorkerOptions = {
 };
 
 export function isolateFreeWorker(options: WorkerOptions = defaultOptions): Worker {
+  const logger = withMetadata(Runtime.instance().logger, {
+    sdkComponent: SdkComponent.worker,
+    taskQueue: options.taskQueue ?? 'default',
+  });
   return new Worker(
     {
       async createWorkflow() {
@@ -192,6 +193,6 @@ export function isolateFreeWorker(options: WorkerOptions = defaultOptions): Work
         /* Nothing to destroy */
       },
     },
-    compileWorkerOptions(addDefaultWorkerOptions(options))
+    compileWorkerOptions(options, logger)
   );
 }

packages/test/src/test-activity-log-interceptor.ts

@@ -5,47 +5,52 @@ import { MockActivityEnvironment, defaultActivityInfo } from '@temporalio/testin
 import { isCancellation } from '@temporalio/workflow';
 import { isAbortError } from '@temporalio/common/lib/type-helpers';
 import * as activity from '@temporalio/activity';
+import { SdkComponent } from '@temporalio/common';
 import { withZeroesHTTPServer } from './zeroes-http-server';
 import { cancellableFetch } from './activities';
 
 interface MyTestActivityContext extends activity.Context {
   logs: Array<LogEntry>;
 }
 
-test.before(() => {
-  const mockLogger = new DefaultLogger('DEBUG', (entry) => {
+const mockLogger = new DefaultLogger('DEBUG', (entry) => {
+  try {
     (activity.Context.current() as MyTestActivityContext).logs ??= [];
     (activity.Context.current() as MyTestActivityContext).logs.push(entry);
-  });
-  Runtime.install({
-    logger: mockLogger,
-  });
+  } catch (e) {
+    // Ignore messages produced from non activity context
+    if ((e as Error).message !== 'Activity context not initialized') throw e;
+  }
+});
+Runtime.install({
+  logger: mockLogger,
 });
 
-test("Activity Context logger defaults to Runtime's Logger", async (t) => {
-  const env = new MockActivityEnvironment({});
+test('Activity Context logger funnel through the parent Logger', async (t) => {
+  const env = new MockActivityEnvironment({}, { logger: mockLogger });
   await env.run(async () => {
     activity.log.debug('log message from activity');
   });
   const logs = (env.context as MyTestActivityContext).logs;
   const entry = logs.find((x) => x.level === 'DEBUG' && x.message === 'log message from activity');
   t.not(entry, undefined);
+  t.deepEqual(entry?.meta, { ...activityLogAttributes(defaultActivityInfo), sdkComponent: SdkComponent.activity });
 });
 
 test('Activity Worker logs when activity starts', async (t) => {
-  const env = new MockActivityEnvironment({});
+  const env = new MockActivityEnvironment({}, { logger: mockLogger });
   await env.run(async () => {
     activity.log.debug('log message from activity');
   });
   const logs = (env.context as MyTestActivityContext).logs;
   const entry = logs.find((x) => x.level === 'DEBUG' && x.message === 'Activity started');
   t.not(entry, undefined);
-  t.deepEqual(entry?.meta, activityLogAttributes(defaultActivityInfo));
+  t.deepEqual(entry?.meta, { ...activityLogAttributes(defaultActivityInfo), sdkComponent: SdkComponent.worker });
 });
 
 test('Activity Worker logs warning when activity fails', async (t) => {
   const err = new Error('Failed for test');
-  const env = new MockActivityEnvironment({});
+  const env = new MockActivityEnvironment({}, { logger: mockLogger });
   try {
     await env.run(async () => {
       throw err;
@@ -60,11 +65,11 @@ test('Activity Worker logs warning when activity fails', async (t) => {
   const { durationMs, error, ...rest } = entry?.meta ?? {};
   t.true(Number.isInteger(durationMs));
   t.is(err, error);
-  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
+  t.deepEqual(rest, { ...activityLogAttributes(defaultActivityInfo), sdkComponent: SdkComponent.worker });
 });
 
 test('Activity Worker logs when activity completes async', async (t) => {
-  const env = new MockActivityEnvironment({});
+  const env = new MockActivityEnvironment({}, { logger: mockLogger });
   try {
     await env.run(async () => {
       throw new activity.CompleteAsyncError();
@@ -77,11 +82,11 @@ test('Activity Worker logs when activity completes async', async (t) => {
   t.not(entry, undefined);
   const { durationMs, ...rest } = entry?.meta ?? {};
   t.true(Number.isInteger(durationMs));
-  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
+  t.deepEqual(rest, { ...activityLogAttributes(defaultActivityInfo), sdkComponent: SdkComponent.worker });
 });
 
 test('Activity Worker logs when activity is cancelled with promise', async (t) => {
-  const env = new MockActivityEnvironment({});
+  const env = new MockActivityEnvironment({}, { logger: mockLogger });
   env.on('heartbeat', () => env.cancel());
   try {
     await env.run(async () => {
@@ -96,11 +101,11 @@ test('Activity Worker logs when activity is cancelled with promise', async (t) =
   t.not(entry, undefined);
   const { durationMs, ...rest } = entry?.meta ?? {};
   t.true(Number.isInteger(durationMs));
-  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
+  t.deepEqual(rest, { ...activityLogAttributes(defaultActivityInfo), sdkComponent: SdkComponent.worker });
 });
 
 test('Activity Worker logs when activity is cancelled with signal', async (t) => {
-  const env = new MockActivityEnvironment({});
+  const env = new MockActivityEnvironment({}, { logger: mockLogger });
   env.on('heartbeat', () => env.cancel());
   try {
     await env.run(async () => {
@@ -116,7 +121,7 @@ test('Activity Worker logs when activity is cancelled with signal', async (t) =>
   t.not(entry, undefined);
   const { durationMs, ...rest } = entry?.meta ?? {};
   t.true(Number.isInteger(durationMs));
-  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
+  t.deepEqual(rest, { ...activityLogAttributes(defaultActivityInfo), sdkComponent: SdkComponent.worker });
 });
 
 test('(Legacy) ActivityInboundLogInterceptor does not override Context.log by default', async (t) => {
@@ -125,6 +130,7 @@ test('(Legacy) ActivityInboundLogInterceptor does not override Context.log by de
     {
       // eslint-disable-next-line deprecation/deprecation
       interceptors: [(ctx) => ({ inbound: new ActivityInboundLogInterceptor(ctx) })],
+      logger: mockLogger,
     }
   );
   await env.run(async () => {
@@ -146,15 +152,14 @@ test('(Legacy) ActivityInboundLogInterceptor overrides Context.log if a logger i
     {
       // eslint-disable-next-line deprecation/deprecation
       interceptors: [(ctx) => ({ inbound: new ActivityInboundLogInterceptor(ctx, logger) })],
+      logger: mockLogger,
     }
   );
   await env.run(async () => {
     activity.log.debug('log message from activity');
   });
-  const entry1 = logs.find((x) => x.level === 'DEBUG' && x.message === 'Activity started');
-  t.not(entry1, undefined);
-  const entry2 = logs.find((x) => x.level === 'DEBUG' && x.message === 'log message from activity');
-  t.not(entry2, undefined);
+  const entry = logs.find((x) => x.level === 'DEBUG' && x.message === 'log message from activity');
+  t.not(entry, undefined);
 });
 
 test('(Legacy) ActivityInboundLogInterceptor overrides Context.log if class is extended', async (t) => {
@@ -172,6 +177,7 @@ test('(Legacy) ActivityInboundLogInterceptor overrides Context.log if class is e
     {},
     {
       interceptors: [(ctx) => ({ inbound: new CustomActivityInboundLogInterceptor(ctx) })],
+      logger: mockLogger,
     }
   );
   await env.run(async () => {

packages/test/src/test-mockactivityenv.ts

@@ -1,6 +1,16 @@
 import test from 'ava';
 import { MockActivityEnvironment } from '@temporalio/testing';
-import { CancelledFailure, Context } from '@temporalio/activity';
+import * as activity from '@temporalio/activity';
+import { Runtime } from '@temporalio/worker';
+
+test("MockActivityEnvironment doesn't implicitly instantiate Runtime", async (t) => {
+  t.is(Runtime._instance, undefined);
+  const env = new MockActivityEnvironment();
+  await env.run(async (): Promise<void> => {
+    activity.log.info('log message from activity');
+  });
+  t.is(Runtime._instance, undefined);
+});
 
 test('MockActivityEnvironment can run a single activity', async (t) => {
   const env = new MockActivityEnvironment();
@@ -19,12 +29,12 @@ test('MockActivityEnvironment emits heartbeat events and can be cancelled', asyn
   });
   await t.throwsAsync(
     env.run(async (x: number): Promise<number> => {
-      Context.current().heartbeat(6);
-      await Context.current().sleep(100);
+      activity.heartbeat(6);
+      await activity.sleep(100);
       return x + 1;
     }, 3),
     {
-      instanceOf: CancelledFailure,
+      instanceOf: activity.CancelledFailure,
       message: 'test',
     }
   );
@@ -33,7 +43,7 @@ test('MockActivityEnvironment emits heartbeat events and can be cancelled', asyn
 test('MockActivityEnvironment injects provided info', async (t) => {
   const env = new MockActivityEnvironment({ attempt: 3 });
   const res = await env.run(async (x: number): Promise<number> => {
-    return x + Context.current().info.attempt;
+    return x + activity.activityInfo().attempt;
   }, 1);
   t.is(res, 4);
 });

packages/test/src/test-runtime.ts

@@ -112,7 +112,7 @@ if (RUN_INTEGRATION_TESTS) {
       t.is(typeof failingWftEntry.meta?.['failure'], 'string');
       t.is(typeof failingWftEntry.meta?.['runId'], 'string');
       t.is(typeof failingWftEntry.meta?.['workflowId'], 'string');
-      t.is(typeof failingWftEntry.meta?.['subsystem'], 'string');
+      t.is(typeof failingWftEntry.meta?.['sdkComponent'], 'string');
     } finally {
       await Runtime.instance().shutdown();
     }

packages/test/src/test-sinks.ts

@@ -13,6 +13,7 @@ import {
 import { LoggerSinksInternal as DefaultLoggerSinks } from '@temporalio/workflow/lib/logs';
 import { SearchAttributes, WorkflowInfo } from '@temporalio/workflow';
 import { UnsafeWorkflowInfo } from '@temporalio/workflow/src/interfaces';
+import { SdkComponent } from '@temporalio/common';
 import { RUN_INTEGRATION_TESTS, Worker, registerDefaultCustomSearchAttributes } from './helpers';
 import { defaultOptions } from './mock-native-worker';
 import * as workflows from './workflows';
@@ -166,6 +167,10 @@ if (RUN_INTEGRATION_TESTS) {
         meta: {
           ...x.meta,
           workflowInfo: fixWorkflowInfoDates(x.meta?.workflowInfo),
+          namespace: info.namespace,
+          runId: info.runId,
+          workflowId: info.workflowId,
+          workflowType: info.workflowType,
         },
         timestampNanos: undefined,
       })),
@@ -177,6 +182,12 @@ if (RUN_INTEGRATION_TESTS) {
           ifaceName: error.ifaceName,
           fnName: error.fnName,
           workflowInfo: info,
+          sdkComponent: SdkComponent.worker,
+          taskQueue,
+          namespace: info.namespace,
+          runId: info.runId,
+          workflowId: info.workflowId,
+          workflowType: info.workflowType,
         },
         timestampNanos: undefined,
       }))

packages/test/src/test-workflows.ts

@@ -8,6 +8,7 @@ import {
   ApplicationFailure,
   defaultFailureConverter,
   defaultPayloadConverter,
+  SdkComponent,
   Payload,
   RetryState,
   toPayloads,
@@ -1578,6 +1579,7 @@ test('logAndTimeout', async (t) => {
           taskQueue: 'test',
           workflowId: 'test-workflowId',
           workflowType: 'logAndTimeout',
+          sdkComponent: SdkComponent.worker,
         },
       ],
     },
@@ -1592,6 +1594,7 @@ test('logAndTimeout', async (t) => {
           taskQueue: 'test',
           workflowId: 'test-workflowId',
           workflowType: 'logAndTimeout',
+          sdkComponent: SdkComponent.workflow,
         },
       ],
     },