Instrument signals with OTel (#1449)

Co-authored-by: James Watkins-Harvey <[email protected]>

Author: iravid

packages/interceptors-opentelemetry/src/client/index.ts

@@ -1,5 +1,5 @@
 import * as otel from '@opentelemetry/api';
-import { Next, WorkflowClientInterceptor, WorkflowStartInput } from '@temporalio/client';
+import { Next, WorkflowClientInterceptor, WorkflowSignalInput, WorkflowStartInput } from '@temporalio/client';
 import { instrument, headersWithContext, RUN_ID_ATTR_KEY } from '../instrumentation';
 import { SpanName, SPAN_DELIMITER } from '../workflow';
 
@@ -31,4 +31,15 @@ export class OpenTelemetryWorkflowClientInterceptor implements WorkflowClientInt
       },
     });
   }
+
+  async signal(input: WorkflowSignalInput, next: Next<WorkflowClientInterceptor, 'signal'>): Promise<void> {
+    return await instrument({
+      tracer: this.tracer,
+      spanName: `${SpanName.WORKFLOW_SIGNAL}${SPAN_DELIMITER}${input.signalName}`,
+      fn: async () => {
+        const headers = await headersWithContext(input.headers);
+        await next({ ...input, headers });
+      },
+    });
+  }
 }

packages/interceptors-opentelemetry/src/workflow/definitions.ts

@@ -44,6 +44,11 @@ export enum SpanName {
    */
   WORKFLOW_START = 'StartWorkflow',
 
+  /**
+   * Workflow is signalled
+   */
+  WORKFLOW_SIGNAL = 'SignalWorkflow',
+
   /**
    * Workflow is client calls signalWithStart
    */

packages/interceptors-opentelemetry/src/workflow/index.ts

@@ -9,6 +9,8 @@ import {
   ContinueAsNewInput,
   DisposeInput,
   Next,
+  SignalInput,
+  SignalWorkflowInput,
   StartChildWorkflowExecutionInput,
   WorkflowExecuteInput,
   WorkflowInboundCallsInterceptor,
@@ -61,6 +63,19 @@ export class OpenTelemetryInboundInterceptor implements WorkflowInboundCallsInte
       acceptableErrors: (err) => err instanceof ContinueAsNew,
     });
   }
+
+  public async handleSignal(
+    input: SignalInput,
+    next: Next<WorkflowInboundCallsInterceptor, 'handleSignal'>
+  ): Promise<void> {
+    const context = await extractContextFromHeaders(input.headers);
+    return await instrument({
+      tracer: this.tracer,
+      spanName: `${SpanName.WORKFLOW_SIGNAL}${SPAN_DELIMITER}${input.signalName}`,
+      fn: () => next(input),
+      context,
+    });
+  }
 }
 
 /**
@@ -122,6 +137,23 @@ export class OpenTelemetryOutboundInterceptor implements WorkflowOutboundCallsIn
       acceptableErrors: (err) => err instanceof ContinueAsNew,
     });
   }
+
+  public async signalWorkflow(
+    input: SignalWorkflowInput,
+    next: Next<WorkflowOutboundCallsInterceptor, 'signalWorkflow'>
+  ): Promise<void> {
+    return await instrument({
+      tracer: this.tracer,
+      spanName: `${SpanName.WORKFLOW_SIGNAL}${SPAN_DELIMITER}${input.signalName}`,
+      fn: async () => {
+        const headers = await headersWithContext(input.headers);
+        return next({
+          ...input,
+          headers,
+        });
+      },
+    });
+  }
 }
 
 export class OpenTelemetryInternalsInterceptor implements WorkflowInternalsInterceptor {

packages/test/src/activities/helpers.ts

@@ -1,11 +1,19 @@
+import { OpenTelemetryWorkflowClientCallsInterceptor } from '@temporalio/interceptors-opentelemetry';
 import { Client, WorkflowHandle } from '@temporalio/client';
 import { QueryDefinition } from '@temporalio/common';
 import { getContext } from './interceptors';
 
 function getSchedulingWorkflowHandle(): WorkflowHandle {
   const { info, connection, dataConverter } = getContext();
   const { workflowExecution } = info;
-  const client = new Client({ connection, namespace: info.workflowNamespace, dataConverter });
+  const client = new Client({
+    connection,
+    namespace: info.workflowNamespace,
+    dataConverter,
+    interceptors: {
+      workflow: [new OpenTelemetryWorkflowClientCallsInterceptor()],
+    },
+  });
   return client.workflow.getHandle(workflowExecution.workflowId, workflowExecution.runId);
 }
 

packages/test/src/test-otel.ts

@@ -232,7 +232,22 @@ if (RUN_INTEGRATION_TESTS) {
         parentSpanId === childWorkflowStartSpan?.spanContext().spanId
     );
     t.true(childWorkflowExecuteSpan !== undefined);
-    t.true(new Set(spans.map((span) => span.spanContext().traceId)).size === 1);
+
+    const signalChildWithUnblockSpan = spans.find(
+      ({ name, parentSpanId }) =>
+        name === `${SpanName.WORKFLOW_SIGNAL}${SPAN_DELIMITER}unblock` &&
+        parentSpanId === parentExecuteSpan?.spanContext().spanId
+    );
+    t.true(signalChildWithUnblockSpan !== undefined);
+
+    const activityStartedSignalSpan = spans.find(
+      ({ name, parentSpanId }) =>
+        name === `${SpanName.WORKFLOW_SIGNAL}${SPAN_DELIMITER}activityStarted` &&
+        parentSpanId === firstActivityExecuteSpan?.spanContext().spanId
+    );
+    t.true(activityStartedSignalSpan !== undefined);
+
+    t.deepEqual(new Set(spans.map((span) => span.spanContext().traceId)).size, 1);
   });
 
   // Un-skip this test and run it by hand to inspect outputted traces