drizzle-otel/packages/otel-upstash-queues/README.md

@kubiks/otel-upstash-queues

OpenTelemetry instrumentation for the Upstash QStash Node.js SDK. Capture spans for every QStash API call, enrich them with operation metadata, and keep an eye on message queue operations from your traces.

Visualize your message queue operations with detailed span information including message publishing, callbacks, and delivery tracking.

Installation

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

Peer Dependencies: @opentelemetry/api >= 1.9.0, @upstash/qstash >= 2.0.0

Quick Start

Publishing Messages

import { Client } from "@upstash/qstash";
import { instrumentUpstash } from "@kubiks/otel-upstash-queues";

const client = instrumentUpstash(
  new Client({ token: process.env.QSTASH_TOKEN! })
);

await client.publishJSON({
  url: "https://your-api-endpoint.com/process-image",
  body: { imageId: "123" },
});

instrumentUpstash wraps the QStash client instance you already use — no configuration changes needed. Every SDK call creates a client span with useful attributes.

Consuming Messages

// app/api/process/route.ts
import { verifySignatureAppRouter } from "@upstash/qstash/nextjs";
import { instrumentConsumer } from "@kubiks/otel-upstash-queues";

async function handler(request: Request) {
  const data = await request.json();
  
  // Process your message
  await processImage(data.imageId);
  
  return Response.json({ success: true });
}

// Instrument first, then verify signature
export const POST = verifySignatureAppRouter(instrumentConsumer(handler));

instrumentConsumer wraps your message handler to trace message consumption, creating a SERVER span for each message received and processed.

With Body Capture

Optionally capture request/response bodies for debugging:

const client = instrumentUpstash(
  new Client({ token: process.env.QSTASH_TOKEN! }),
  {
    captureBody: true,      // Enable body capture (default: false)
    maxBodyLength: 2048,    // Max characters to capture (default: 1024)
  }
);

What Gets Traced

This instrumentation provides two main functions:

1. instrumentUpstash - Wraps the QStash client to trace message publishing
2. instrumentConsumer - Wraps your message handler to trace message consumption

Publisher Instrumentation

The instrumentUpstash function wraps the client.publishJSON method, creating a span with SpanKind.CLIENT for each message publish operation.

Consumer Instrumentation

The instrumentConsumer function wraps your message handler, creating a span with SpanKind.SERVER for each message received and processed.

Span Attributes

Publisher Spans (instrumentUpstash)

Attribute	Description	Example
messaging.system	Constant value qstash	qstash
messaging.operation	Operation type	publish
qstash.resource	Resource name	messages
qstash.target	Full operation target	messages.publish
qstash.url	Target URL for the message	https://example.com/api/process
qstash.method	HTTP method (default: POST)	POST, PUT, GET
qstash.message_id	Message ID returned by QStash	msg_123
qstash.delay	Delay before processing (seconds or string)	60 or "1h"
qstash.not_before	Unix timestamp for earliest processing	1672531200
qstash.deduplication_id	Deduplication ID for idempotent operations	unique-id-123
qstash.retries	Number of retry attempts (max)	3
qstash.callback_url	Success callback URL	https://example.com/callback
qstash.failure_callback_url	Failure callback URL	https://example.com/failure

Consumer Spans (instrumentConsumer)

Attribute	Description	Example
messaging.system	Constant value qstash	qstash
messaging.operation	Operation type	receive
qstash.resource	Resource name	messages
qstash.target	Full operation target	messages.receive
qstash.message_id	Message ID from QStash	msg_456
qstash.retried	Number of times retried (actual count)	2
qstash.schedule_id	Schedule ID (if from scheduled message)	schedule_123
qstash.caller_ip	IP address of the caller	192.168.1.1
http.status_code	HTTP response status code	200

Body/Payload Attributes (Optional)

When captureBody is enabled in configuration:

Attribute	Description	Captured By
qstash.request.body	Request/message body content	Both publisher and consumer
qstash.response.body	Response body content	Consumer only

The instrumentation captures message metadata and configuration to help with debugging and monitoring. Body capture is disabled by default to protect sensitive data.

Usage Examples

Basic Message Publishing

import { Client } from "@upstash/qstash";
import { instrumentUpstash } from "@kubiks/otel-upstash-queues";

const client = instrumentUpstash(
  new Client({ token: process.env.QSTASH_TOKEN! })
);

// Publish a message
await client.publishJSON({
  url: "https://your-api.com/webhook",
  body: {
    userId: "user_123",
    action: "process_data",
  },
});

Delayed Message Publishing

// Delay message processing by 60 seconds
await client.publishJSON({
  url: "https://your-api.com/delayed-task",
  body: { taskId: "task_456" },
  delay: 60,
});

// Or use a human-readable delay
await client.publishJSON({
  url: "https://your-api.com/delayed-task",
  body: { taskId: "task_789" },
  delay: "1h", // 1 hour
});

Message with Callbacks

await client.publishJSON({
  url: "https://your-api.com/process",
  body: { orderId: "order_123" },
  callback: "https://your-api.com/success",
  failureCallback: "https://your-api.com/failure",
});

Message with Retries and Deduplication

await client.publishJSON({
  url: "https://your-api.com/critical-task",
  body: { taskId: "critical_123" },
  retries: 5,
  deduplicationId: "task-critical-123",
});

Scheduled Message

// Schedule for a specific time
const scheduledTime = Math.floor(Date.now() / 1000) + 3600; // 1 hour from now

await client.publishJSON({
  url: "https://your-api.com/scheduled-task",
  body: { reportId: "report_456" },
  notBefore: scheduledTime,
});

Message Consumer Instrumentation

Use instrumentConsumer to trace your message handler that receives QStash messages:

// app/api/process/route.ts
import { verifySignatureAppRouter } from "@upstash/qstash/nextjs";
import { instrumentConsumer } from "@kubiks/otel-upstash-queues";

async function handler(request: Request) {
  const data = await request.json();
  
  // Process your message
  await processImage(data.imageId);
  
  return Response.json({ success: true });
}

// Instrument first, then verify signature
export const POST = verifySignatureAppRouter(instrumentConsumer(handler));

With body capture:

export const POST = verifySignatureAppRouter(
  instrumentConsumer(handler, {
    captureBody: true,
    maxBodyLength: 2048,
  })
);

The instrumentConsumer function:

• Extracts QStash headers (message ID, retry count, schedule ID, caller IP)
• Creates a SERVER span for the message processing
• Tracks response status codes
• Captures errors during processing
• Optionally captures request and response bodies (when configured)

Complete Next.js Integration Example

Publishing messages:

// app/actions.ts
"use server";
import { Client } from "@upstash/qstash";
import { instrumentUpstash } from "@kubiks/otel-upstash-queues";

const qstashClient = instrumentUpstash(
  new Client({
    token: process.env.QSTASH_TOKEN!,
  })
);

export async function startBackgroundJob(imageId: string) {
  await qstashClient.publishJSON({
    url: "https://your-app.vercel.app/api/process",
    body: { imageId },
  });
}

Receiving messages:

// app/api/process/route.ts
import { verifySignatureAppRouter } from "@upstash/qstash/nextjs";
import { instrumentConsumer } from "@kubiks/otel-upstash-queues";

async function handler(request: Request) {
  const { imageId } = await request.json();
  
  // Your processing logic
  await processImage(imageId);
  
  return Response.json({ success: true });
}

export const POST = verifySignatureAppRouter(instrumentConsumer(handler));

License

MIT