feat: Revise SDK log attributes and levels (#750)

Author: bergundy

packages/activity/src/index.ts

@@ -148,6 +148,11 @@ export interface Info {
    * Use this in order to resume your Activity from checkpoint.
    */
   heartbeatDetails: any;
+
+  /**
+   * Task queue the Activity is scheduled in, set to the Workflow's task queue in case of local Activity.
+   */
+  taskQueue: string;
 }
 
 /**

packages/common/src/failure.ts

@@ -181,6 +181,8 @@ export class TimeoutFailure extends TemporalFailure {
  * This exception is expected to be thrown only by the framework code.
  */
 export class ActivityFailure extends TemporalFailure {
+  public readonly name: string = 'ActivityFailure';
+
   public constructor(
     public readonly activityType: string,
     public readonly activityId: string | undefined,
@@ -199,6 +201,8 @@ export class ActivityFailure extends TemporalFailure {
  * This exception is expected to be thrown only by the framework code.
  */
 export class ChildWorkflowFailure extends TemporalFailure {
+  public readonly name: string = 'ChildWorkflowFailure';
+
   public constructor(
     public readonly namespace: string | undefined,
     public readonly execution: WorkflowExecution,

packages/test/src/run-a-worker.ts

@@ -1,17 +1,18 @@
 import arg from 'arg';
-import { Worker, Runtime, DefaultLogger } from '@temporalio/worker';
+import { Worker, Runtime, DefaultLogger, LogLevel } from '@temporalio/worker';
 import * as activities from './activities';
 
 async function main() {
   const argv = arg({
-    '--debug': Boolean,
+    '--log-level': String,
   });
-  if (argv['--debug']) {
+  if (argv['--log-level']) {
+    const logLevel = argv['--log-level'].toUpperCase();
     Runtime.install({
-      logger: new DefaultLogger('DEBUG'),
+      logger: new DefaultLogger(logLevel as LogLevel),
       telemetryOptions: {
         tracingFilter: 'temporal_sdk_core=DEBUG',
-        logging: { forward: { level: 'DEBUG' } },
+        logging: { forward: { level: logLevel as LogLevel } },
       },
     });
   }

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

@@ -0,0 +1,105 @@
+import test from 'ava';
+import { ActivityInboundLogInterceptor, activityLogAttributes, DefaultLogger, LogEntry } from '@temporalio/worker';
+import { MockActivityEnvironment, defaultActivityInfo } from '@temporalio/testing';
+import { composeInterceptors } from '@temporalio/internal-workflow-common';
+import * as activity from '@temporalio/activity';
+import { withZeroesHTTPServer } from './zeroes-http-server';
+import { cancellableFetch } from './activities/cancellable-fetch';
+
+async function runActivity(
+  fn: activity.ActivityFunction,
+  heartbeatCallback = (_env: MockActivityEnvironment) => {
+    // not an empty body eslint
+  }
+): Promise<[LogEntry, LogEntry]> {
+  const logs = Array<LogEntry>();
+  const env = new MockActivityEnvironment();
+  env.on('heartbeat', () => heartbeatCallback(env));
+  await env.run(async () => {
+    const ctx = activity.Context.current();
+    const interceptor = new ActivityInboundLogInterceptor(
+      ctx,
+      new DefaultLogger('DEBUG', (entry) => {
+        logs.push(entry);
+      })
+    );
+    const execute = composeInterceptors([interceptor], 'execute', fn);
+    try {
+      await execute({ args: [], headers: {} });
+    } catch {
+      // ignore
+    }
+  });
+  if (logs.length !== 2) {
+    throw new Error('Expected exactly 2 log entries');
+  }
+  return logs as [LogEntry, LogEntry];
+}
+
+test('ActivityInboundLogInterceptor logs when activity starts', async (t) => {
+  const [startLog] = await runActivity(async () => {
+    // not an empty body eslint
+  });
+  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 () => {
+    throw err;
+  });
+  t.is(endLog.level, 'WARN');
+  t.is(endLog.message, 'Activity failed');
+  const { durationMs, error, ...rest } = endLog.meta ?? {};
+  t.true(Number.isInteger(durationMs));
+  t.is(err, error);
+  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
+});
+
+test('ActivityInboundLogInterceptor logs when activity completes async', async (t) => {
+  const [_, endLog] = await runActivity(async () => {
+    throw new activity.CompleteAsyncError();
+  });
+  t.is(endLog.level, 'DEBUG');
+  t.is(endLog.message, 'Activity will complete asynchronously');
+  const { durationMs, ...rest } = endLog.meta ?? {};
+  t.true(Number.isInteger(durationMs));
+  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
+});
+
+test('ActivityInboundLogInterceptor logs when activity is cancelled with promise', async (t) => {
+  const [_, endLog] = await runActivity(
+    async () => {
+      activity.Context.current().heartbeat();
+      await activity.Context.current().cancelled;
+    },
+    (env) => {
+      env.cancel();
+    }
+  );
+  t.is(endLog.level, 'DEBUG');
+  t.is(endLog.message, 'Activity completed as cancelled');
+  const { durationMs, ...rest } = endLog.meta ?? {};
+  t.true(Number.isInteger(durationMs));
+  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
+});
+
+test('ActivityInboundLogInterceptor logs when activity is cancelled with signal', async (t) => {
+  const [_, endLog] = await runActivity(
+    async () => {
+      await withZeroesHTTPServer(async (port) => {
+        await cancellableFetch(`http:127.0.0.1:${port}`);
+      });
+    },
+    (env) => {
+      env.cancel();
+    }
+  );
+  t.is(endLog.level, 'DEBUG');
+  t.is(endLog.message, 'Activity completed as cancelled');
+  const { durationMs, ...rest } = endLog.meta ?? {};
+  t.true(Number.isInteger(durationMs));
+  t.deepEqual(rest, activityLogAttributes(defaultActivityInfo));
+});

packages/test/src/test-custom-payload-converter.ts

@@ -121,6 +121,7 @@ test('Worker with proto data converter runs an activity and reports completion',
       taskToken,
       start: {
         activityType: 'protoActivity',
+        workflowExecution: { workflowId: 'wfid', runId: 'runId' },
         input: toPayloads(payloadConverter, messageInstance),
       },
     });

packages/test/src/test-worker-activities.ts

@@ -59,6 +59,7 @@ test('Worker runs an activity and reports completion', async (t) => {
       taskToken,
       start: {
         activityType: 'httpGet',
+        workflowExecution: { workflowId: 'wfid', runId: 'runId' },
         input: toPayloads(defaultPayloadConverter, url),
       },
     });
@@ -77,6 +78,7 @@ test('Worker runs an activity and reports failure', async (t) => {
       taskToken,
       start: {
         activityType: 'throwAnError',
+        workflowExecution: { workflowId: 'wfid', runId: 'runId' },
         input: toPayloads(defaultPayloadConverter, false, message),
       },
     });
@@ -105,6 +107,7 @@ test('Worker cancels activity and reports cancellation', async (t) => {
         taskToken,
         start: {
           activityType: 'waitForCancellation',
+          workflowExecution: { workflowId: 'wfid', runId: 'runId' },
           input: toPayloads(defaultPayloadConverter),
         },
       },
@@ -131,6 +134,7 @@ test('Activity Context AbortSignal cancels a fetch request', async (t) => {
           taskToken,
           start: {
             activityType: 'cancellableFetch',
+            workflowExecution: { workflowId: 'wfid', runId: 'runId' },
             input: toPayloads(defaultPayloadConverter, `http://127.0.0.1:${port}`, false),
           },
         },
@@ -158,6 +162,7 @@ test('Activity cancel with reason "NOT_FOUND" is valid', async (t) => {
           taskToken,
           start: {
             activityType: 'cancellableFetch',
+            workflowExecution: { workflowId: 'wfid', runId: 'runId' },
             input: toPayloads(defaultPayloadConverter, `http://127.0.0.1:${port}`, false),
           },
         },
@@ -183,6 +188,7 @@ test('Activity Context heartbeat is sent to core', async (t) => {
       taskToken,
       start: {
         activityType: 'progressiveSleep',
+        workflowExecution: { workflowId: 'wfid', runId: 'runId' },
         input: toPayloads(defaultPayloadConverter),
       },
     });
@@ -206,6 +212,7 @@ test('Worker fails activity with proper message when it is not registered', asyn
       taskToken,
       start: {
         activityType: 'notFound',
+        workflowExecution: { workflowId: 'wfid', runId: 'runId' },
         input: toPayloads(defaultPayloadConverter),
       },
     });
@@ -240,6 +247,7 @@ test('Worker cancels activities after shutdown', async (t) => {
         taskToken,
         start: {
           activityType: 'cancellationSnitch',
+          workflowExecution: { workflowId: 'wfid', runId: 'runId' },
           input: toPayloads(defaultPayloadConverter),
         },
       }),
@@ -271,6 +279,7 @@ test('Non ApplicationFailure TemporalFailures thrown from Activity are wrapped w
       taskToken,
       start: {
         activityType: 'throwTemporalFailure',
+        workflowExecution: { workflowId: 'wfid', runId: 'runId' },
         input: toPayloads(defaultPayloadConverter),
       },
     });

packages/test/src/test-worker-heartbeats.ts

@@ -8,7 +8,10 @@ import { isolateFreeWorker, Worker } from './mock-native-worker';
 async function runActivity(worker: Worker, callback?: (completion: coresdk.ActivityTaskCompletion) => void) {
   const taskToken = Buffer.from(uuid4());
   await worker.runUntil(async () => {
-    const completion = await worker.native.runActivityTask({ taskToken, start: { activityType: 'rapidHeartbeater' } });
+    const completion = await worker.native.runActivityTask({
+      taskToken,
+      start: { activityType: 'rapidHeartbeater', workflowExecution: { workflowId: 'wfid', runId: 'runid' } },
+    });
     callback?.(completion);
   });
 }

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

@@ -0,0 +1,90 @@
+import anyTest, { TestInterface, ExecutionContext } from 'ava';
+import { v4 as uuid4 } from 'uuid';
+import { Worker, DefaultLogger, LogEntry, defaultSinks } from '@temporalio/worker';
+import { TestWorkflowEnvironment } from '@temporalio/testing';
+import * as workflows from './workflows';
+
+interface Context {
+  testEnv: TestWorkflowEnvironment;
+}
+const test = anyTest as TestInterface<Context>;
+
+test.before(async (t) => {
+  t.context = {
+    testEnv: await TestWorkflowEnvironment.create({
+      testServer: {
+        stdio: 'inherit',
+      },
+    }),
+  };
+});
+
+test.after.always(async (t) => {
+  await t.context.testEnv?.teardown();
+});
+
+async function withWorker(t: ExecutionContext<Context>, p: Promise<any>): Promise<[LogEntry, LogEntry]> {
+  const { nativeConnection } = t.context.testEnv;
+  const logs = Array<LogEntry>();
+  const logger = new DefaultLogger('DEBUG', (entry) => logs.push(entry));
+  const worker = await Worker.create({
+    connection: nativeConnection,
+    taskQueue: 'test',
+    workflowsPath: require.resolve('./workflows'),
+    sinks: defaultSinks(logger),
+  });
+  await worker.runUntil(p);
+  t.true(logs.length >= 2);
+  return logs as [LogEntry, LogEntry];
+}
+
+test.serial('WorkflowInboundLogInterceptor logs when workflow completes', async (t) => {
+  const { workflowClient } = t.context.testEnv;
+  const workflowId = uuid4();
+  const [startLog, endLog] = await withWorker(
+    t,
+    workflowClient.execute(workflows.successString, { workflowId, taskQueue: 'test' })
+  );
+  t.is(startLog.level, 'DEBUG');
+  t.is(startLog.message, 'Workflow started');
+  t.is(startLog.meta?.workflowId, workflowId);
+  t.true(typeof startLog.meta?.runId === 'string');
+  t.is(startLog.meta?.taskQueue, 'test');
+  t.is(startLog.meta?.namespace, 'default');
+  t.is(startLog.meta?.workflowType, 'successString');
+  t.is(endLog.level, 'DEBUG');
+  t.is(endLog.message, 'Workflow completed');
+});
+
+test.serial('WorkflowInboundLogInterceptor logs when workflow continues as new', async (t) => {
+  const { workflowClient } = t.context.testEnv;
+  const [_, endLog] = await withWorker(
+    t,
+    t.throwsAsync(
+      workflowClient.execute(workflows.continueAsNewSameWorkflow, {
+        args: ['execute', 'execute'],
+        workflowId: uuid4(),
+        taskQueue: 'test',
+        followRuns: false,
+      })
+    )
+  );
+  t.is(endLog.level, 'DEBUG');
+  t.is(endLog.message, 'Workflow continued as new');
+});
+
+test.serial('WorkflowInboundLogInterceptor logs warning when workflow fails', async (t) => {
+  const { workflowClient } = t.context.testEnv;
+  const [_, endLog] = await withWorker(
+    t,
+    t.throwsAsync(
+      workflowClient.execute(workflows.throwAsync, {
+        workflowId: uuid4(),
+        taskQueue: 'test',
+        followRuns: false,
+      })
+    )
+  );
+  t.is(endLog.level, 'WARN');
+  t.is(endLog.message, 'Workflow failed');
+});

packages/testing/src/index.ts

@@ -292,6 +292,7 @@ export class TestWorkflowEnvironment {
  */
 export const defaultActivityInfo: activity.Info = {
   attempt: 1,
+  taskQueue: 'test',
   isLocal: false,
   taskToken: Buffer.from('test'),
   activityId: 'test',
@@ -333,6 +334,9 @@ export class MockActivityEnvironment extends events.EventEmitter {
       abortController.signal,
       heartbeatCallback
     );
+    promise.catch(() => {
+      /* avoid unhandled rejection */
+    });
   }
 
   /**