feat(node): Add Crons check-ins for BullMQ job schedulers

dev-packages/node-integration-tests/package.json

@@ -63,6 +63,7 @@
     "ai": "^4.3.16",
     "amqplib": "^0.10.9",
     "body-parser": "^2.3.0",
+    "bullmq": "5.81.5",
     "consola": "^3.2.3",
     "cors": "^2.8.5",
     "cron": "^3.1.6",

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:
+      - '6384: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/scenario.ts

@@ -0,0 +1,44 @@
+import * as Sentry from '@sentry/node';
+import { loggingTransport } from '@sentry-internal/node-integration-tests';
+import { Queue, Worker } from 'bullmq';
+
+Sentry.init({
+  dsn: 'https://public@dsn.ingest.sentry.io/1337',
+  release: '1.0',
+  transport: loggingTransport,
+});
+
+const connection = { host: 'localhost', port: 6384 };
+
+async function run(): Promise<void> {
+  const queue = new Queue('crons', { connection });
+  await queue.obliterate({ force: true });
+
+  const worker = new Worker(
+    'crons',
+    async job => {
+      if (job.name === 'failing') {
+        throw new Error('Error in scheduled job');
+      }
+    },
+    { connection },
+  );
+  Sentry.cron.instrumentBullMQWorker(worker);
+
+  await queue.add('plain', {});
+  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' },
+  );
+
+  setTimeout(async () => {
+    await worker.close();
+    await queue.obliterate({ force: true });
+    await queue.close();
+    process.exit();
+  }, 5_000);
+}
+
+void run();

dev-packages/node-integration-tests/suites/cron/bullmq/test.ts

@@ -0,0 +1,51 @@
+import { afterAll, expect, test } from 'vitest';
+import { cleanupChildProcesses, createRunner, describeWithDockerCompose } from '../../../utils/runner';
+
+describeWithDockerCompose('bullmq job scheduler check-ins', { workingDirectory: [__dirname] }, () => {
+  afterAll(() => {
+    cleanupChildProcesses();
+  });
+
+  test('sends check-ins with the job scheduler schedule', { timeout: 60_000 }, async () => {
+    await createRunner(__dirname, 'scenario.ts')
+      .unordered()
+      .expect({
+        check_in: {
+          check_in_id: expect.any(String),
+          monitor_slug: 'every-scheduler',
+          status: 'in_progress',
+          release: '1.0',
+          monitor_config: { schedule: { type: 'interval', value: 5, unit: 'minute' } },
+        },
+      })
+      .expect({
+        check_in: {
+          check_in_id: expect.any(String),
+          monitor_slug: 'every-scheduler',
+          status: 'ok',
+          release: '1.0',
+          duration: expect.any(Number),
+        },
+      })
+      .expect({
+        check_in: {
+          check_in_id: expect.any(String),
+          monitor_slug: 'pattern-scheduler',
+          status: 'in_progress',
+          release: '1.0',
+          monitor_config: { schedule: { type: 'crontab', value: '0 9 * * 1-5' }, timezone: 'Europe/Vienna' },
+        },
+      })
+      .expect({
+        check_in: {
+          check_in_id: expect.any(String),
+          monitor_slug: 'pattern-scheduler',
+          status: 'error',
+          release: '1.0',
+          duration: expect.any(Number),
+        },
+      })
+      .start()
+      .completed();
+  });
+});

packages/node/src/cron/bullmq.ts

@@ -0,0 +1,181 @@
+import { captureCheckIn, debug, type MonitorConfig, timestampInSeconds } from '@sentry/core';
+import { DEBUG_BUILD } from '../debug-build';
+import { replaceCronNames } from './common';
+
+interface BullMQRepeatOptions {
+  pattern?: unknown;
+  every?: unknown;
+  tz?: unknown;
+  limit?: unknown;
+  endDate?: unknown;
+}
+
+interface BullMQJob {
+  repeatJobKey?: unknown;
+  opts?: { repeat?: BullMQRepeatOptions };
+}
+
+export interface InstrumentBullMQWorkerOptions {
+  /** Job scheduler ids to not monitor. Strings match the id exactly. */
+  excludeJobSchedulers?: Array<string | RegExp>;
+}
+
+const CRON_PRESETS: Record<string, string> = {
+  '@yearly': '0 0 1 1 *',
+  '@annually': '0 0 1 1 *',
+  '@monthly': '0 0 1 * *',
+  '@weekly': '0 0 * * 0',
+  '@daily': '0 0 * * *',
+  '@hourly': '0 * * * *',
+};
+
+const FULL_DAY_OF_WEEK_RANGES = new Set(['0-6', '0-7', '1-7']);
+
+// Legacy repeatable jobs (`queue.add` with `repeat`) are keyed by a hash or `name:id:endDate:tz:pattern`.
+const LEGACY_REPEAT_KEY = /^[\da-f]{32,}$|^(?:[^:]*:){4}/;
+
+const instrumentedWorkers = new WeakSet<object>();
+
+/**
+ * Sends check-ins to Sentry for every job a BullMQ worker runs from a job scheduler.
+ * The job scheduler id is the monitor slug, and its `pattern` or `every` is sent as the monitor schedule.
+ *
+ * ```ts
+ * import * as Sentry from '@sentry/node';
+ * import { Worker } from 'bullmq';
+ *
+ * const worker = new Worker('reports', processor, { connection });
+ * Sentry.cron.instrumentBullMQWorker(worker, { excludeJobSchedulers: ['cache-warmup'] });
+ * ```
+ */
+export function instrumentBullMQWorker<T extends { on: (...args: never[]) => unknown }>(
+  worker: T,
+  options: InstrumentBullMQWorkerOptions = {},
+): T {
+  if (instrumentedWorkers.has(worker)) {
+    return worker;
+  }
+  instrumentedWorkers.add(worker);
+
+  const running = new WeakMap<object, { monitorSlug: string; checkInId: string; start: number }>();
+  const on = (event: string, listener: (job?: BullMQJob) => void): unknown =>
+    (worker.on as unknown as typeof on).call(worker, event, listener);
+
+  on('active', job => {
+    try {
+      const monitorSlug = getJobSchedulerId(job, options.excludeJobSchedulers);
+      if (job && monitorSlug) {
+        const checkInId = captureCheckIn({ monitorSlug, status: 'in_progress' }, getMonitorConfig(job.opts?.repeat));
+        running.set(job, { monitorSlug, checkInId, start: timestampInSeconds() });
+      }
+    } catch (e) {
+      DEBUG_BUILD && debug.error('[BullMQ] Failed to send check-in:', e);
+    }
+  });
+
+  const finish =
+    (status: 'ok' | 'error') =>
+    (job?: BullMQJob): void => {
+      try {
+        const run = job && running.get(job);
+        if (job && run) {
+          running.delete(job);
+          const { monitorSlug, checkInId, start } = run;
+          captureCheckIn({ monitorSlug, status, checkInId, duration: timestampInSeconds() - start });
+        }
+      } catch (e) {
+        DEBUG_BUILD && debug.error('[BullMQ] Failed to send check-in:', e);
+      }
+    };
+  on('completed', finish('ok'));
+  on('failed', finish('error'));
+
+  return worker;
+}
+
+function getJobSchedulerId(job: BullMQJob | undefined, exclude: Array<string | RegExp> = []): string | undefined {
+  const id = job?.repeatJobKey;
+  if (typeof id !== 'string' || !id || LEGACY_REPEAT_KEY.test(id)) {
+    return undefined;
+  }
+  return exclude.some(pattern => (typeof pattern === 'string' ? pattern === id : pattern.test(id))) ? undefined : id;
+}
+
+function getMonitorConfig(repeat: BullMQRepeatOptions | undefined): MonitorConfig | undefined {
+  // A scheduler that stops after `limit` runs or `endDate` would show up as missed.
+  if (!repeat || repeat.limit != null || repeat.endDate != null) {
+    return undefined;
+  }
+
+  if (typeof repeat.every === 'number') {
+    const minutes = repeat.every / 60_000;
+    if (!Number.isInteger(minutes) || minutes <= 0) {
+      return undefined;
+    }
+    if (minutes % 1440 === 0) {
+      return { schedule: { type: 'interval', value: minutes / 1440, unit: 'day' } };
+    }
+    if (minutes % 60 === 0) {
+      return { schedule: { type: 'interval', value: minutes / 60, unit: 'hour' } };
+    }
+    return { schedule: { type: 'interval', value: minutes, unit: 'minute' } };
+  }
+
+  const crontab = typeof repeat.pattern === 'string' ? toCrontab(repeat.pattern) : undefined;
+  // Without `tz`, cron-parser uses the server's time zone.
+  const timezone = toSentryTimeZone(typeof repeat.tz === 'string' && repeat.tz ? repeat.tz : getLocalTimeZone());
+  return crontab && timezone ? { schedule: { type: 'crontab', value: crontab }, timezone } : undefined;
+}
+
+/**
+ * Converts a cron-parser pattern to a crontab that Sentry reads the same way, if there is one.
+ */
+function toCrontab(pattern: string): string | undefined {
+  const preset = CRON_PRESETS[pattern.trim().toLowerCase()];
+  if (preset) {
+    return preset;
+  }
+
+  const fields = replaceCronNames(pattern).trim().split(/\s+/);
+  if (fields.length === 6 && /^0+$/.test(fields[0] as string)) {
+    fields.shift();
+  }
+  if (fields.length !== 5 || fields.some(field => !/^[\d*,/\-L#]+$/.test(field))) {
+    return undefined;
+  }
+
+  // With both day fields set, cron-parser runs on either, but Sentry needs both when one starts with `*`.
+  // cron-parser 4 also treats `1-31` as `*`.
+  const [, , dayOfMonth = '', , dayOfWeek = ''] = fields;
+  if (
+    dayOfMonth !== '*' &&
+    dayOfWeek !== '*' &&
+    (dayOfMonth.startsWith('*') ||
+      dayOfWeek.startsWith('*') ||
+      dayOfMonth === '1-31' ||
+      FULL_DAY_OF_WEEK_RANGES.has(dayOfWeek))
+  ) {
+    return undefined;
+  }
+
+  return fields.join(' ');
+}
+
+function toSentryTimeZone(timezone: string | undefined): string | undefined {
+  if (!timezone || timezone === 'Etc/Unknown' || /^(?:utc|gmt)?[+-]/i.test(timezone)) {
+    return undefined;
+  }
+  try {
+    return new Intl.DateTimeFormat('en-US', { timeZone: timezone }).resolvedOptions().timeZone || undefined;
+  } catch {
+    return undefined;
+  }
+}
+
+function getLocalTimeZone(): string | undefined {
+  try {
+    return Intl.DateTimeFormat().resolvedOptions().timeZone || undefined;
+  } catch {
+    return undefined;
+  }
+}

packages/node/src/cron/index.ts

@@ -1,9 +1,11 @@
+import { instrumentBullMQWorker } from './bullmq';
 import { instrumentCron } from './cron';
 import { instrumentNodeCron } from './node-cron';
 import { instrumentNodeSchedule } from './node-schedule';
 
 /** Methods to instrument cron libraries for Sentry check-ins */
 export const cron = {
+  instrumentBullMQWorker,
   instrumentCron,
   instrumentNodeCron,
   instrumentNodeSchedule,