Observability

Table of Contents

• Job Logs
• Job Counts and Metrics
• OpenTelemetry Integration
• AI Usage Telemetry
• @glidemq/dashboard

---

Job Logs

Append log lines from inside a processor using job.log(), then fetch them from any Queue instance.

import { Queue, Worker } from 'glide-mq';

// Inside the processor
const worker = new Worker(
  'tasks',
  async (job) => {
    await job.log('Starting step 1');
    await doStep1();
    await job.log('Step 1 done, starting step 2');
    await doStep2();
    return { done: true };
  },
  { connection },
);

// Fetching logs externally
const queue = new Queue('tasks', { connection });
const { logs, count } = await queue.getJobLogs(jobId);
// logs: ['Starting step 1', 'Step 1 done, starting step 2']
// count: 2

// Paginate logs for long-running jobs
const { logs: page1 } = await queue.getJobLogs(jobId, 0, 49); // first 50
const { logs: page2 } = await queue.getJobLogs(jobId, 50, 99); // next 50

---

Job Counts and Metrics

getJobCounts()

Returns counts for every job state:

const counts = await queue.getJobCounts();
// {
//   waiting:   12,
//   active:     3,
//   delayed:    5,
//   completed: 842,
//   failed:     7,
// }

getMetrics(type, opts?)

Returns aggregate count plus per-minute time-series data for completed or failed jobs:

const metrics = await queue.getMetrics('completed');
// {
//   count: 15234,
//   data: [
//     { timestamp: 1709654400000, count: 142, avgDuration: 234 },
//     { timestamp: 1709654460000, count: 156, avgDuration: 218 },
//   ],
//   meta: { resolution: 'minute' }
// }

// Slice data points (e.g. last 10 data points):
const recent = await queue.getMetrics('completed', { start: -10 });

Data points are recorded server-side with zero extra RTTs. Minute-resolution buckets are retained for 24 hours and trimmed automatically.

Disabling server-side telemetry

For maximum throughput, disable both events and metrics:

const worker = new Worker('tasks', handler, {
  connection,
  events: false, // skip XADD event emission per job
  metrics: false, // skip HINCRBY metrics recording per job
});

const queue = new Queue('tasks', { connection, events: false });

TS-side EventEmitter and OpenTelemetry spans are unaffected by these options.

count()

Returns the number of waiting jobs (stream length):

const waitingCount = await queue.count();

---

OpenTelemetry Integration

glide-mq emits OpenTelemetry spans automatically when @opentelemetry/api is installed. No code changes are required.

Setup

npm install @opentelemetry/api

Initialise your tracer provider before creating any Queue or Worker instances (standard OTel setup), then use glide-mq normally:

import { Queue } from 'glide-mq';

const queue = new Queue('tasks', { connection });
await queue.add('my-job', data);
// → creates span: glide-mq.queue.add
//     attributes: glide-mq.queue, glide-mq.job.name, glide-mq.job.id,
//                 glide-mq.job.delay, glide-mq.job.priority

Custom tracer

If you need to use a specific tracer instance instead of the global one:

import { setTracer, isTracingEnabled } from 'glide-mq';
import { trace } from '@opentelemetry/api';

const myTracer = trace.getTracer('my-service', '1.0.0');
setTracer(myTracer);

console.log('Tracing active:', isTracingEnabled());

Instrumented operations

Operation	Span name	Key attributes
queue.add()	glide-mq.queue.add	glide-mq.queue, glide-mq.job.name, glide-mq.job.id, glide-mq.job.delay, glide-mq.job.priority
flowProducer.add()	glide-mq.flow.add	glide-mq.queue, glide-mq.flow.name, glide-mq.flow.childCount

---

AI Usage Telemetry

reportUsage and persisted usage fields

When job.reportUsage() is called inside a processor, the usage metadata is persisted to the job hash. Usage data is stored as hash fields on the job key:

Hash field	Type	Description
usage:model	string	Model identifier (e.g. gpt-5.4)
usage:provider	string	Provider identifier (e.g. openai)
usage:tokens	string (JSON)	Token breakdown by category (e.g. {"input":100,"output":50})
usage:totalTokens	string (int)	Total tokens (auto-computed if not set)
usage:costs	string (JSON)	Cost breakdown (e.g. {"total":0.003})
usage:totalCost	string (float)	Total cost
usage:costUnit	string	Cost unit (e.g. "usd")
usage:latencyMs	string (int)	Inference latency in ms
usage:cached	string	true if response was cached

reportUsage() overwrites the previous usage payload for that job and updates the rolling per-minute usage buckets used by getUsageSummary().

Flow-level usage aggregation

queue.getFlowUsage(parentJobId) traverses the job tree and returns aggregated tokens and costs maps, totalTokens, totalCost, plus a models map (model name to job count), costUnit, and jobCount.

Rolling usage summary

Use queue.getUsageSummary() (or Queue.getUsageSummary()) to aggregate usage across queues without scanning job hashes:

const summary = await queue.getUsageSummary({
  queues: ['llm-tasks', 'embeddings'],
  windowMs: 3_600_000,
});

// {
//   totalTokens,
//   totalCost,
//   jobCount,
//   models: { 'gpt-5.4': 12 },
//   perQueue: { 'llm-tasks': { ... } }
// }

The proxy exposes the same rollup over HTTP at GET /usage/summary.

Budget exceeded events

When a flow budget is exceeded (via glidemq_recordUsageAndCheckBudget), the budget hash exceeded field is set to 1. Subsequent jobs in the flow are either failed or paused depending on the onExceeded setting. queue.updateFlowBudget(flowId, limits) changes the limits and re-evaluates the flag; paused jobs re-check the budget every 60 seconds.

Budget state can be queried via queue.getFlowBudget(flowId) which returns { maxTotalTokens, maxTokens, tokenWeights, maxTotalCost, maxCosts, costUnit, usedTokens, usedCost, exceeded, onExceeded }.

Token-per-minute (TPM) metrics

When tokenLimiter is configured on a worker, token consumption is tracked either locally (in-memory), in Valkey (glide:{queueName}:tpm), or both (default). This enables dual-axis monitoring: RPM (jobs per window) and TPM (tokens per window).

---

Proxy SSE Endpoints

When you want live observability from another process or language, the HTTP proxy exposes three SSE surfaces:

Path	Description
/queues/:name/events	Queue-wide lifecycle events with Last-Event-ID / ?lastId= replay
/queues/:name/jobs/:id/stream	Per-job output stream for job.stream() / job.streamChunk()
/broadcast/:name/events	Broadcast fan-out SSE stream with subscription and optional subjects filters

These endpoints require the proxy to be created with connection, because they allocate blocking readers internally.

---

@glidemq/dashboard

The @glidemq/dashboard package provides a web UI for inspecting queues in real time.

Quick start - built-in demo server

cd demo
npm install
npm run dashboard   # http://localhost:3000

REST API

Method	Path	Description
GET	/api/queues	List all queues with counts and metrics
GET	/api/queues/:name	Queue details + recent jobs
GET	/api/queues/:name/jobs/:id	Single job details, state, and logs
POST	/api/queues/:name/jobs	Add a new job
POST	/api/queues/:name/pause	Pause a queue
POST	/api/queues/:name/resume	Resume a queue
POST	/api/queues/:name/jobs/:id/retry	Retry a failed job
DELETE	/api/queues/:name/jobs/:id	Remove a job
POST	/api/queues/:name/drain	Drain all waiting jobs
POST	/api/queues/:name/obliterate	Obliterate queue and all data
GET	/api/events	SSE stream for real-time job events

Real-time events via SSE

const es = new EventSource('http://localhost:3000/api/events');
es.onmessage = ({ data }) => {
  const { queue, event, jobId } = JSON.parse(data);
  // event: 'added' | 'completed' | 'failed' | 'progress' | 'stalled' | 'heartbeat'
  console.log(`[${queue}] ${event}  -  job ${jobId}`);
};

Embedding in an existing Express app

import express from 'express';
import { Queue } from 'glide-mq';

const app = express();
app.use(express.json());

const queues: Record<string, Queue> = {
  orders: new Queue('orders', { connection }),
  payments: new Queue('payments', { connection }),
};

// Expose queue counts to the dashboard
app.get('/api/queues', async (_req, res) => {
  const data = await Promise.all(
    Object.entries(queues).map(async ([name, q]) => ({
      name,
      counts: await q.getJobCounts(),
      isPaused: await q.isPaused(),
    })),
  );
  res.json(data);
});