@kubiks/otel-upstash-workflow

OpenTelemetry instrumentation for the Upstash Workflow Node.js SDK. Capture spans for every workflow execution and step, enrich them with operation metadata, and keep an eye on workflow operations from your traces.

Visualize your workflow executions with detailed span information including steps, sleep operations, API calls, and performance metrics.

Note: This package instruments the Upstash Workflow SDK, which is currently in pre-release. The API may change as the Workflow SDK evolves.

Installation

npm install @kubiks/otel-upstash-workflow
# or
pnpm add @kubiks/otel-upstash-workflow

Peer Dependencies: @opentelemetry/api >= 1.9.0, @upstash/workflow >= 0.0.0

Quick Start

Instrumenting Workflow Handlers

import { serve as originalServe } from "@upstash/workflow";
import { instrumentWorkflowServe } from "@kubiks/otel-upstash-workflow";

const serve = instrumentWorkflowServe(originalServe);

export const POST = serve(async (context) => {
  const result1 = await context.run("step-1", async () => {
    return await processData();
  });

  await context.sleep("wait-5s", 5);

  const result2 = await context.run("step-2", async () => {
    return await saveResults(result1);
  });

  return result2;
});

instrumentWorkflowServe wraps the serve function to trace the entire workflow execution and all steps — no configuration changes needed. Every workflow execution creates a server span with child spans for each step.

Instrumenting Workflow Client

import { Client } from "@upstash/workflow";
import { instrumentWorkflowClient } from "@kubiks/otel-upstash-workflow";

const client = instrumentWorkflowClient(
  new Client({ baseUrl: process.env.QSTASH_URL!, token: process.env.QSTASH_TOKEN! })
);

await client.trigger({
  url: "https://your-app.com/api/workflow",
  body: { data: "example" },
});

instrumentWorkflowClient wraps the workflow client to trace workflow triggers, creating client spans for each trigger operation.

With Step Data Capture

Optionally capture step inputs and outputs for debugging:

const serve = instrumentWorkflowServe(originalServe, {
  captureStepData: true,       // Enable step data capture (default: false)
  maxStepDataLength: 2048,     // Max characters to capture (default: 1024)
});

export const POST = serve(async (context) => {
  // Your workflow - all steps are traced with input/output capture
});

What Gets Traced

This instrumentation provides two main functions:

1. instrumentWorkflowClient - Wraps the Workflow Client to trace workflow triggers
2. instrumentWorkflowServe - Wraps the serve function to trace execution and all workflow steps

Workflow Handler Instrumentation

The instrumentWorkflowServe function wraps the serve function, creating a span with SpanKind.SERVER for the entire workflow execution. All workflow steps (context.run, context.sleep, etc.) automatically create child spans.

Client Instrumentation

The instrumentWorkflowClient function wraps the client's trigger method, creating a span with SpanKind.CLIENT for each workflow trigger operation.

Span Hierarchy

The instrumentation creates the following span hierarchy:

[SERVER] workflow.execute
  ├─ [INTERNAL] workflow.step.step-1 (context.run)
  ├─ [INTERNAL] workflow.step.wait-5s (context.sleep)
  ├─ [CLIENT] workflow.step.api-call (context.call)
  └─ [INTERNAL] workflow.step.wait-event (context.waitForEvent)

Separate client-side triggers create independent traces:

[CLIENT] workflow.trigger

Span Attributes

Workflow Handler Spans (instrumentWorkflowServe)

| Attribute | Description | Example | | --- | --- | --- | | workflow.system | Constant value upstash | upstash | | workflow.operation | Operation type | execute | | workflow.id | Workflow ID from headers | wf_123 | | workflow.run_id | Workflow run ID from headers | run_456 | | workflow.url | Workflow URL from headers | https://example.com/api/workflow | | http.status_code | HTTP response status | 200 |

Client Trigger Spans (instrumentWorkflowClient)

| Attribute | Description | Example | | --- | --- | --- | | workflow.system | Constant value upstash | upstash | | workflow.operation | Operation type | trigger | | workflow.url | Target workflow URL | https://example.com/api/workflow | | workflow.id | Workflow ID from response | wf_123 | | workflow.run_id | Workflow run ID from response | run_456 |

Step Spans (context.run)

| Attribute | Description | Example | | --- | --- | --- | | workflow.system | Constant value upstash | upstash | | workflow.operation | Operation type | step | | workflow.step.name | Step name | step-1 | | workflow.step.type | Step type | run | | workflow.step.duration_ms | Step execution time in ms | 150 | | workflow.step.output | Step output (if enabled) | {"result":"success"} |

Sleep Spans (context.sleep, context.sleepFor, context.sleepUntil)

| Attribute | Description | Example | | --- | --- | --- | | workflow.system | Constant value upstash | upstash | | workflow.operation | Operation type | step | | workflow.step.name | Step name (if named sleep) | wait-5s | | workflow.step.type | Step type | sleep | | workflow.sleep.duration_ms | Sleep duration in ms | 5000 | | workflow.sleep.until_timestamp | Target timestamp (sleepUntil) | 1704067200000 |

Call Spans (context.call)

| Attribute | Description | Example | | --- | --- | --- | | workflow.system | Constant value upstash | upstash | | workflow.operation | Operation type | step | | workflow.step.name | Step name | api-call | | workflow.step.type | Step type | call | | workflow.call.url | Target URL | https://api.example.com/data | | workflow.call.method | HTTP method | POST | | workflow.call.status_code | Response status code | 200 | | workflow.step.input | Request body (if enabled) | {"userId":"123"} | | workflow.step.output | Response data (if enabled) | {"status":"ok"} |

Event Spans (context.waitForEvent)

| Attribute | Description | Example | | --- | --- | --- | | workflow.system | Constant value upstash | upstash | | workflow.operation | Operation type | step | | workflow.step.name | Step name | wait-event | | workflow.step.type | Step type | waitForEvent | | workflow.event.id | Event ID | evt_123 | | workflow.event.timeout_ms | Timeout in ms | 60000 | | workflow.step.output | Event data (if enabled) | {"received":true} |

Step Data Attributes (Optional)

When captureStepData is enabled in configuration:

| Attribute | Description | Captured By | | --- | --- | --- | | workflow.step.input | Step input data | Client trigger, context.call | | workflow.step.output | Step output data | All context methods |

The instrumentation captures workflow metadata and step details to help with debugging and monitoring. Step data capture is disabled by default to protect sensitive data.

Usage Examples

Basic Workflow Execution

import { serve as originalServe } from "@upstash/workflow";
import { instrumentWorkflowServe } from "@kubiks/otel-upstash-workflow";

const serve = instrumentWorkflowServe(originalServe);

export const POST = serve(async (context) => {
  const data = await context.run("fetch-data", async () => {
    return await fetchFromDatabase();
  });

  const processed = await context.run("process-data", async () => {
    return await processData(data);
  });

  return { success: true, result: processed };
});

Workflow with Sleep

const serve = instrumentWorkflowServe(originalServe);

export const POST = serve(async (context) => {
  await context.run("send-email", async () => {
    await sendEmail();
  });

  await context.sleep("wait-5s", 5);

  await context.run("check-status", async () => {
    return await checkEmailStatus();
  });

  return { done: true };
});

Workflow with External API Calls

const serve = instrumentWorkflowServe(originalServe);

export const POST = serve(async (context) => {
  const apiResponse = await context.call("fetch-user", "https://api.example.com/users/123", {
    method: "GET",
  });

  const result = await context.run("process-user", async () => {
    return await processUser(apiResponse);
  });

  return result;
});

Workflow with Event Waiting

const serve = instrumentWorkflowServe(originalServe);

export const POST = serve(async (context) => {
  await context.run("start-process", async () => {
    await startLongRunningProcess();
  });

  const event = await context.waitForEvent("process-complete", "evt_123", 60000);

  await context.run("finalize", async () => {
    return await finalizeProcess(event);
  });

  return { success: true };
});

Client Triggering Workflows

import { Client } from "@upstash/workflow";
import { instrumentWorkflowClient } from "@kubiks/otel-upstash-workflow";

const client = instrumentWorkflowClient(
  new Client({
    baseUrl: process.env.QSTASH_URL!,
    token: process.env.QSTASH_TOKEN!,
  })
);

const result = await client.trigger({
  url: "https://your-app.vercel.app/api/workflow",
  body: {
    userId: "user_123",
    action: "process_data",
  },
});

console.log("Workflow triggered:", result.workflowId);

With Step Data Capture

const serve = instrumentWorkflowServe(originalServe, {
  captureStepData: true,      // Enable input/output capture
  maxStepDataLength: 2048,    // Increase truncation limit
});

export const POST = serve(async (context) => {
  const result = await context.run("complex-calculation", async () => {
    return {
      value: 42,
      timestamp: Date.now(),
      metadata: { processed: true },
    };
  });

  return result;
});

Complete Next.js Integration Example

Workflow handler:

// app/api/workflow/route.ts
import { serve as originalServe } from "@upstash/workflow";
import { instrumentWorkflowServe } from "@kubiks/otel-upstash-workflow";

const serve = instrumentWorkflowServe(originalServe);

export const POST = serve(async (context) => {
  const orderId = context.requestPayload.orderId;

  const result = await context.run("process-order", async () => {
    return await processOrder(orderId);
  });

  await context.sleep("wait-1-minute", 60);

  await context.run("send-notification", async () => {
    return await sendNotification(orderId);
  });

  return { success: true, order: result };
});

Triggering workflows:

// app/actions.ts
"use server";
import { Client } from "@upstash/workflow";
import { instrumentWorkflowClient } from "@kubiks/otel-upstash-workflow";

const workflowClient = instrumentWorkflowClient(
  new Client({
    baseUrl: process.env.QSTASH_URL!,
    token: process.env.QSTASH_TOKEN!,
  })
);

export async function createOrder(orderId: string) {
  const result = await workflowClient.trigger({
    url: "https://your-app.vercel.app/api/workflow",
    body: { orderId },
  });

  return {
    workflowId: result.workflowId,
    runId: result.workflowRunId,
  };
}

Configuration Options

interface InstrumentationConfig {
  /**
   * Whether to capture step inputs/outputs in spans.
   * @default false
   */
  captureStepData?: boolean;

  /**
   * Maximum length of step input/output to capture.
   * Data longer than this will be truncated.
   * @default 1024
   */
  maxStepDataLength?: number;

  /**
   * Custom tracer name.
   * @default "@kubiks/otel-upstash-workflow"
   */
  tracerName?: string;
}

Best Practices

1. Step Data Capture: Only enable captureStepData in development or when debugging specific issues. Capturing step data can expose sensitive information and increase trace size.

2. Step Naming: Use descriptive step names that clearly indicate what the step does. This makes traces easier to understand.

3. Error Handling: The instrumentation automatically captures errors. Make sure your workflow handlers have proper error handling.

4. Idempotency: The instrumentation functions are idempotent — calling them multiple times on the same handler/client has no additional effect.

License

MIT