佇列工作處理程序可觀測性

npm install telemetry-sh

import telemetry from "telemetry-sh";

telemetry.init("YOUR_API_KEY");

async function processJob(job) {
  const startedAtMs = Date.now();
  let terminalAttemptStartedAtMs = startedAtMs;
  let attemptCount = 0;

  try {
    const result = await runWithRetry(async () => {
      attemptCount += 1;
      terminalAttemptStartedAtMs = Date.now();
      return performJob(job);
    });

    await telemetry.log("background_job_completed", {
      event_id: crypto.randomUUID(),
      job_id: job.id,
      job_name: job.name,
      queue_name: job.queue,
      account_id: job.accountId,
      attempt_count: attemptCount,
      queue_wait_ms: startedAtMs - job.enqueuedAtMs,
      duration_ms: Date.now() - terminalAttemptStartedAtMs,
      total_elapsed_ms: Date.now() - startedAtMs,
      status: "success",
      release: process.env.APP_RELEASE ?? "unknown"
    });

    return result;
  } catch (error) {
    await telemetry.log("background_job_completed", {
      event_id: crypto.randomUUID(),
      job_id: job.id,
      job_name: job.name,
      queue_name: job.queue,
      account_id: job.accountId,
      attempt_count: attemptCount,
      queue_wait_ms: startedAtMs - job.enqueuedAtMs,
      duration_ms: Date.now() - terminalAttemptStartedAtMs,
      total_elapsed_ms: Date.now() - startedAtMs,
      status: "error",
      error_type: classifyJobError(error),
      release: process.env.APP_RELEASE ?? "unknown"
    });

    throw error;
  }
}

async function recordQueueSnapshot(queue) {
  const state = await queue.inspect();

  await telemetry.log("queue_snapshot", {
    event_id: crypto.randomUUID(),
    queue_name: queue.name,
    depth: state.waitingCount,
    active_workers: state.activeWorkers,
    oldest_wait_ms: state.oldestEnqueuedAtMs
      ? Date.now() - state.oldestEnqueuedAtMs
      : 0,
    release: process.env.APP_RELEASE ?? "unknown"
  });
}

SELECT
  job_name,
  COUNT(*) AS jobs,
  approx_percentile_cont(queue_wait_ms, 0.50) AS p50_wait_ms,
  approx_percentile_cont(queue_wait_ms, 0.95) AS p95_wait_ms,
  approx_percentile_cont(duration_ms, 0.95) AS p95_run_ms
FROM background_job_completed
WHERE timestamp_utc >= now() - INTERVAL '24 hours'
GROUP BY job_name
HAVING COUNT(*) >= 20
ORDER BY p95_wait_ms DESC;

SELECT
  job_name,
  COUNT(*) AS jobs,
  SUM(CASE WHEN attempt_count > 1 THEN 1 ELSE 0 END) AS retried_jobs,
  SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS permanent_failures,
  100.0 * SUM(CASE WHEN attempt_count > 1 THEN 1 ELSE 0 END)
    / NULLIF(COUNT(*), 0) AS retried_job_rate_pct,
  100.0 * SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END)
    / NULLIF(COUNT(*), 0) AS permanent_failure_rate_pct
FROM background_job_completed
WHERE timestamp_utc >= now() - INTERVAL '7 days'
GROUP BY job_name
ORDER BY permanent_failure_rate_pct DESC, retried_job_rate_pct DESC;