[DRAFT] Add context propagation + custom log attributes sample

.scripts/list-of-samples.json

@@ -4,6 +4,7 @@
     "activities-dependency-injection",
     "activities-examples",
     "child-workflows",
+    "context-propagation",
     "continue-as-new",
     "cron-workflows",
     "custom-logger",

README.md

@@ -143,6 +143,7 @@ and you'll be given the list of sample options.
 
 - **Interceptors**:
   - [**OpenTelemetry**](./interceptors-opentelemetry): Use the Interceptors feature to add OpenTelemetry metrics reporting to your workflows.
+  - [**Context Propagation**](./context-propagation): Uses SDK interceptors to propagate contextual data from the client, to workflows, child workflows and activities, and automatically include these as metadata emitted log messages.
   - [**Query Subscriptions**](./query-subscriptions): Use Redis Streams, Immer, and SDK Interceptors to subscribe to Workflow state.
 - [**gRPC calls**](./grpc-calls): Make raw gRPC calls for advanced queries not covered by the WorkflowClient API.
 

context-propagation/.eslintignore

@@ -0,0 +1,3 @@
+node_modules
+lib
+.eslintrc.js

context-propagation/.eslintrc.js

@@ -0,0 +1,48 @@
+const { builtinModules } = require('module');
+
+const ALLOWED_NODE_BUILTINS = new Set(['assert']);
+
+module.exports = {
+  root: true,
+  parser: '@typescript-eslint/parser',
+  parserOptions: {
+    project: './tsconfig.json',
+    tsconfigRootDir: __dirname,
+  },
+  plugins: ['@typescript-eslint', 'deprecation'],
+  extends: [
+    'eslint:recommended',
+    'plugin:@typescript-eslint/eslint-recommended',
+    'plugin:@typescript-eslint/recommended',
+    'prettier',
+  ],
+  rules: {
+    // recommended for safety
+    '@typescript-eslint/no-floating-promises': 'error', // forgetting to await Activities and Workflow APIs is bad
+    'deprecation/deprecation': 'warn',
+
+    // code style preference
+    'object-shorthand': ['error', 'always'],
+
+    // relaxed rules, for convenience
+    '@typescript-eslint/no-unused-vars': [
+      'warn',
+      {
+        argsIgnorePattern: '^_',
+        varsIgnorePattern: '^_',
+      },
+    ],
+    '@typescript-eslint/no-explicit-any': 'off',
+  },
+  overrides: [
+    {
+      files: ['src/workflows.ts', 'src/workflows-*.ts', 'src/workflows/*.ts'],
+      rules: {
+        'no-restricted-imports': [
+          'error',
+          ...builtinModules.filter((m) => !ALLOWED_NODE_BUILTINS.has(m)).flatMap((m) => [m, `node:${m}`]),
+        ],
+      },
+    },
+  ],
+};

context-propagation/.gitignore

@@ -0,0 +1,2 @@
+lib
+node_modules

context-propagation/.npmrc

@@ -0,0 +1 @@
+package-lock=false

context-propagation/.nvmrc

@@ -0,0 +1 @@
+16

context-propagation/.post-create

@@ -0,0 +1,18 @@
+To begin development, install the Temporal CLI:
+
+  Mac: {cyan brew install temporal}
+  Other: Download and extract the latest release from https://github.com/temporalio/cli/releases/latest
+
+Start Temporal Server:
+
+  {cyan temporal server start-dev}
+
+Use Node version 16+:
+
+  Mac: {cyan brew install node@16}
+  Other: https://nodejs.org/en/download/
+
+Then, in the project directory, using two other shells, run these commands:
+
+  {cyan npm run start.watch}
+  {cyan npm run workflow}

context-propagation/.prettierignore

@@ -0,0 +1 @@
+lib

context-propagation/.prettierrc

@@ -0,0 +1,2 @@
+printWidth: 120
+singleQuote: true

context-propagation/README.md

@@ -0,0 +1,175 @@
+# Context Propagation demo
+
+This project demonstrate an advanced use case where interceptors are used to propagate some contextual data from client to workflow, to child workflow, and to activities.
+
+In this demo, that contextual data is inject custom log
+attributes on log entries produced from workflow and activities.
+
+In particular, this sample demonstrate:
+
+- Using
+  attributes,
+
+### Running this sample
+
+1. `temporal server start-dev` to start [Temporal Server](https://github.com/temporalio/cli/#installation).
+1. `npm install` to install dependencies.
+1. `npm run start.watch` to start the Worker.
+1. In another shell, `npm run workflow` to run the Workflow.
+
+<details>
+<summary>
+Sample worker output
+</summary>
+
+```
+2023-09-05T18:19:39.646Z [worker] info: Creating worker {
+  options: {
+    namespace: 'default',
+    identity: '1234@MyComputer',
+    useVersioning: false,
+    buildId: undefined,
+    shutdownGraceTime: 0,
+    maxConcurrentLocalActivityExecutions: 100,
+    enableNonLocalActivities: true,
+    maxConcurrentWorkflowTaskPolls: 10,
+    maxConcurrentActivityTaskPolls: 10,
+    stickyQueueScheduleToStartTimeout: '10s',
+    maxHeartbeatThrottleInterval: '60s',
+    defaultHeartbeatThrottleInterval: '30s',
+    isolateExecutionTimeout: '5s',
+    workflowThreadPoolSize: 2,
+    maxCachedWorkflows: 914,
+    showStackTraceSources: false,
+    debugMode: false,
+    workflowsPath: '/Users/user/samples-typescript/custom-logger/src/workflows/index.ts',
+    activities: { greet: [AsyncFunction: greet] },
+    taskQueue: 'custom-logger',
+    maxConcurrentWorkflowTaskExecutions: 40,
+    maxConcurrentActivityTaskExecutions: 100,
+    shutdownGraceTimeMs: 0,
+    shutdownForceTimeMs: undefined,
+    stickyQueueScheduleToStartTimeoutMs: 10000,
+    isolateExecutionTimeoutMs: 5000,
+    maxHeartbeatThrottleIntervalMs: 60000,
+    defaultHeartbeatThrottleIntervalMs: 30000,
+    loadedDataConverter: {
+      payloadConverter: DefaultPayloadConverter {
+        converterByEncoding: Map(3) {
+          'binary/null' => [UndefinedPayloadConverter],
+          'binary/plain' => [BinaryPayloadConverter],
+          'json/plain' => [JsonPayloadConverter]
+        },
+        converters: [
+          [UndefinedPayloadConverter],
+          [BinaryPayloadConverter],
+          [JsonPayloadConverter]
+        ]
+      },
+      failureConverter: DefaultFailureConverter {
+        options: { encodeCommonAttributes: false }
+      },
+      payloadCodecs: []
+    }
+  }
+}
+2023-09-05T18:19:40.084Z [worker] info: asset workflow-bundle-95e3c04c487ab5112957.js 755 KiB [emitted] [immutable] (name: main)
+2023-09-05T18:19:40.084Z [worker] info: runtime modules 937 bytes 4 modules
+2023-09-05T18:19:40.084Z [worker] info: modules by path ./node_modules/@temporalio/ 182 KiB
+2023-09-05T18:19:40.084Z [worker] info:   modules by path ./node_modules/@temporalio/common/lib/ 77.9 KiB 21 modules
+2023-09-05T18:19:40.084Z [worker] info:   modules by path ./node_modules/@temporalio/workflow/ 102 KiB
+2023-09-05T18:19:40.084Z [worker] info:     ./node_modules/@temporalio/workflow/lib/worker-interface.js 11 KiB [built] [code generated]
+2023-09-05T18:19:40.084Z [worker] info:     + 13 modules
+2023-09-05T18:19:40.084Z [worker] info:   ./node_modules/@temporalio/worker/lib/workflow-log-interceptor.js 2.42 KiB [built] [code generated]
+2023-09-05T18:19:40.084Z [worker] info: modules by path ./src/workflows/ 878 bytes
+2023-09-05T18:19:40.084Z [worker] info:   ./src/workflows/index-autogenerated-entrypoint.cjs 597 bytes [built] [code generated]
+2023-09-05T18:19:40.084Z [worker] info:   ./src/workflows/index.ts 281 bytes [built] [code generated]
+2023-09-05T18:19:40.084Z [worker] info: __temporal_custom_payload_converter (ignored) 15 bytes [built] [code generated]
+2023-09-05T18:19:40.084Z [worker] info: __temporal_custom_failure_converter (ignored) 15 bytes [built] [code generated]
+2023-09-05T18:19:40.084Z [worker] info: ./node_modules/long/umd/index.js 43.1 KiB [built] [code generated]
+2023-09-05T18:19:40.084Z [worker] info: ./node_modules/ms/dist/index.cjs 3.41 KiB [built] [code generated]
+2023-09-05T18:19:40.084Z [worker] info: webpack 5.88.2 compiled successfully in 264 ms
+2023-09-05T18:19:40.085Z [worker] info: Workflow bundle created { size: '0.74MB' }
+2023-09-05T18:19:40.165Z [worker] info: Initializing worker
+2023-09-05T18:19:40.166Z [worker] info: Worker state changed { state: 'RUNNING' }
+2023-09-05T18:19:51.963Z [worker] debug: New WFT
+2023-09-05T18:19:51.963Z [worker] debug: Applying new workflow task from server
+2023-09-05T18:19:51.963Z [worker] debug: Driven WF start
+2023-09-05T18:19:51.963Z [worker] debug: Sending activation to lang
+2023-09-05T18:19:51.987Z [workflow] debug: Workflow started {
+  namespace: 'default',
+  taskQueue: 'custom-logger',
+  workflowId: 'workflow-2YriOlmgK2vMT-rhRFQQp',
+  runId: 'fa0b415b-ef89-434d-839c-ec031d31668e',
+  workflowType: 'logSampleWorkflow'
+}
+2023-09-05T18:19:51.987Z [worker] debug: wf bridge iteration fetch
+2023-09-05T18:19:51.987Z [worker] debug: handling command
+2023-09-05T18:19:51.987Z [worker] debug: prepared commands
+2023-09-05T18:19:51.987Z [worker] debug: Sending responses to server
+2023-09-05T18:19:51.989Z [worker] debug: Server returned 1 fewer activities for eager execution than we requested
+2023-09-05T18:19:51.989Z [worker] debug: Marking WFT completed
+2023-09-05T18:19:51.994Z [workflow] debug: Activity started {
+  isLocal: false,
+  attempt: 1,
+  namespace: 'default',
+  taskToken: 'CiRjZGNlMGM2Mi1mYzdjLTQyODctOGM2MC1kNWYwOWIzNDhjMWESHndvcmtmbG93LTJZcmlPbG1nSzJ2TVQtcmhSRlFRcBokZmEwYjQxNWItZWY4OS00MzRkLTgzOWMtZWMwMzFkMzE2NjhlIAUoATIBMUIFZ3JlZXRKCAgBEPKUQxgB',
+  workflowId: 'workflow-2YriOlmgK2vMT-rhRFQQp',
+  workflowRunId: 'fa0b415b-ef89-434d-839c-ec031d31668e',
+  workflowType: 'logSampleWorkflow',
+  activityId: '1',
+  activityType: 'greet',
+  taskQueue: 'custom-logger'
+}
+2023-09-05T18:19:51.994Z [activity] info: Log from activity {
+  isLocal: false,
+  attempt: 1,
+  namespace: 'default',
+  taskToken: 'CiRjZGNlMGM2Mi1mYzdjLTQyODctOGM2MC1kNWYwOWIzNDhjMWESHndvcmtmbG93LTJZcmlPbG1nSzJ2TVQtcmhSRlFRcBokZmEwYjQxNWItZWY4OS00MzRkLTgzOWMtZWMwMzFkMzE2NjhlIAUoATIBMUIFZ3JlZXRKCAgBEPKUQxgB',
+  workflowId: 'workflow-2YriOlmgK2vMT-rhRFQQp',
+  workflowRunId: 'fa0b415b-ef89-434d-839c-ec031d31668e',
+  workflowType: 'logSampleWorkflow',
+  activityId: '1',
+  activityType: 'greet',
+  taskQueue: 'custom-logger',
+  name: 'Temporal'
+}
+2023-09-05T18:19:51.994Z [activity] debug: Activity completed {
+  isLocal: false,
+  attempt: 1,
+  namespace: 'default',
+  taskToken: 'CiRjZGNlMGM2Mi1mYzdjLTQyODctOGM2MC1kNWYwOWIzNDhjMWESHndvcmtmbG93LTJZcmlPbG1nSzJ2TVQtcmhSRlFRcBokZmEwYjQxNWItZWY4OS00MzRkLTgzOWMtZWMwMzFkMzE2NjhlIAUoATIBMUIFZ3JlZXRKCAgBEPKUQxgB',
+  workflowId: 'workflow-2YriOlmgK2vMT-rhRFQQp',
+  workflowRunId: 'fa0b415b-ef89-434d-839c-ec031d31668e',
+  workflowType: 'logSampleWorkflow',
+  activityId: '1',
+  activityType: 'greet',
+  taskQueue: 'custom-logger',
+  durationMs: 0
+}
+2023-09-05T18:19:51.999Z [workerd] debug: New WFT
+2023-09-05T18:19:51.999Z [workerd] debug: Applying new workflow task from server
+2023-09-05T18:19:51.999Z [workerd] debug: Sending activation to lang
+2023-09-05T18:19:52.001Z [workflow] info: Greeted {
+  namespace: 'default',
+  taskQueue: 'custom-logger',
+  workflowId: 'workflow-2YriOlmgK2vMT-rhRFQQp',
+  runId: 'fa0b415b-ef89-434d-839c-ec031d31668e',
+  workflowType: 'logSampleWorkflow',
+  greeting: 'Hello, Temporal!'
+}
+2023-09-05T18:19:52.001Z [workflow] debug: Workflow completed {
+  namespace: 'default',
+  taskQueue: 'custom-logger',
+  workflowId: 'workflow-2YriOlmgK2vMT-rhRFQQp',
+  runId: 'fa0b415b-ef89-434d-839c-ec031d31668e',
+  workflowType: 'logSampleWorkflow'
+}
+2023-09-05T18:19:52.001Z [worker] debug: wf bridge iteration fetch
+2023-09-05T18:19:52.001Z [worker] debug: handling command
+2023-09-05T18:19:52.001Z [worker] debug: prepared commands
+2023-09-05T18:19:52.001Z [worker] debug: Sending responses to server
+2023-09-05T18:19:52.004Z [worker] debug: Marking WFT completed
+```
+
+</details>

context-propagation/package.json

@@ -0,0 +1,43 @@
+{
+  "name": "temporal-context-propagation",
+  "version": "0.1.0",
+  "private": true,
+  "scripts": {
+    "build": "tsc --build",
+    "build.watch": "tsc --build --watch",
+    "lint": "eslint .",
+    "start": "ts-node src/worker.ts",
+    "start.watch": "nodemon src/worker.ts",
+    "workflow": "ts-node src/client.ts"
+  },
+  "nodemonConfig": {
+    "execMap": {
+      "ts": "ts-node"
+    },
+    "ext": "ts",
+    "watch": [
+      "src"
+    ]
+  },
+  "dependencies": {
+    "@temporalio/activity": "^1.9.0",
+    "@temporalio/client": "^1.9.0",
+    "@temporalio/worker": "^1.9.0",
+    "@temporalio/workflow": "^1.9.0",
+    "nanoid": "3.x"
+  },
+  "devDependencies": {
+    "@tsconfig/node16": "^1.0.0",
+    "@types/node": "^16.11.43",
+    "@types/triple-beam": "^1.3.2",
+    "@typescript-eslint/eslint-plugin": "^5.0.0",
+    "@typescript-eslint/parser": "^5.0.0",
+    "eslint": "^7.32.0",
+    "eslint-config-prettier": "^8.3.0",
+    "eslint-plugin-deprecation": "^1.2.1",
+    "nodemon": "^2.0.12",
+    "prettier": "^2.8.8",
+    "ts-node": "^10.2.1",
+    "typescript": "^4.4.2"
+  }
+}

context-propagation/src/activities/index.ts

@@ -0,0 +1,8 @@
+import { log } from '@temporalio/activity';
+import { getContext } from '../context/context-injection';
+
+export async function greet(name: string): Promise<string> {
+  const propagatedContext = getContext();
+  log.info(`Log from activity with customer ${propagatedContext.customer ?? 'unknown'}`);
+  return `Hello, ${name}!`;
+}

context-propagation/src/client.ts

@@ -0,0 +1,32 @@
+import { nanoid } from 'nanoid';
+import { Connection, Client } from '@temporalio/client';
+import { sampleWorkflow } from './workflows';
+import { ContextClientInterceptor } from './context/client-interceptors';
+import { withContext } from './context/context-injection';
+
+async function run() {
+  const connection = await Connection.connect({
+    address: 'localhost:7233',
+    // If appropriate, configure TLS and other settings.
+    // This is optional but we leave this here to remind you there is a gRPC connection being established.
+  });
+
+  const clientInterceptor = new ContextClientInterceptor();
+  const client = new Client({
+    connection,
+    namespace: 'default',
+    interceptors: { workflow: [clientInterceptor], schedule: [clientInterceptor] },
+  });
+
+  await withContext({ customer: 'Acme Inc.' }, async () => {
+    await client.workflow.execute(sampleWorkflow, {
+      taskQueue: 'context-propagation',
+      workflowId: 'workflow-' + nanoid(),
+    });
+  });
+}
+
+run().catch((err) => {
+  console.error(err);
+  process.exit(1);
+});

context-propagation/src/context/activity-interceptors.ts

@@ -0,0 +1,28 @@
+import { ActivityInterceptors } from '@temporalio/worker';
+import { extractContextHeader } from './context-type';
+import { withContext, getContext } from './context-injection';
+
+/**
+ * Intercepts activity start, to restore the context received through headers.
+ * This interceptor also add the content of the context as log metadata.
+ */
+export function newContextActivityInterceptor(): ActivityInterceptors {
+  return {
+    inbound: {
+      async execute(input, next) {
+        const inboundContext = extractContextHeader(input.headers);
+        return await withContext(inboundContext ?? {}, () => {
+          return next(input);
+        });
+      },
+    },
+    outbound: {
+      getLogAttributes(input, next) {
+        return next({
+          input,
+          ...getContext(),
+        });
+      },
+    },
+  };
+}