feat(logging): enhance structured logging support across client and worker

- Added structured logging capabilities to ConsoleLogger and NoOpLogger.
- Implemented logEvent method in ConsoleLogger and NoOpLogger to handle structured log events.
- Introduced emitLog utility to centralize logging logic for structured and non-structured loggers.
- Created centralized log definitions for Durable Task Client and Worker, aligning with .NET SDK event IDs and message templates.
- Added various logging functions for orchestration and activity lifecycle events in the worker.
- Implemented error handling utilities for safe error message extraction.
- Updated tests to cover new structured logging functionality and ensure proper logging behavior.

Author: torosent

examples/azure-managed/distributed-tracing.ts

@@ -45,8 +45,8 @@ console.log(`OpenTelemetry SDK started – exporting traces to ${otlpEndpoint}`)
 import {
   DurableTaskAzureManagedClientBuilder,
   DurableTaskAzureManagedWorkerBuilder,
-  createAzureLogger,
 } from "@microsoft/durabletask-js-azuremanaged";
+import { ConsoleLogger } from "@microsoft/durabletask-js";
 import { ActivityContext } from "@microsoft/durabletask-js/dist/task/context/activity-context";
 import { OrchestrationContext } from "@microsoft/durabletask-js/dist/task/context/orchestration-context";
 import { TOrchestrator } from "@microsoft/durabletask-js/dist/types/orchestrator.type";
@@ -56,10 +56,11 @@ import { whenAll } from "@microsoft/durabletask-js/dist/task";
 // 3. Application code
 // --------------------------------------------------------------------------
 (async () => {
-  // Use the Azure logger adapter for the SDK internals – it respects AZURE_LOG_LEVEL
-  // and suppresses debug/verbose messages by default, keeping the console output clean.
-  // Set AZURE_LOG_LEVEL=verbose to see internal SDK logs.
-  const sdkLogger = createAzureLogger();
+  // Use ConsoleLogger so structured log events (with event IDs and categories) are
+  // printed to the console by default, similar to .NET's default ILogger<T> output.
+  // For production, consider using createAzureLogger() which integrates with @azure/logger
+  // and respects the AZURE_LOG_LEVEL environment variable.
+  const sdkLogger = new ConsoleLogger();
 
   // --- Configuration ---
   const connectionString = process.env.DURABLE_TASK_SCHEDULER_CONNECTION_STRING;
@@ -204,6 +205,9 @@ import { whenAll } from "@microsoft/durabletask-js/dist/task";
     // --- Start worker ---
     console.log("Starting worker...");
     await worker.start();
+    // Allow the worker time to establish the gRPC stream with the scheduler.
+    // worker.start() returns before the connection is fully established.
+    await new Promise((r) => setTimeout(r, 2000));
     console.log("Worker started.");
 
     // --- Run orchestrations ---
@@ -212,14 +216,14 @@ import { whenAll } from "@microsoft/durabletask-js/dist/task";
     console.log("\n=== Sequence Orchestration ===");
     const seqId = await client.scheduleNewOrchestration(sequenceOrchestrator);
     console.log(`Scheduled: ${seqId}`);
-    const seqState = await client.waitForOrchestrationCompletion(seqId, undefined, 60);
+    const seqState = await client.waitForOrchestrationCompletion(seqId, true, 60);
     console.log(`Completed – result: ${seqState?.serializedOutput}`);
 
     // 2) Data pipeline orchestration (fan-out/fan-in)
     console.log("\n=== Data Pipeline Orchestration ===");
     const pipelineId = await client.scheduleNewOrchestration(dataPipelineOrchestrator);
     console.log(`Scheduled: ${pipelineId}`);
-    const pipelineState = await client.waitForOrchestrationCompletion(pipelineId, undefined, 60);
+    const pipelineState = await client.waitForOrchestrationCompletion(pipelineId, true, 60);
     console.log(`Completed – result: ${pipelineState?.serializedOutput}`);
 
     console.log("\n=== All orchestrations completed! ===");

examples/azure-managed/index.ts

@@ -128,14 +128,17 @@ import { whenAll } from "@microsoft/durabletask-js/dist/task";
     // Start the worker
     logger.info("Starting worker...");
     await worker.start();
+    // Allow the worker time to establish the gRPC stream with the scheduler.
+    // worker.start() returns before the connection is fully established.
+    await new Promise((resolve) => setTimeout(resolve, 2000));
     logger.info("Worker started successfully!");
 
     // Run the sequence orchestrator
     logger.info("\n--- Running Sequence Orchestrator ---");
     const sequenceId = await client.scheduleNewOrchestration(sequenceOrchestrator);
     logger.info(`Orchestration scheduled with ID: ${sequenceId}`);
 
-    const sequenceState = await client.waitForOrchestrationCompletion(sequenceId, undefined, 60);
+    const sequenceState = await client.waitForOrchestrationCompletion(sequenceId, true, 60);
     logger.info(`Sequence orchestration completed!`);
     logger.info(`Result: ${sequenceState?.serializedOutput}`);
 
@@ -144,7 +147,7 @@ import { whenAll } from "@microsoft/durabletask-js/dist/task";
     const fanOutId = await client.scheduleNewOrchestration(fanOutFanInOrchestrator);
     logger.info(`Orchestration scheduled with ID: ${fanOutId}`);
 
-    const fanOutState = await client.waitForOrchestrationCompletion(fanOutId, undefined, 60);
+    const fanOutState = await client.waitForOrchestrationCompletion(fanOutId, true, 60);
     logger.info(`Fan-out/fan-in orchestration completed!`);
     logger.info(`Result: ${fanOutState?.serializedOutput}`);
 

packages/durabletask-js-azuremanaged/src/azure-logger-adapter.ts

@@ -2,7 +2,7 @@
 // Licensed under the MIT License.
 
 import { createClientLogger, AzureLogger } from "@azure/logger";
-import { Logger } from "@microsoft/durabletask-js";
+import { StructuredLogger, createLogEventHandler } from "@microsoft/durabletask-js";
 
 /**
  * Pre-configured logger adapter that uses the default "durabletask" namespace.
@@ -29,19 +29,22 @@ import { Logger } from "@microsoft/durabletask-js";
  *   .build();
  * ```
  */
-export const AzureLoggerAdapter: Logger = createAzureLogger();
+export const AzureLoggerAdapter: StructuredLogger = createAzureLogger();
 
 /**
- * Creates a Logger instance that integrates with Azure SDK's logging infrastructure.
+ * Creates a StructuredLogger instance that integrates with Azure SDK's logging infrastructure.
  * 
  * The created logger uses `@azure/logger` under the hood, which means:
  * - Log output can be controlled via the `AZURE_LOG_LEVEL` environment variable
  * - Log output can be redirected using `setLogLevel()` from `@azure/logger`
  * - Logs are prefixed with the namespace for easy filtering
  * 
+ * When structured log events are emitted, the event ID and category are included in the
+ * log message prefix for easy filtering and correlation.
+ * 
  * @param namespace - Optional sub-namespace to append to "durabletask".
  *                    For example, "client" results in "durabletask:client".
- * @returns A Logger instance configured for the specified namespace.
+ * @returns A StructuredLogger instance configured for the specified namespace.
  * 
  * @example
  * ```typescript
@@ -58,7 +61,7 @@ export const AzureLoggerAdapter: Logger = createAzureLogger();
  * // Logs will be prefixed with "durabletask"
  * ```
  */
-export function createAzureLogger(namespace?: string): Logger {
+export function createAzureLogger(namespace?: string): StructuredLogger {
   const fullNamespace = namespace ? `durabletask:${namespace}` : "durabletask";
   const azureLogger: AzureLogger = createClientLogger(fullNamespace);
 
@@ -75,5 +78,11 @@ export function createAzureLogger(namespace?: string): Logger {
     debug: (message: string, ...args: unknown[]): void => {
       azureLogger.verbose(message, ...args);
     },
+    logEvent: createLogEventHandler({
+      error: azureLogger.error.bind(azureLogger),
+      warn: azureLogger.warning.bind(azureLogger),
+      info: azureLogger.info.bind(azureLogger),
+      debug: azureLogger.verbose.bind(azureLogger),
+    }),
   };
 }

packages/durabletask-js/src/client/client.ts

@@ -29,6 +29,7 @@ import { Logger, ConsoleLogger } from "../types/logger.type";
 import { StartOrchestrationOptions } from "../task/options";
 import { mapToRecord } from "../utils/tags.util";
 import { populateTagsMap } from "../utils/pb-helper.util";
+import * as ClientLogs from "./logs";
 import {
   startSpanForNewOrchestration,
   startSpanForEventRaisedFromClient,
@@ -228,7 +229,9 @@ export class TaskHubGrpcClient {
     // Create a tracing span for the new orchestration (if OTEL is available)
     const span = startSpanForNewOrchestration(req);
 
-    this._logger.info(`Starting new ${name} instance with ID = ${req.getInstanceid()}${effectiveVersion ? ` (version: ${effectiveVersion})` : ""}`);
+    const serializedInput = i.getValue() ?? "";
+    const inputSizeInBytes = Buffer.byteLength(serializedInput, "utf8");
+    ClientLogs.schedulingOrchestration(this._logger, req.getInstanceid(), name, inputSizeInBytes);
 
     try {
       const res = await callWithMetadata<pb.CreateInstanceRequest, pb.CreateInstanceResponse>(
@@ -340,7 +343,7 @@ export class TaskHubGrpcClient {
     req.setInstanceid(instanceId);
     req.setGetinputsandoutputs(fetchPayloads);
 
-    this._logger.info(`Waiting ${timeout} seconds for instance ${instanceId} to complete...`);
+    ClientLogs.waitingForInstanceCompletion(this._logger, instanceId);
 
     const callPromise = callWithMetadata<pb.GetInstanceRequest, pb.GetInstanceResponse>(
       this._stub.waitForInstanceCompletion.bind(this._stub),
@@ -364,11 +367,11 @@ export class TaskHubGrpcClient {
 
     if (state.runtimeStatus === OrchestrationStatus.FAILED && state.failureDetails) {
       details = state.failureDetails;
-      this._logger.info(`Instance ${instanceId} failed: [${details.errorType}] ${details.message}`);
+      ClientLogs.instanceFailed(this._logger, instanceId, details.errorType, details.message);
     } else if (state.runtimeStatus === OrchestrationStatus.TERMINATED) {
-      this._logger.info(`Instance ${instanceId} was terminated`);
+      ClientLogs.instanceTerminated(this._logger, instanceId);
     } else if (state.runtimeStatus === OrchestrationStatus.COMPLETED) {
-      this._logger.info(`Instance ${instanceId} completed`);
+      ClientLogs.instanceCompleted(this._logger, instanceId);
     }
 
     return state;
@@ -397,7 +400,7 @@ export class TaskHubGrpcClient {
     // Create a tracing span for the raised event (if OTEL is available)
     const span = startSpanForEventRaisedFromClient(eventName, instanceId);
 
-    this._logger.info(`Raising event '${eventName}' for instance '${instanceId}'`);
+    ClientLogs.raisingEvent(this._logger, instanceId, eventName);
 
     try {
       await callWithMetadata<pb.RaiseEventRequest, pb.RaiseEventResponse>(
@@ -461,7 +464,7 @@ export class TaskHubGrpcClient {
     req.setOutput(i);
     req.setRecursive(recursive);
 
-    this._logger.info(`Terminating '${instanceId}'${recursive ? ' (recursive)' : ''}`);
+    ClientLogs.terminatingInstance(this._logger, instanceId);
 
     await callWithMetadata<pb.TerminateRequest, pb.TerminateResponse>(
       this._stub.terminateInstance.bind(this._stub),
@@ -474,7 +477,7 @@ export class TaskHubGrpcClient {
     const req = new pb.SuspendRequest();
     req.setInstanceid(instanceId);
 
-    this._logger.info(`Suspending '${instanceId}'`);
+    ClientLogs.suspendingInstance(this._logger, instanceId);
 
     await callWithMetadata<pb.SuspendRequest, pb.SuspendResponse>(
       this._stub.suspendInstance.bind(this._stub),
@@ -487,7 +490,7 @@ export class TaskHubGrpcClient {
     const req = new pb.ResumeRequest();
     req.setInstanceid(instanceId);
 
-    this._logger.info(`Resuming '${instanceId}'`);
+    ClientLogs.resumingInstance(this._logger, instanceId);
 
     await callWithMetadata<pb.ResumeRequest, pb.ResumeResponse>(
       this._stub.resumeInstance.bind(this._stub),
@@ -525,7 +528,7 @@ export class TaskHubGrpcClient {
       req.setReason(reasonValue);
     }
 
-    this._logger.info(`Rewinding '${instanceId}' with reason: ${reason}`);
+    ClientLogs.rewindingInstance(this._logger, instanceId, reason);
 
     try {
       await callWithMetadata<pb.RewindInstanceRequest, pb.RewindInstanceResponse>(
@@ -582,7 +585,7 @@ export class TaskHubGrpcClient {
     req.setInstanceid(instanceId);
     req.setRestartwithnewinstanceid(restartWithNewInstanceId);
 
-    this._logger.info(`Restarting '${instanceId}' with restartWithNewInstanceId=${restartWithNewInstanceId}`);
+    ClientLogs.restartingInstance(this._logger, instanceId, restartWithNewInstanceId);
 
     try {
       const res = await callWithMetadata<pb.RestartInstanceRequest, pb.RestartInstanceResponse>(
@@ -637,7 +640,7 @@ export class TaskHubGrpcClient {
       req.setInstanceid(instanceId);
       req.setRecursive(options?.recursive ?? false);
 
-      this._logger.info(`Purging Instance '${instanceId}'${options?.recursive ? ' (recursive)' : ''}`);
+      ClientLogs.purgingInstanceMetadata(this._logger, instanceId);
 
       res = await callWithMetadata<pb.PurgeInstancesRequest, pb.PurgeInstancesResponse>(
         this._stub.purgeInstances.bind(this._stub),
@@ -668,7 +671,7 @@ export class TaskHubGrpcClient {
       req.setRecursive(options?.recursive ?? false);
       const timeout = purgeInstanceCriteria.getTimeout();
 
-      this._logger.info(`Purging Instances using purging criteria${options?.recursive ? " (recursive)" : ""}`);
+      ClientLogs.purgingInstances(this._logger, createdTimeFrom, createdTimeTo, runtimeStatusList.map(String).join(", "));
 
       const callPromise = callWithMetadata<pb.PurgeInstancesRequest, pb.PurgeInstancesResponse>(
         this._stub.purgeInstances.bind(this._stub),