feat: Workflow logger improvements (#1117)

- [Lazily initialize the default sinks
logger](540896e)
- [Export `log` from
workflow](5a27896)
- a logger that funnels messages to the default worker logger sink.

### Example:
```ts
import * as workflow from '@temporalio/workflow';

export async function myWorkflow(): Promise<void> {
  workflow.log.info('hello from my workflow', { key: 'value' });
}
```

Author: bergundy

packages/test/src/test-workflows.ts

@@ -14,6 +14,7 @@ import {
 } from '@temporalio/common';
 import { msToTs } from '@temporalio/common/lib/time';
 import { coresdk } from '@temporalio/proto';
+import { LogTimestamp } from '@temporalio/worker';
 import { WorkflowCodeBundler } from '@temporalio/worker/lib/workflow/bundler';
 import { VMWorkflow, VMWorkflowCreator } from '@temporalio/worker/lib/workflow/vm';
 import { ReusableVMWorkflow, ReusableVMWorkflowCreator } from '@temporalio/worker/lib/workflow/reusable-vm';
@@ -1532,6 +1533,7 @@ test('logAndTimeout', async (t) => {
     message: 'Script execution timed out after 200ms',
   });
   const calls = await workflow.getAndResetSinkCalls();
+  delete calls[0].args[1][LogTimestamp];
   t.deepEqual(calls, [
     {
       ifaceName: 'defaultWorkerLogger',

packages/worker/src/logger.ts

@@ -25,7 +25,7 @@ export interface Logger {
 
 export { LogLevel };
 
-export const LogTimestamp = Symbol('log_timestamp');
+export const LogTimestamp = Symbol.for('log_timestamp');
 
 const severities: LogLevel[] = ['TRACE', 'DEBUG', 'INFO', 'WARN', 'ERROR'];
 

packages/worker/src/worker-options.ts

@@ -8,6 +8,7 @@ import { LoggerSinks } from '@temporalio/workflow';
 import { ActivityInboundLogInterceptor } from './activity-log-interceptor';
 import { NativeConnection } from './connection';
 import { WorkerInterceptors } from './interceptors';
+import { Logger } from './logger';
 import { Runtime } from './runtime';
 import { InjectedSinks } from './sinks';
 import { GiB } from './utils';
@@ -431,31 +432,36 @@ export interface ReplayWorkerOptions
  *
  * @param logger a {@link Logger} - defaults to the {@link Runtime} singleton logger.
  */
-export function defaultSinks(logger = Runtime.instance().logger): InjectedSinks<LoggerSinks> {
+export function defaultSinks(logger?: Logger): InjectedSinks<LoggerSinks> {
   return {
     defaultWorkerLogger: {
       trace: {
         fn(_, message, attrs) {
+          logger ??= Runtime.instance().logger;
           logger.trace(message, attrs);
         },
       },
       debug: {
         fn(_, message, attrs) {
+          logger ??= Runtime.instance().logger;
           logger.debug(message, attrs);
         },
       },
       info: {
         fn(_, message, attrs) {
+          logger ??= Runtime.instance().logger;
           logger.info(message, attrs);
         },
       },
       warn: {
         fn(_, message, attrs) {
+          logger ??= Runtime.instance().logger;
           logger.warn(message, attrs);
         },
       },
       error: {
         fn(_, message, attrs) {
+          logger ??= Runtime.instance().logger;
           logger.error(message, attrs);
         },
       },

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

@@ -1,13 +1,12 @@
 import {
   isCancellation,
-  LoggerSinks,
   Next,
-  proxySinks,
   WorkflowExecuteInput,
   WorkflowInboundCallsInterceptor,
   workflowInfo,
   WorkflowInfo,
   WorkflowInterceptorsFactory,
+  log,
 } from '@temporalio/workflow';
 import { untrackPromise } from '@temporalio/workflow/lib/stack-helpers';
 
@@ -31,27 +30,25 @@ export class WorkflowInboundLogInterceptor implements WorkflowInboundCallsInterc
   }
 
   execute(input: WorkflowExecuteInput, next: Next<WorkflowInboundCallsInterceptor, 'execute'>): Promise<unknown> {
-    const { defaultWorkerLogger: logger } = proxySinks<LoggerSinks>();
-
-    logger.debug('Workflow started', this.logAttributes());
+    log.debug('Workflow started', this.logAttributes());
     const p = next(input).then(
       (res) => {
-        logger.debug('Workflow completed', this.logAttributes());
+        log.debug('Workflow completed', this.logAttributes());
         return res;
       },
       (error) => {
         // Avoid using instanceof checks in case the modules they're defined in loaded more than once,
         // e.g. by jest or when multiple versions are installed.
         if (typeof error === 'object' && error != null) {
           if (isCancellation(error)) {
-            logger.debug('Workflow completed as cancelled', this.logAttributes());
+            log.debug('Workflow completed as cancelled', this.logAttributes());
             throw error;
           } else if (error.name === 'ContinueAsNew') {
-            logger.debug('Workflow continued as new', this.logAttributes());
+            log.debug('Workflow continued as new', this.logAttributes());
             throw error;
           }
         }
-        logger.warn('Workflow failed', { error, ...this.logAttributes() });
+        log.warn('Workflow failed', { error, ...this.logAttributes() });
         throw error;
       }
     );

packages/worker/src/workflow/reusable-vm.ts

@@ -3,6 +3,8 @@ import { AsyncLocalStorage } from 'node:async_hooks';
 import vm from 'node:vm';
 import * as internals from '@temporalio/workflow/lib/worker-interface';
 import { IllegalStateError } from '@temporalio/common';
+import { getTimeOfDay } from '@temporalio/core-bridge';
+import { timeOfDayToBigint } from '../logger';
 import { Workflow, WorkflowCreateOptions, WorkflowCreator } from './interface';
 import { WorkflowBundleWithSourceMapAndFilename } from './workflow-worker-thread/input';
 import { BaseVMWorkflow, globalHandlers, injectConsole, setUnhandledRejectionHandler } from './vm-shared';
@@ -160,7 +162,11 @@ export class ReusableVMWorkflowCreator implements WorkflowCreator {
       }
     ) as any;
 
-    workflowModule.initRuntime({ ...options, sourceMap: this.workflowBundle.sourceMap });
+    workflowModule.initRuntime({
+      ...options,
+      sourceMap: this.workflowBundle.sourceMap,
+      getTimeOfDay: () => timeOfDayToBigint(getTimeOfDay()),
+    });
     const activator = bag.__TEMPORAL_ACTIVATOR__ as any;
 
     const newVM = new ReusableVMWorkflow(options.info, context, activator, workflowModule, isolateExecutionTimeoutMs);

packages/worker/src/workflow/vm.ts

@@ -2,6 +2,8 @@ import assert from 'node:assert';
 import { AsyncLocalStorage } from 'node:async_hooks';
 import vm from 'node:vm';
 import { IllegalStateError } from '@temporalio/common';
+import { getTimeOfDay } from '@temporalio/core-bridge';
+import { timeOfDayToBigint } from '../logger';
 import { Workflow, WorkflowCreateOptions, WorkflowCreator } from './interface';
 import { WorkflowBundleWithSourceMapAndFilename } from './workflow-worker-thread/input';
 import {
@@ -56,7 +58,11 @@ export class VMWorkflowCreator implements WorkflowCreator {
       }
     ) as any;
 
-    workflowModule.initRuntime({ ...options, sourceMap: this.workflowBundle.sourceMap });
+    workflowModule.initRuntime({
+      ...options,
+      sourceMap: this.workflowBundle.sourceMap,
+      getTimeOfDay: () => timeOfDayToBigint(getTimeOfDay()),
+    });
     const activator = context.__TEMPORAL_ACTIVATOR__ as any;
 
     const newVM = new VMWorkflow(options.info, context, activator, workflowModule, isolateExecutionTimeoutMs);

packages/workflow/src/interfaces.ts

@@ -384,8 +384,9 @@ export interface WorkflowCreateOptions {
   showStackTraceSources: boolean;
 }
 
-export interface WorkflowCreateOptionsWithSourceMap extends WorkflowCreateOptions {
+export interface WorkflowCreateOptionsInternal extends WorkflowCreateOptions {
   sourceMap: RawSourceMap;
+  getTimeOfDay(): bigint;
 }
 
 /**

packages/workflow/src/internals.ts

@@ -29,7 +29,7 @@ import {
   EnhancedStackTrace,
   FileLocation,
   WorkflowInfo,
-  WorkflowCreateOptionsWithSourceMap,
+  WorkflowCreateOptionsInternal,
 } from './interfaces';
 import { SinkCall } from './sinks';
 import { untrackPromise } from './stack-helpers';
@@ -271,16 +271,26 @@ export class Activator implements ActivationHandler {
    */
   public readonly sentPatches = new Set<string>();
 
+  /**
+   * Buffered sink calls per activation
+   */
   sinkCalls = Array<SinkCall>();
 
+  /**
+   * A nanosecond resolution time function, externally injected
+   */
+  public readonly getTimeOfDay: () => bigint;
+
   constructor({
     info,
     now,
     showStackTraceSources,
     sourceMap,
+    getTimeOfDay,
     randomnessSeed,
     patches,
-  }: WorkflowCreateOptionsWithSourceMap) {
+  }: WorkflowCreateOptionsInternal) {
+    this.getTimeOfDay = getTimeOfDay;
     this.info = info;
     this.now = now;
     this.showStackTraceSources = showStackTraceSources;

packages/workflow/src/worker-interface.ts

@@ -10,7 +10,7 @@ import type { coresdk } from '@temporalio/proto';
 import { disableStorage } from './cancellation-scope';
 import { DeterminismViolationError } from './errors';
 import { WorkflowInterceptorsFactory } from './interceptors';
-import { WorkflowCreateOptionsWithSourceMap, WorkflowInfo } from './interfaces';
+import { WorkflowCreateOptionsInternal, WorkflowInfo } from './interfaces';
 import { Activator, getActivator } from './internals';
 import { SinkCall } from './sinks';
 import { setActivatorUntyped } from './global-attributes';
@@ -92,7 +92,7 @@ export function overrideGlobals(): void {
  *
  * Sets required internal state and instantiates the workflow and interceptors.
  */
-export function initRuntime(options: WorkflowCreateOptionsWithSourceMap): void {
+export function initRuntime(options: WorkflowCreateOptionsInternal): void {
   const info: WorkflowInfo = fixPrototypes(options.info);
   info.unsafe.now = OriginalDate.now;
   const activator = new Activator({ ...options, info });

packages/workflow/src/workflow.ts

@@ -38,7 +38,7 @@ import {
   WorkflowInfo,
 } from './interfaces';
 import { LocalActivityDoBackoff, getActivator, maybeGetActivator } from './internals';
-import { Sinks } from './sinks';
+import { LoggerSinks, Sinks } from './sinks';
 import { untrackPromise } from './stack-helpers';
 import { ChildWorkflowHandle, ExternalWorkflowHandle } from './workflow-handle';
 
@@ -1253,3 +1253,40 @@ export function upsertSearchAttributes(searchAttributes: SearchAttributes): void
 
 export const stackTraceQuery = defineQuery<string>('__stack_trace');
 export const enhancedStackTraceQuery = defineQuery<EnhancedStackTrace>('__enhanced_stack_trace');
+
+const loggerSinks = proxySinks<LoggerSinks>();
+
+/**
+ * Symbol used by the SDK logger to extract a timestamp from log attributes.
+ * Also defined in `worker/logger.ts` - intentionally not shared.
+ */
+const LogTimestamp = Symbol.for('log_timestamp');
+
+/**
+ * Default workflow logger.
+ * This logger is replay-aware and will omit log messages on workflow replay.
+ * The messages emitted by this logger are funnelled to the worker's `defaultSinks`, which are installed by default.
+ *
+ * Note that since sinks are used to power this logger, any log attributes must be transferable via the
+ * {@link https://nodejs.org/api/worker_threads.html#worker_threads_port_postmessage_value_transferlist | postMessage}
+ * API.
+ *
+ * `defaultSinks` accepts a user logger and defaults to the `Runtime`'s logger.
+ *
+ * See the documentation for `WorkerOptions`, `defaultSinks`, and `Runtime` for more information.
+ */
+export const log: LoggerSinks['defaultWorkerLogger'] = Object.fromEntries(
+  (['trace', 'debug', 'info', 'warn', 'error'] as Array<keyof LoggerSinks['defaultWorkerLogger']>).map((level) => {
+    return [
+      level,
+      (message: string, attrs: Record<string, unknown>) => {
+        return loggerSinks.defaultWorkerLogger[level](message, {
+          // Inject the call time in nanosecond resolution as expected by the worker logger.
+          [LogTimestamp]: getActivator().getTimeOfDay(),
+          // Only available from node 17.
+          ...((globalThis as any).structuredClone ? (globalThis as any).structuredClone(attrs) : attrs),
+        });
+      },
+    ];
+  })
+) as any;