chore: Deprecate defaultSinks (#1283)

Author: mjameswh

package-lock.json

@@ -1,18 +1,19 @@
 /* eslint @typescript-eslint/no-non-null-assertion: 0 */
 import test from 'ava';
 import { v4 as uuid4 } from 'uuid';
-import { WorkflowClient } from '@temporalio/client';
+import { Connection, WorkflowClient } from '@temporalio/client';
 import {
   DefaultLogger,
   InjectedSinks,
   Runtime,
-  LoggerSinks as DefaultLoggerSinks,
   InjectedSinkFunction,
   WorkerOptions,
+  LogEntry,
 } from '@temporalio/worker';
+import { LoggerSinksInternal as DefaultLoggerSinks } from '@temporalio/workflow/lib/logs';
 import { SearchAttributes, WorkflowInfo } from '@temporalio/workflow';
 import { UnsafeWorkflowInfo } from '@temporalio/workflow/src/interfaces';
-import { RUN_INTEGRATION_TESTS, Worker } from './helpers';
+import { RUN_INTEGRATION_TESTS, Worker, registerDefaultCustomSearchAttributes } from './helpers';
 import { defaultOptions } from './mock-native-worker';
 import * as workflows from './workflows';
 
@@ -22,12 +23,12 @@ class DependencyError extends Error {
   }
 }
 
-function asDefaultLoggerSink(
+function asSdkLoggerSink(
   fn: (info: WorkflowInfo, message: string, attrs?: Record<string, unknown>) => Promise<void>,
   opts?: Omit<InjectedSinkFunction<any>, 'fn'>
 ): InjectedSinks<DefaultLoggerSinks> {
   return {
-    defaultWorkerLogger: {
+    __temporal_logger: {
       trace: { fn, ...opts },
       debug: { fn, ...opts },
       info: { fn, ...opts },
@@ -38,17 +39,20 @@ function asDefaultLoggerSink(
 }
 
 if (RUN_INTEGRATION_TESTS) {
-  const recordedLogs: any[] = [];
-  test.before((_) => {
+  const recordedLogs: { [workflowId: string]: LogEntry[] } = {};
+
+  test.before(async (_) => {
+    await registerDefaultCustomSearchAttributes(await Connection.connect({}));
     Runtime.install({
-      logger: new DefaultLogger('DEBUG', ({ level, message, meta }) => {
-        if (message === 'External sink function threw an error') recordedLogs.push({ level, message, meta });
+      logger: new DefaultLogger('DEBUG', (entry: LogEntry) => {
+        const workflowId = (entry.meta as any)?.workflowInfo?.workflowId;
+        recordedLogs[workflowId] ??= [];
+        recordedLogs[workflowId].push(entry);
       }),
     });
   });
 
-  // Must be serial because it uses the global Runtime to check for error messages
-  test.serial('Worker injects sinks', async (t) => {
+  test('Worker injects sinks', async (t) => {
     interface RecordedCall {
       info: WorkflowInfo;
       counter: number;
@@ -153,12 +157,13 @@ if (RUN_INTEGRATION_TESTS) {
     ]);
 
     t.deepEqual(
-      recordedLogs.map((x) => ({
+      recordedLogs[info.workflowId].map((x: LogEntry) => ({
         ...x,
         meta: {
           ...x.meta,
-          workflowInfo: fixWorkflowInfoDates(x.meta.workflowInfo),
+          workflowInfo: fixWorkflowInfoDates(x.meta?.workflowInfo),
         },
+        timestampNanos: undefined,
       })),
       thrownErrors.map((error) => ({
         level: 'ERROR',
@@ -169,6 +174,7 @@ if (RUN_INTEGRATION_TESTS) {
           fnName: error.fnName,
           workflowInfo: info,
         },
+        timestampNanos: undefined,
       }))
     );
   });
@@ -369,7 +375,7 @@ if (RUN_INTEGRATION_TESTS) {
     const taskQueue = `${__filename}-${t.title}`;
 
     const recordedMessages = Array<{ message: string; searchAttributes: SearchAttributes }>();
-    const sinks = asDefaultLoggerSink(async (info, message, _attrs) => {
+    const sinks = asSdkLoggerSink(async (info, message, _attrs) => {
       recordedMessages.push({
         message,
         searchAttributes: info.searchAttributes,

packages/test/src/test-sinks.ts

@@ -1,7 +1,16 @@
 import anyTest, { TestFn, ExecutionContext } from 'ava';
 import { v4 as uuid4 } from 'uuid';
-import { DefaultLogger, LogEntry, defaultSinks } from '@temporalio/worker';
+import {
+  DefaultLogger,
+  InjectedSinks,
+  LogEntry,
+  LogLevel,
+  LoggerSinks,
+  Runtime,
+  defaultSinks,
+} from '@temporalio/worker';
 import { TestWorkflowEnvironment } from '@temporalio/testing';
+import { WorkflowInfo } from '@temporalio/workflow';
 import * as workflows from './workflows';
 import { Worker } from './helpers';
 
@@ -11,9 +20,19 @@ interface Context {
 }
 const test = anyTest as TestFn<Context>;
 
+const recordedLogs: { [workflowId: string]: LogEntry[] } = {};
+
 test.before(async (t) => {
+  Runtime.install({
+    logger: new DefaultLogger('DEBUG', (entry) => {
+      const workflowId = (entry.meta as any)?.workflowInfo?.workflowId ?? (entry.meta as any)?.workflowId;
+      recordedLogs[workflowId] ??= [];
+      recordedLogs[workflowId].push(entry);
+    }),
+  });
+
   t.context = {
-    testEnv: await TestWorkflowEnvironment.createTimeSkipping(),
+    testEnv: await TestWorkflowEnvironment.createLocal(),
     taskQueue: '', // Will be set in beforeEach
   };
 });
@@ -26,27 +45,30 @@ test.after.always(async (t) => {
   await t.context.testEnv?.teardown();
 });
 
-async function withWorker(t: ExecutionContext<Context>, p: Promise<any>): Promise<[LogEntry, LogEntry]> {
+async function withWorker(
+  t: ExecutionContext<Context>,
+  p: Promise<any>,
+  workflowId: string
+): 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: t.context.taskQueue,
     workflowsPath: require.resolve('./workflows'),
-    sinks: defaultSinks(logger),
   });
   await worker.runUntil(p);
+  const logs = recordedLogs[workflowId];
   t.true(logs.length >= 2);
   return logs as [LogEntry, LogEntry];
 }
 
-test.serial('WorkflowInboundLogInterceptor logs when workflow completes', async (t) => {
+test('WorkflowInboundLogInterceptor logs when workflow completes', async (t) => {
   const { client } = t.context.testEnv;
   const workflowId = uuid4();
   const [startLog, endLog] = await withWorker(
     t,
-    client.workflow.execute(workflows.successString, { workflowId, taskQueue: t.context.taskQueue })
+    client.workflow.execute(workflows.successString, { workflowId, taskQueue: t.context.taskQueue }),
+    workflowId
   );
   t.is(startLog.level, 'DEBUG');
   t.is(startLog.message, 'Workflow started');
@@ -59,35 +81,121 @@ test.serial('WorkflowInboundLogInterceptor logs when workflow completes', async
   t.is(endLog.message, 'Workflow completed');
 });
 
-test.serial('WorkflowInboundLogInterceptor logs when workflow continues as new', async (t) => {
+test('WorkflowInboundLogInterceptor logs when workflow continues as new', async (t) => {
   const { client } = t.context.testEnv;
+  const workflowId = uuid4();
   const [_, endLog] = await withWorker(
     t,
     t.throwsAsync(
       client.workflow.execute(workflows.continueAsNewSameWorkflow, {
         args: ['execute', 'execute'],
-        workflowId: uuid4(),
+        workflowId,
         taskQueue: t.context.taskQueue,
         followRuns: false,
       })
-    )
+    ),
+    workflowId
   );
   t.is(endLog.level, 'DEBUG');
   t.is(endLog.message, 'Workflow continued as new');
 });
 
-test.serial('WorkflowInboundLogInterceptor logs warning when workflow fails', async (t) => {
+test('WorkflowInboundLogInterceptor logs warning when workflow fails', async (t) => {
   const { client } = t.context.testEnv;
+  const workflowId = uuid4();
   const [_, endLog] = await withWorker(
     t,
     t.throwsAsync(
       client.workflow.execute(workflows.throwAsync, {
-        workflowId: uuid4(),
+        workflowId,
         taskQueue: t.context.taskQueue,
         followRuns: false,
       })
-    )
+    ),
+    workflowId
   );
   t.is(endLog.level, 'WARN');
   t.is(endLog.message, 'Workflow failed');
 });
+
+test('(Legacy) defaultSinks(logger) can be used to customize where logs are sent', async (t) => {
+  const { client } = t.context.testEnv;
+  const workflowId = uuid4();
+  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: t.context.taskQueue,
+    workflowsPath: require.resolve('./workflows'),
+    // eslint-disable-next-line deprecation/deprecation
+    sinks: defaultSinks(logger),
+  });
+  await worker.runUntil(
+    client.workflow.execute(workflows.successString, { workflowId, taskQueue: t.context.taskQueue })
+  );
+  t.false(workflowId in recordedLogs);
+  t.true(logs.length >= 2);
+  const [startLog, endLog] = logs;
+  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, t.context.taskQueue);
+  t.is(startLog.meta?.namespace, 'default');
+  t.is(startLog.meta?.workflowType, 'successString');
+  t.is(endLog.level, 'DEBUG');
+  t.is(endLog.message, 'Workflow completed');
+});
+
+test('(Legacy) Can register defaultWorkerLogger sink to customize where logs are sent', async (t) => {
+  const { client } = t.context.testEnv;
+  const workflowId = uuid4();
+  const { nativeConnection } = t.context.testEnv;
+  const logs = Array<LogEntry>();
+  const fn = (level: LogLevel, _info: WorkflowInfo, message: string, attrs?: Record<string, unknown>) => {
+    logs.push({ level, message, meta: attrs, timestampNanos: 0n });
+  };
+  const worker = await Worker.create({
+    connection: nativeConnection,
+    taskQueue: t.context.taskQueue,
+    workflowsPath: require.resolve('./workflows'),
+    // eslint-disable-next-line deprecation/deprecation
+    sinks: <InjectedSinks<LoggerSinks>>{
+      defaultWorkerLogger: {
+        trace: { fn: fn.bind(undefined, 'TRACE') },
+        debug: { fn: fn.bind(undefined, 'DEBUG') },
+        info: { fn: fn.bind(undefined, 'INFO') },
+        warn: { fn: fn.bind(undefined, 'WARN') },
+        error: { fn: fn.bind(undefined, 'ERROR') },
+      },
+    },
+  });
+  await worker.runUntil(
+    client.workflow.execute(workflows.successString, { workflowId, taskQueue: t.context.taskQueue })
+  );
+  t.false(workflowId in recordedLogs);
+  t.true(logs.length >= 2);
+  const [startLog, endLog] = logs;
+  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, t.context.taskQueue);
+  t.is(startLog.meta?.namespace, 'default');
+  t.is(startLog.meta?.workflowType, 'successString');
+  t.is(endLog.level, 'DEBUG');
+  t.is(endLog.message, 'Workflow completed');
+});
+
+test('(Legacy) Can explicitly call defaultWorkerLogger sink to emit logs', async (t) => {
+  const { client } = t.context.testEnv;
+  const workflowId = uuid4();
+  const [_, midLog] = await withWorker(
+    t,
+    client.workflow.execute(workflows.useDepreatedLoggerSinkWorkflow, { workflowId, taskQueue: t.context.taskQueue }),
+    workflowId
+  );
+  t.is(midLog.level, 'INFO');
+  t.is(midLog.message, 'Log message from workflow');
+});

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

@@ -1568,7 +1568,7 @@ test('logAndTimeout', async (t) => {
   });
   t.deepEqual(calls, [
     {
-      ifaceName: 'defaultWorkerLogger',
+      ifaceName: '__temporal_logger',
       fnName: 'debug',
       args: [
         'Workflow started',
@@ -1582,7 +1582,7 @@ test('logAndTimeout', async (t) => {
       ],
     },
     {
-      ifaceName: 'defaultWorkerLogger',
+      ifaceName: '__temporal_logger',
       fnName: 'info',
       args: [
         'logging before getting stuck',

packages/test/src/test-workflows.ts

@@ -1,12 +1,12 @@
 import { proxySinks, WorkflowInterceptors, Sinks, sleep } from '@temporalio/workflow';
 
-export interface LoggerSinks extends Sinks {
+export interface MyLoggerSinks extends Sinks {
   logger: {
     log(event: string): void;
   };
 }
 
-const { logger } = proxySinks<LoggerSinks>();
+const { logger } = proxySinks<MyLoggerSinks>();
 
 export async function internalsInterceptorExample(): Promise<void> {
   await sleep(10);

packages/test/src/workflows/internals-interceptor-example.ts

@@ -34,3 +34,10 @@ export async function logSinkTester(): Promise<void> {
     }`
   );
 }
+
+// eslint-disable-next-line deprecation/deprecation
+const { defaultWorkerLogger } = wf.proxySinks<wf.LoggerSinks>();
+
+export async function useDepreatedLoggerSinkWorkflow(): Promise<void> {
+  defaultWorkerLogger.info(`Log message from workflow`, { workflowId: wf.workflowInfo().workflowId });
+}