feat(node): Add bullmqJobSchedulerIntegration to monitor BullMQ job schedulers

CHANGELOG.md

@@ -12,6 +12,17 @@ Work in this release was contributed by @kunal649. Thank you for your contributi
 
   `Sentry.cron.instrumentCron`, `instrumentNodeCron` and `instrumentNodeSchedule` read schedules the same way as `@SentryCron` and the cron integrations. Without a time zone, they send the server's time zone, where the scheduler runs the job, instead of letting Sentry assume UTC. For expressions Sentry would read differently than the scheduler, like per-second schedules, they send no monitor config, so existing monitors keep their schedule. They send names and nicknames like `MON` or `@daily` as numbers, and drop a fixed seconds field. If the seconds field isn't `0`, a check-in margin of 2 minutes is sent too, since the job checks in up to 59 seconds after the minute Sentry expects. This replaces a margin set in Sentry. Expressions with a day of month that none of their months have, like `0 9 31 2,4 *`, send no monitor config. `instrumentNodeSchedule` also sends `L` and `#`, where Sentry reads them the same way.
 
+- **feat(node): Add `bullmqJobSchedulerIntegration` to monitor BullMQ job schedulers ([#25472](https://github.com/getsentry/sentry-javascript/pull/25472))**
+
+  Add `Sentry.bullmqJobSchedulerIntegration()` to send check-ins for every job a BullMQ worker (5.16.0 to 6.x) runs from a job scheduler, with `<queue>-<scheduler id>` as the monitor slug and the scheduler's `pattern` or `every` as the schedule. Each monitored scheduler creates a monitor in Sentry, and monitors are billed. Use `excludeJobSchedulers` to skip schedulers by id. A retried job keeps one check-in.
+
+  ```js
+  Sentry.init({
+    dsn: '__DSN__',
+    integrations: [Sentry.bullmqJobSchedulerIntegration({ excludeJobSchedulers: ['cache-warmup'] })],
+  });
+  ```
+
 ## 11.12.0
 
 ### Important Changes

dev-packages/bun-integration-tests/node-suites/excludes.ts

@@ -4,13 +4,15 @@
 // `supports({ bunMin })`. Neither is listed here.
 
 // Node-only features: ANR and native thread watchdogs, child processes, the AWS Lambda Node runtime,
-// `node:sqlite`, which `flue` needs, and the Vercel keep-alive, which needs `http.server.response.finish`.
+// `node:sqlite`, which `flue` needs, the Vercel keep-alive, which needs `http.server.response.finish`, and
+// `bullmqJobSchedulerIntegration`, which `bun run` can't inject and the bundler plugins don't instrument.
 const NODE_ONLY = [
   'suites/anr/test.ts',
   'suites/aws-lambda-web/**',
   'suites/aws-serverless/**',
   'suites/breadcrumbs/**',
   'suites/child-process/test.ts',
+  'suites/cron/bullmq/test.ts',
   'suites/thread-blocked-native/test.ts',
   'suites/tracing/flue/test.ts',
   'suites/vercel/keep-alive/test.ts',

dev-packages/e2e-tests/test-applications/node-exports-test-app/scripts/consistentExports.ts

@@ -59,6 +59,7 @@ const DEPENDENTS: Dependent[] = [
       'childProcessIntegration',
       'workerThreadsIntegration',
       'pinoIntegration',
+      'bullmqJobSchedulerIntegration',
       // Bun will get its own runtime metrics integration
       'nodeRuntimeMetricsIntegration',
       'NodeRuntimeMetricsOptions',

dev-packages/node-integration-tests/suites/cron/bullmq/docker-compose.yml

@@ -0,0 +1,14 @@
+version: '3.9'
+
+services:
+  db:
+    image: redis:latest
+    restart: always
+    ports:
+      - '6385:6379'
+    healthcheck:
+      test: ['CMD-SHELL', 'redis-cli ping | grep -q PONG']
+      interval: 2s
+      timeout: 3s
+      retries: 30
+      start_period: 5s

dev-packages/node-integration-tests/suites/cron/bullmq/instrument-tracing.mjs

@@ -0,0 +1,10 @@
+import * as Sentry from '@sentry/node';
+import { loggingTransport } from '@sentry-internal/node-integration-tests';
+
+Sentry.init({
+  dsn: 'https://public@dsn.ingest.sentry.io/1337',
+  release: '1.0',
+  tracesSampleRate: 1.0,
+  transport: loggingTransport,
+  integrations: [Sentry.bullmqJobSchedulerIntegration()],
+});

dev-packages/node-integration-tests/suites/cron/bullmq/instrument.mjs

@@ -0,0 +1,9 @@
+import * as Sentry from '@sentry/node';
+import { loggingTransport } from '@sentry-internal/node-integration-tests';
+
+Sentry.init({
+  dsn: 'https://public@dsn.ingest.sentry.io/1337',
+  release: '1.0',
+  transport: loggingTransport,
+  integrations: [Sentry.bullmqJobSchedulerIntegration({ excludeJobSchedulers: ['excluded-scheduler'] })],
+});

dev-packages/node-integration-tests/suites/cron/bullmq/processor.cjs

@@ -0,0 +1 @@
+module.exports = async () => {};

dev-packages/node-integration-tests/suites/cron/bullmq/scenario-telemetry.mjs

@@ -0,0 +1,55 @@
+import * as Sentry from '@sentry/node';
+import { Queue, Worker } from 'bullmq';
+
+const connection = { host: 'localhost', port: 6385 };
+const telemetry = new Sentry.BullMQTelemetry();
+const autorun = process.env.AUTORUN === 'true';
+
+async function run() {
+  const queue = new Queue('crons-telemetry', { connection, telemetry });
+  await queue.obliterate({ force: true });
+
+  await queue.upsertJobScheduler('ok-scheduler', { every: 300_000 }, { name: 'ok-job' });
+  await queue.upsertJobScheduler(
+    'failing-scheduler',
+    { pattern: '0 0 9 * * 1-5', tz: 'Europe/Vienna', immediately: true },
+    { name: 'failing-job' },
+  );
+
+  const worker = new Worker(
+    'crons-telemetry',
+    async job => {
+      await Sentry.startSpan({ name: `work ${job.name}` }, async () => {
+        await new Promise(resolve => setTimeout(resolve, 50));
+      });
+      if (job.name === 'failing-job') {
+        throw new Error('Error in scheduled job');
+      }
+    },
+    { connection, telemetry, autorun },
+  );
+
+  const finished = new Promise(resolve => {
+    let count = 0;
+    const done = () => {
+      if (++count === 2) {
+        resolve();
+      }
+    };
+    worker.on('completed', done);
+    worker.on('failed', done);
+  });
+
+  if (!autorun) {
+    void worker.run();
+  }
+
+  await finished;
+  await worker.close();
+  await queue.obliterate({ force: true });
+  await queue.close();
+  await Sentry.flush();
+  process.exit();
+}
+
+void run();

dev-packages/node-integration-tests/suites/cron/bullmq/scenario.mjs

@@ -0,0 +1,101 @@
+import { dirname, join } from 'node:path';
+import { Queue, Worker } from 'bullmq';
+
+const connection = { host: 'localhost', port: 6385 };
+
+async function run() {
+  const queue = new Queue('crons', { connection });
+  const sandboxedQueue = new Queue('crons-sandboxed', { connection });
+  await queue.obliterate({ force: true });
+  await sandboxedQueue.obliterate({ force: true });
+
+  const worker = new Worker(
+    'crons',
+    async job => {
+      if (job.name === 'failing' || (job.name === 'flaky' && job.attemptsMade === 0)) {
+        throw new Error('Error in scheduled job');
+      }
+    },
+    { connection },
+  );
+  const sandboxedWorker = new Worker('crons-sandboxed', join(dirname(process.argv[1]), 'processor.cjs'), {
+    connection,
+  });
+
+  // Waits for every job to finish, including the ones that send no check-ins.
+  const pending = new Map([
+    ['plain', 1],
+    ['legacy', 1],
+    ['legacy-key', 1],
+    ['excluded', 1],
+    ['every', 2],
+    ['failing', 1],
+    ['flaky', 1],
+    ['sandboxed', 1],
+  ]);
+  const finished = new Promise(resolve => {
+    const done = job => {
+      const left = pending.get(job?.name);
+      if (left === undefined) {
+        return;
+      }
+      if (left > 1) {
+        pending.set(job.name, left - 1);
+      } else {
+        pending.delete(job.name);
+      }
+      if (!pending.size) {
+        resolve();
+      }
+    };
+    for (const w of [worker, sandboxedWorker]) {
+      w.on('completed', done);
+      w.on('failed', job => job?.finishedOn && done(job));
+    }
+  });
+
+  // Runs delayed scheduler jobs now. The loop below and the `completed` listener can race to promote the same job.
+  const promote = job => job?.promote().catch(() => {});
+
+  // Runs the iteration the worker created after the first run, instead of waiting 5 minutes.
+  let promoted = false;
+  worker.on('completed', async job => {
+    if (job.name !== 'every' || promoted) {
+      return;
+    }
+    promoted = true;
+    const delayed = await queue.getDelayed();
+    await promote(delayed.find(next => next.repeatJobKey === 'every-scheduler'));
+  });
+
+  await queue.add('plain', {});
+  await queue.add('legacy', {}, { repeat: { every: 300_000, immediately: true } });
+  await queue.add('legacy-key', {}, { repeat: { every: 300_000, key: 'custom-key', immediately: true } });
+  await queue.upsertJobScheduler('excluded-scheduler', { every: 300_000 }, { name: 'excluded' });
+  await queue.upsertJobScheduler('every-scheduler', { every: 300_000 }, { name: 'every' });
+  await queue.upsertJobScheduler(
+    'pattern-scheduler',
+    { pattern: '0 0 9 * * 1-5', tz: 'Europe/Vienna', immediately: true },
+    { name: 'failing' },
+  );
+  await queue.upsertJobScheduler('retry-scheduler', { every: 600_000 }, { name: 'flaky', opts: { attempts: 2 } });
+  await sandboxedQueue.upsertJobScheduler('sandboxed-scheduler', { every: 300_000 }, { name: 'sandboxed' });
+
+  // BullMQ 5.16 starts an `every` scheduler at its next full interval, so run the first jobs now.
+  for (const job of [...(await queue.getDelayed()), ...(await sandboxedQueue.getDelayed())]) {
+    if (job.opts.repeat?.count === 1) {
+      await promote(job);
+    }
+  }
+
+  await finished;
+  await worker.close();
+  await sandboxedWorker.close();
+  await queue.obliterate({ force: true });
+  await sandboxedQueue.obliterate({ force: true });
+  await queue.close();
+  await sandboxedQueue.close();
+  process.exit();
+}
+
+void run();