Add logSource and taskQueue metadata on log messages (#1391)

Author: mjameswh

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).
+   * will automatically be included as metadata on every log entries. An extra `logSource` metadata attribute will also
+   * be 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,44 @@ export interface Logger {
   warn(message: string, meta?: LogMetadata): any;
   error(message: string, meta?: LogMetadata): any;
 }
+
+/**
+ * Possible values of the `logSource` 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 `logSource` value may use a different value in the future.
+ */
+export enum LogSource {
+  /**
+   * Log Source value for messages emited from Workflow code, using the {@link Workflow
+   * context logger|workflow.log}. The SDK itself never publishes messages with this source.
+   */
+  workflow = 'workflow',
+
+  /**
+   * Log Source value for messages emited from an activity, using the {@link activity
+   * context logger|Context.log}. The SDK itself never publishes messages with this source.
+   */
+  activity = 'activity',
+
+  /**
+   * Log Source value for messages emited from a Temporal Worker instance.
+   *
+   * This notably includes:
+   * - Issues with worker or runtime options 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;
+   * - Issues in interracting with the Core SDK library.
+   */
+  worker = 'worker',
+
+  /**
+   * Log Source value for all messages emited by the Core SDK library.
+   */
+  core = 'core',
+}

packages/core-bridge/scripts/build.js

@@ -82,11 +82,12 @@ function compile(requestedTarget) {
   const cmd = which.sync('cargo-cp-artifact');
 
   console.log('Running', cmd, argv);
-  const { status } = spawnSync(cmd, argv, {
+  const { status, error } = spawnSync(cmd, argv, {
     stdio: 'inherit',
+    shell: process.platform === 'win32',
   });
-  if (status !== 0) {
-    throw new Error(`Failed to build${target ? ' for ' + target : ''}`);
+  if (status !== 0 || error) {
+    throw new Error(`Failed to build${target ? ' for ' + target : ''}: status code ${status}`, error);
   }
 }
 

packages/create-project/src/helpers/install.ts

@@ -20,13 +20,15 @@ interface InstallArgs {
  * @returns A Promise that resolves once the installation is finished.
  */
 export function install({ root, useYarn }: InstallArgs): Promise<void> {
-  const npm = /^win/.test(process.platform) ? 'npm.cmd' : 'npm';
+  const isWindows = process.platform === 'win32';
+  const npm = isWindows ? 'npm.cmd' : 'npm';
   const command: string = useYarn ? 'yarn' : npm;
 
   return spawn(command, ['install'], {
     cwd: root,
     stdio: 'inherit',
     env: { ...process.env, ADBLOCK: '1', DISABLE_OPENCOLLECTIVE: '1' },
+    shell: isWindows,
   });
 }
 

packages/test/src/helpers-integration.ts

@@ -21,7 +21,7 @@ import {
 } from '@temporalio/worker';
 import * as workflow from '@temporalio/workflow';
 import { ConnectionInjectorInterceptor } from './activities/interceptors';
-import { Worker, test as anyTest, bundlerOptions } from './helpers';
+import { Worker, test as anyTest, bundlerOptions, registerDefaultCustomSearchAttributes } from './helpers';
 
 export interface Context {
   env: TestWorkflowEnvironment;
@@ -59,6 +59,7 @@ export function makeTestFunction(opts: {
         ],
       },
     });
+    await registerDefaultCustomSearchAttributes(env.connection);
     const workflowBundle = await bundleWorkflowCode({
       ...bundlerOptions,
       workflowInterceptorModules: [...defaultWorkflowInterceptorModules, ...(opts.workflowInterceptorModules ?? [])],

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 { LogSource, 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, {
+      logSource: LogSource.worker,
+      taskQueue: opts.taskQueue ?? 'default',
+    });
     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, {
+    logSource: LogSource.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 { LogSource } 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), logSource: LogSource.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), logSource: LogSource.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), logSource: LogSource.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), logSource: LogSource.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), logSource: LogSource.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), logSource: LogSource.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-integration-workflows.ts

@@ -153,12 +153,12 @@ test('HistorySize is visible in WorkflowExecutionInfo', async (t) => {
 
 export async function suggestedCAN(): Promise<boolean> {
   const maxEvents = 40_000;
-  const batchSize = 100;
+  const batchSize = 1000;
   if (workflow.workflowInfo().continueAsNewSuggested) {
     return false;
   }
   while (workflow.workflowInfo().historyLength < maxEvents) {
-    await Promise.all(new Array(batchSize).fill(undefined).map((_) => workflow.sleep(1)));
+    await Promise.all(Array.from({ length: batchSize }, (_) => workflow.sleep(1)));
     if (workflow.workflowInfo().continueAsNewSuggested) {
       return true;
     }

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 instanciate 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?.['logSource'], 'string');
     } finally {
       await Runtime.instance().shutdown();
     }