@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.

!Upstash QStash Trace Visualization

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

Installation

``bash
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

$3

`ts
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.

$3

`ts
// 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.

$3

Optionally capture request/response bodies for debugging:

`ts
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

$3

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

$3

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

Span Attributes

$3

| 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 |

$3

| 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 |

$3

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

$3

`ts
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",
},
});
`

$3

`ts
// 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
});
`

$3

`ts
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",
});
`

$3

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

$3

`ts
// 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,
});
`

$3

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

`ts
// 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:

`ts
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)

$3

Publishing messages:
`ts
// 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:
`ts
// 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