Revert "Add logSource and taskQueue metadata on log messages (#1399)

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. 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.
+   * 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).
    *
    * To customize log attributes, register a {@link ActivityOutboundCallsInterceptor} that intercepts the
    * `getLogAttributes()` method.

packages/common/src/logger.ts

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

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

@@ -20,15 +20,13 @@ interface InstallArgs {
  * @returns A Promise that resolves once the installation is finished.
  */
 export function install({ root, useYarn }: InstallArgs): Promise<void> {
-  const isWindows = process.platform === 'win32';
-  const npm = isWindows ? 'npm.cmd' : 'npm';
+  const npm = /^win/.test(process.platform) ? '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, registerDefaultCustomSearchAttributes } from './helpers';
+import { Worker, test as anyTest, bundlerOptions } from './helpers';
 
 export interface Context {
   env: TestWorkflowEnvironment;
@@ -59,7 +59,6 @@ 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,13 +1,17 @@
 /* eslint-disable @typescript-eslint/no-non-null-assertion */
 import { lastValueFrom } from 'rxjs';
-import { LogSource, defaultPayloadConverter, fromPayloadsAtIndex } from '@temporalio/common';
+import { 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 { withMetadata } from '@temporalio/worker/lib/logger';
-import { CompiledWorkerOptions, compileWorkerOptions, WorkerOptions } from '@temporalio/worker/lib/worker-options';
+import {
+  addDefaultWorkerOptions,
+  CompiledWorkerOptions,
+  compileWorkerOptions,
+  WorkerOptions,
+} from '@temporalio/worker/lib/worker-options';
 import type { WorkflowCreator } from '@temporalio/worker/lib/workflow/interface';
 import * as activities from './activities';
 
@@ -159,12 +163,11 @@ export class Worker extends RealWorker {
   }
 
   public constructor(workflowCreator: WorkflowCreator, opts: CompiledWorkerOptions) {
-    const logger = withMetadata(Runtime.instance().logger, {
-      logSource: LogSource.worker,
-      taskQueue: opts.taskQueue ?? 'default',
-    });
+    // 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, logger);
+    super(nativeWorker, workflowCreator, opts);
   }
 
   public runWorkflows(...args: Parameters<Worker['workflow$']>): Promise<void> {
@@ -180,10 +183,6 @@ 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() {
@@ -193,6 +192,6 @@ export function isolateFreeWorker(options: WorkerOptions = defaultOptions): Work
         /* Nothing to destroy */
       },
     },
-    compileWorkerOptions(options, logger)
+    compileWorkerOptions(addDefaultWorkerOptions(options))
   );
 }

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

@@ -5,52 +5,47 @@ 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>;
 }
 
-const mockLogger = new DefaultLogger('DEBUG', (entry) => {
-  try {
+test.before(() => {
+  const mockLogger = new DefaultLogger('DEBUG', (entry) => {
     (activity.Context.current() as MyTestActivityContext).logs ??= [];
     (activity.Context.current() as MyTestActivityContext).logs.push(entry);
-  } catch (e) {
-    // Ignore messages produced from non activity context
-    if ((e as Error).message !== 'Activity context not initialized') throw e;
-  }
-});
-Runtime.install({
-  logger: mockLogger,
+  });
+  Runtime.install({
+    logger: mockLogger,
+  });
 });
 
-test('Activity Context logger funnel through the parent Logger', async (t) => {
-  const env = new MockActivityEnvironment({}, { logger: mockLogger });
+test("Activity Context logger defaults to Runtime's Logger", async (t) => {
+  const env = new MockActivityEnvironment({});
   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({}, { logger: mockLogger });
+  const env = new MockActivityEnvironment({});
   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), logSource: LogSource.worker });
+  t.deepEqual(entry?.meta, activityLogAttributes(defaultActivityInfo));
 });
 
 test('Activity Worker logs warning when activity fails', async (t) => {
   const err = new Error('Failed for test');
-  const env = new MockActivityEnvironment({}, { logger: mockLogger });
+  const env = new MockActivityEnvironment({});
   try {
     await env.run(async () => {
       throw err;
@@ -65,11 +60,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), logSource: LogSource.worker });
+  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
 });
 
 test('Activity Worker logs when activity completes async', async (t) => {
-  const env = new MockActivityEnvironment({}, { logger: mockLogger });
+  const env = new MockActivityEnvironment({});
   try {
     await env.run(async () => {
       throw new activity.CompleteAsyncError();
@@ -82,11 +77,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), logSource: LogSource.worker });
+  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
 });
 
 test('Activity Worker logs when activity is cancelled with promise', async (t) => {
-  const env = new MockActivityEnvironment({}, { logger: mockLogger });
+  const env = new MockActivityEnvironment({});
   env.on('heartbeat', () => env.cancel());
   try {
     await env.run(async () => {
@@ -101,11 +96,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), logSource: LogSource.worker });
+  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
 });
 
 test('Activity Worker logs when activity is cancelled with signal', async (t) => {
-  const env = new MockActivityEnvironment({}, { logger: mockLogger });
+  const env = new MockActivityEnvironment({});
   env.on('heartbeat', () => env.cancel());
   try {
     await env.run(async () => {
@@ -121,7 +116,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), logSource: LogSource.worker });
+  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
 });
 
 test('(Legacy) ActivityInboundLogInterceptor does not override Context.log by default', async (t) => {
@@ -130,7 +125,6 @@ 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 () => {
@@ -152,14 +146,15 @@ 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 entry = logs.find((x) => x.level === 'DEBUG' && x.message === 'log message from activity');
-  t.not(entry, undefined);
+  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);
 });
 
 test('(Legacy) ActivityInboundLogInterceptor overrides Context.log if class is extended', async (t) => {
@@ -177,7 +172,6 @@ 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 = 1000;
+  const batchSize = 100;
   if (workflow.workflowInfo().continueAsNewSuggested) {
     return false;
   }
   while (workflow.workflowInfo().historyLength < maxEvents) {
-    await Promise.all(Array.from({ length: batchSize }, (_) => workflow.sleep(1)));
+    await Promise.all(new Array(batchSize).fill(undefined).map((_) => workflow.sleep(1)));
     if (workflow.workflowInfo().continueAsNewSuggested) {
       return true;
     }

packages/test/src/test-mockactivityenv.ts

@@ -1,16 +1,6 @@
 import test from 'ava';
 import { MockActivityEnvironment } from '@temporalio/testing';
-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);
-});
+import { CancelledFailure, Context } from '@temporalio/activity';
 
 test('MockActivityEnvironment can run a single activity', async (t) => {
   const env = new MockActivityEnvironment();
@@ -29,12 +19,12 @@ test('MockActivityEnvironment emits heartbeat events and can be cancelled', asyn
   });
   await t.throwsAsync(
     env.run(async (x: number): Promise<number> => {
-      activity.heartbeat(6);
-      await activity.sleep(100);
+      Context.current().heartbeat(6);
+      await Context.current().sleep(100);
       return x + 1;
     }, 3),
     {
-      instanceOf: activity.CancelledFailure,
+      instanceOf: CancelledFailure,
       message: 'test',
     }
   );
@@ -43,7 +33,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 + activity.activityInfo().attempt;
+    return x + Context.current().info.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?.['logSource'], 'string');
+      t.is(typeof failingWftEntry.meta?.['subsystem'], 'string');
     } finally {
       await Runtime.instance().shutdown();
     }