clean up tests and flags

Author: chris-olszewski

packages/interceptors-opentelemetry/src/instrumentation.ts

@@ -32,7 +32,7 @@ export function extractContextFromHeaders(headers: Headers): otel.Context | unde
 /**
  * Given headers, return new headers with the current otel context inserted
  */
-export async function headersWithContext(headers: Headers): Promise<Headers> {
+export function headersWithContext(headers: Headers): Headers {
   const carrier = {};
   otel.propagation.inject(otel.context.active(), carrier, otel.defaultTextMapSetter);
   return { ...headers, [TRACE_HEADER]: payloadConverter.toPayload(carrier) };

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

@@ -66,11 +66,12 @@ export class OpenTelemetryInboundInterceptor implements WorkflowInboundCallsInte
     const { getActivator, SdkFlags } = getSdkFlagsChecking();
     const shouldInjectYield = getActivator().hasFlag(SdkFlags.OpenTelemetryInterceptorInsertYield);
     const context = extractContextFromHeaders(input.headers);
+    if (shouldInjectYield) await Promise.resolve();
     return await instrument({
       tracer: this.tracer,
       spanName: `${SpanName.WORKFLOW_EXECUTE}${SPAN_DELIMITER}${workflowInfo().workflowType}`,
       fn: () => next(input),
-      context: shouldInjectYield ? await Promise.resolve(context) : context,
+      context,
       acceptableErrors: (err) => err instanceof ContinueAsNew,
     });
   }
@@ -82,11 +83,12 @@ export class OpenTelemetryInboundInterceptor implements WorkflowInboundCallsInte
     const { getActivator, SdkFlags } = getSdkFlagsChecking();
     const shouldInjectYield = getActivator().hasFlag(SdkFlags.OpenTelemetryHandleSignalInterceptorInsertYield);
     const context = extractContextFromHeaders(input.headers);
+    if (shouldInjectYield) await Promise.resolve();
     return await instrument({
       tracer: this.tracer,
       spanName: `${SpanName.WORKFLOW_SIGNAL}${SPAN_DELIMITER}${input.signalName}`,
       fn: () => next(input),
-      context: shouldInjectYield ? await Promise.resolve(context) : context,
+      context,
     });
   }
 }
@@ -109,11 +111,14 @@ export class OpenTelemetryOutboundInterceptor implements WorkflowOutboundCallsIn
     input: ActivityInput,
     next: Next<WorkflowOutboundCallsInterceptor, 'scheduleActivity'>
   ): Promise<unknown> {
+    const { getActivator, SdkFlags } = getSdkFlagsChecking();
+    const shouldInjectYield = getActivator().hasFlag(SdkFlags.OpenTelemetryInterceptorInsertYield);
     return await instrument({
       tracer: this.tracer,
       spanName: `${SpanName.ACTIVITY_START}${SPAN_DELIMITER}${input.activityType}`,
       fn: async () => {
-        const headers = await headersWithContext(input.headers);
+        const headers = headersWithContext(input.headers);
+        if (shouldInjectYield) await Promise.resolve();
         return next({
           ...input,
           headers,
@@ -126,11 +131,14 @@ export class OpenTelemetryOutboundInterceptor implements WorkflowOutboundCallsIn
     input: LocalActivityInput,
     next: Next<WorkflowOutboundCallsInterceptor, 'scheduleLocalActivity'>
   ): Promise<unknown> {
+    const { getActivator, SdkFlags } = getSdkFlagsChecking();
+    const shouldInjectYield = getActivator().hasFlag(SdkFlags.OpenTelemetryScheduleLocalActivityInterceptorInsertYield);
     return await instrument({
       tracer: this.tracer,
       spanName: `${SpanName.ACTIVITY_START}${SPAN_DELIMITER}${input.activityType}`,
       fn: async () => {
-        const headers = await headersWithContext(input.headers);
+        const headers = headersWithContext(input.headers);
+        if (shouldInjectYield) await Promise.resolve();
         return next({
           ...input,
           headers,
@@ -143,11 +151,14 @@ export class OpenTelemetryOutboundInterceptor implements WorkflowOutboundCallsIn
     input: StartChildWorkflowExecutionInput,
     next: Next<WorkflowOutboundCallsInterceptor, 'startChildWorkflowExecution'>
   ): Promise<[Promise<string>, Promise<unknown>]> {
+    const { getActivator, SdkFlags } = getSdkFlagsChecking();
+    const shouldInjectYield = getActivator().hasFlag(SdkFlags.OpenTelemetryInterceptorInsertYield);
     return await instrument({
       tracer: this.tracer,
       spanName: `${SpanName.CHILD_WORKFLOW_START}${SPAN_DELIMITER}${input.workflowType}`,
       fn: async () => {
-        const headers = await headersWithContext(input.headers);
+        const headers = headersWithContext(input.headers);
+        if (shouldInjectYield) await Promise.resolve();
         return next({
           ...input,
           headers,
@@ -161,11 +172,14 @@ export class OpenTelemetryOutboundInterceptor implements WorkflowOutboundCallsIn
     next: Next<WorkflowOutboundCallsInterceptor, 'continueAsNew'>
   ): Promise<never> {
     const { ContinueAsNew } = getWorkflowModule();
+    const { getActivator, SdkFlags } = getSdkFlagsChecking();
+    const shouldInjectYield = getActivator().hasFlag(SdkFlags.OpenTelemetryInterceptorInsertYield);
     return await instrument({
       tracer: this.tracer,
       spanName: `${SpanName.CONTINUE_AS_NEW}${SPAN_DELIMITER}${input.options.workflowType}`,
       fn: async () => {
-        const headers = await headersWithContext(input.headers);
+        const headers = headersWithContext(input.headers);
+        if (shouldInjectYield) await Promise.resolve();
         return next({
           ...input,
           headers,
@@ -179,11 +193,14 @@ export class OpenTelemetryOutboundInterceptor implements WorkflowOutboundCallsIn
     input: SignalWorkflowInput,
     next: Next<WorkflowOutboundCallsInterceptor, 'signalWorkflow'>
   ): Promise<void> {
+    const { getActivator, SdkFlags } = getSdkFlagsChecking();
+    const shouldInjectYield = getActivator().hasFlag(SdkFlags.OpenTelemetryInterceptorInsertYield);
     return await instrument({
       tracer: this.tracer,
       spanName: `${SpanName.WORKFLOW_SIGNAL}${SPAN_DELIMITER}${input.signalName}`,
       fn: async () => {
-        const headers = await headersWithContext(input.headers);
+        const headers = headersWithContext(input.headers);
+        if (shouldInjectYield) await Promise.resolve();
         return next({
           ...input,
           headers,

packages/test/src/test-flags.ts

@@ -16,6 +16,7 @@ function composeConditions(conditions: Conditions): NonNullable<Conditions>[numb
 
 test('OpenTelemetryHandleSignalInterceptorInsertYield enabled by version', (t) => {
   const cases = [
+    { version: undefined, expected: false },
     { version: '1.0.0', expected: false },
     { version: '1.11.3', expected: false },
     { version: '1.11.5', expected: true },
@@ -40,6 +41,8 @@ test('OpenTelemetryHandleSignalInterceptorInsertYield enabled by version', (t) =
 
 test('OpenTelemetryInterceptorInsertYield enabled by version', (t) => {
   const cases = [
+    // If there isn't any SDK version available we enable this flag as these yields were present since the initial version of the OTEL interceptors
+    { version: undefined, expected: true },
     { version: '0.1.0', expected: true },
     { version: '1.0.0', expected: true },
     { version: '1.9.0-rc.0', expected: true },
@@ -57,3 +60,30 @@ test('OpenTelemetryInterceptorInsertYield enabled by version', (t) => {
     t.is(actual, expected, `Expected OpenTelemetryInterceptorInsertYield on ${version} to evaluate as ${expected}`);
   }
 });
+
+test('OpenTelemetryScheduleLocalActivityInterceptorInsertYield enabled by version', (t) => {
+  const cases = [
+    { version: undefined, expected: false },
+    { version: '1.0.0', expected: false },
+    { version: '1.11.3', expected: false },
+    { version: '1.11.5', expected: false },
+    { version: '1.11.6', expected: true },
+    { version: '1.12.0', expected: true },
+    { version: '1.13.1', expected: true },
+    { version: '1.13.2', expected: false },
+    { version: '1.14.0', expected: false },
+  ];
+  for (const { version, expected } of cases) {
+    const actual = composeConditions(
+      SdkFlags.OpenTelemetryScheduleLocalActivityInterceptorInsertYield.alternativeConditions
+    )({
+      info: {} as WorkflowInfo,
+      sdkVersion: version,
+    });
+    t.is(
+      actual,
+      expected,
+      `Expected OpenTelemetryScheduleLocalActivityInterceptorInsertYield on ${version} to evaluate as ${expected}`
+    );
+  }
+});

packages/test/src/test-otel.ts

@@ -24,7 +24,7 @@ import {
 import { OpenTelemetrySinks, SpanName, SPAN_DELIMITER } from '@temporalio/interceptors-opentelemetry/lib/workflow';
 import { DefaultLogger, InjectedSinks, Runtime } from '@temporalio/worker';
 import * as activities from './activities';
-import { loadHistory, RUN_INTEGRATION_TESTS, saveHistory, TestWorkflowEnvironment, Worker } from './helpers';
+import { loadHistory, RUN_INTEGRATION_TESTS, TestWorkflowEnvironment, Worker } from './helpers';
 import * as workflows from './workflows';
 import { createTestWorkflowBundle } from './helpers-integration';
 
@@ -559,40 +559,7 @@ test('Can replay otel history from 1.13.1', async (t) => {
 });
 
 test('Can replay smorgasbord from 1.13.1', async (t) => {
-  /*
-  const staticResource = new opentelemetry.resources.Resource({
-    [SemanticResourceAttributes.SERVICE_NAME]: 'ts-test-otel-worker',
-  });
-  const worker = await Worker.create({
-    workflowsPath: require.resolve('./workflows'),
-    activities,
-    taskQueue: 'test-otel-inbound',
-    sinks: {
-      exporter: makeWorkflowExporter(new InMemorySpanExporter(), staticResource),
-    },
-    interceptors: {
-      workflowModules: [require.resolve('./workflows/otel-interceptors')],
-      activity: [
-        (ctx) => ({
-          inbound: new OpenTelemetryActivityInboundInterceptor(ctx),
-        }),
-      ],
-    },
-  });
-  const client = new WorkflowClient();
-
-  const result = await worker.runUntil(async () => {
-    const handle = await client.start(workflows.smorgasbord, {
-      taskQueue: 'test-otel-inbound',
-      workflowId: uuid4(),
-    });
-    const result = await handle.result();
-    const history = await handle.fetchHistory();
-    await saveHistory('smorg_with_otel.json', history);
-    return result;
-  });
-  */
-
+  // This test will trigger NDE if yield points for `scheduleActivity` and `startChildWorkflowExecution` are not inserted
   const hist = await loadHistory('otel_smorgasbord_1_13_1.json');
   await t.notThrowsAsync(async () => {
     await Worker.runReplayHistory(
@@ -616,41 +583,6 @@ test('Can replay smorgasbord from 1.13.1', async (t) => {
 });
 
 test('Can replay signal workflow from 1.13.1', async (t) => {
-  /*
-  const staticResource = new opentelemetry.resources.Resource({
-    [SemanticResourceAttributes.SERVICE_NAME]: 'ts-test-otel-worker',
-  });
-  const worker = await Worker.create({
-    workflowsPath: require.resolve('./workflows/signal-workflow'),
-    activities: [],
-    taskQueue: 'test-otel-inbound',
-    sinks: {
-      exporter: makeWorkflowExporter(new InMemorySpanExporter(), staticResource),
-    },
-    interceptors: {
-      workflowModules: [require.resolve('./workflows/otel-interceptors')],
-      activity: [
-        (ctx) => ({
-          inbound: new OpenTelemetryActivityInboundInterceptor(ctx),
-        }),
-      ],
-    },
-  });
-  const client = new WorkflowClient();
-
-  const result = await worker.runUntil(async () => {
-    const handle = await client.start(workflows.topSecretGreeting, {
-      args: ['Temporal'],
-      taskQueue: 'test-otel-inbound',
-      workflowId: uuid4(),
-    });
-    const result = await handle.result();
-    const history = await handle.fetchHistory();
-    await saveHistory('signal_workflow_1_13_1.json', history);
-    return result;
-  });
-  */
-
   const hist = await loadHistory('signal_workflow_1_13_1.json');
   await t.notThrowsAsync(async () => {
     await Worker.runReplayHistory(
@@ -671,5 +603,4 @@ test('Can replay signal workflow from 1.13.1', async (t) => {
       hist
     );
   });
-  t.pass();
 });

packages/workflow/src/flags.ts

@@ -63,9 +63,25 @@ export const SdkFlags = {
    * The interceptors provided by @temporalio/interceptors-opentelemetry initially had unnecessary yield points.
    * If replaying a workflow created from these versions a yield point is injected to prevent any NDE.
    *
+   * If the history does not include the SDK version, default to enabled since the yields were present since the OTEL
+   * package was created.
+   *
+   * @since Introduced in 1.13.2
+   */
+  OpenTelemetryInterceptorInsertYield: defineFlag(3, false, [isBefore({ major: 1, minor: 13, patch: 2 }, true)]),
+  /**
+   * In 1.11.6, the `scheduleLocalActivity` interceptor was added to
+   * `@temporalio/interceptors-opentelemetry` which added a yield point to the
+   * outbound interceptor. This yield point was removed in 1.13.2.
+   *
+   * If replaying a workflow from 1.11.6 up to 1.13.1, we insert a yield point
+   * in the interceptor to match the behavior.
+   *
    * @since Introduced in 1.13.2
    */
-  OpenTelemetryInterceptorInsertYield: defineFlag(3, false, [isBefore({ major: 1, minor: 13, patch: 2 })]),
+  OpenTelemetryScheduleLocalActivityInterceptorInsertYield: defineFlag(4, false, [
+    isBetween({ major: 1, minor: 11, patch: 5 }, { major: 1, minor: 13, patch: 2 }),
+  ]),
 } as const;
 
 function defineFlag(id: number, def: boolean, alternativeConditions?: AltConditionFn[]): SdkFlag {
@@ -103,6 +119,40 @@ type SemVer = {
   patch: number;
 };
 
+/**
+ * Creates an `AltConditionFn` that checks if the SDK version is before the provided version.
+ * An optional default can be provided in case the SDK version is not available.
+ */
+function isBefore(compare: SemVer, missingDefault?: boolean): AltConditionFn {
+  return isCompared(compare, 1, missingDefault);
+}
+
+/**
+ * Creates an `AltConditionFn` that checks if the SDK version is after the provided version.
+ * An optional default can be provided in case the SDK version is not available.
+ */
+function isAfter(compare: SemVer, missingDefault?: boolean): AltConditionFn {
+  return isCompared(compare, -1, missingDefault);
+}
+
+/**
+ * Creates an `AltConditionFn` that checks if the SDK version is between the provided versions.
+ * The range check is exclusive.
+ * An optional default can be provided in case the SDK version is not available.
+ */
+function isBetween(lowEnd: SemVer, highEnd: SemVer, missingDefault?: boolean): AltConditionFn {
+  return (ctx) => isAfter(lowEnd, missingDefault)(ctx) && isBefore(highEnd, missingDefault)(ctx);
+}
+
+function isCompared(compare: SemVer, comparison: -1 | 0 | 1, missingDefault: boolean = false): AltConditionFn {
+  return ({ sdkVersion }) => {
+    if (!sdkVersion) return missingDefault;
+    const version = parseSemver(sdkVersion);
+    if (!version) return missingDefault;
+    return compareSemver(compare, version) === comparison;
+  };
+}
+
 function parseSemver(version: string): SemVer | undefined {
   const matches = version.match(/(\d+)\.(\d+)\.(\d+)/);
   if (!matches) return undefined;
@@ -132,30 +182,3 @@ function compareSemver(a: SemVer, b: SemVer): -1 | 0 | 1 {
   if (a.patch > b.patch) return 1;
   return 0;
 }
-
-function isCompared(compare: SemVer, comparison: -1 | 0 | 1): AltConditionFn {
-  return ({ sdkVersion }) => {
-    if (!sdkVersion) throw new Error('no sdk version');
-    if (!sdkVersion) return false;
-    const version = parseSemver(sdkVersion);
-    if (!version) throw new Error(`no version for ${sdkVersion}`);
-    if (!version) return false;
-    return compareSemver(compare, version) === comparison;
-  };
-}
-
-function isBefore(compare: SemVer): AltConditionFn {
-  return isCompared(compare, 1);
-}
-
-function isEqual(compare: SemVer): AltConditionFn {
-  return isCompared(compare, 0);
-}
-
-function isAfter(compare: SemVer): AltConditionFn {
-  return isCompared(compare, -1);
-}
-
-function isBetween(lowEnd: SemVer, highEnd: SemVer): AltConditionFn {
-  return (ctx) => isAfter(lowEnd)(ctx) && isBefore(highEnd)(ctx);
-}