feat(node): Add Crons monitoring for BullMQ job schedulers

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/scenario-telemetry.mjs

@@ -0,0 +1,64 @@
+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',
+  tracesSampleRate: 1.0,
+  transport: loggingTransport,
+});
+
+const connection = { host: 'localhost', port: 6385 };
+const telemetry = new Sentry.BullMQTelemetry();
+const instrumentBeforeRun = process.env.INSTRUMENT_BEFORE_RUN === '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: !instrumentBeforeRun },
+  );
+  Sentry.cron.instrumentBullMQWorker(worker);
+
+  const finished = new Promise(resolve => {
+    let count = 0;
+    const done = () => {
+      if (++count === 2) {
+        resolve();
+      }
+    };
+    worker.on('completed', done);
+    worker.on('failed', done);
+  });
+
+  if (instrumentBeforeRun) {
+    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.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: 6385 };
+
+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,118 @@
+import type { Event, SerializedCheckIn } from '@sentry/core';
+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();
+  });
+
+  test.each([
+    ['after the worker starts', 'false'],
+    ['before the worker runs', 'true'],
+  ])(
+    'sends one check-in pair per run in the job trace with BullMQTelemetry, instrumented %s',
+    { timeout: 60_000 },
+    async (_, instrumentBeforeRun) => {
+      const checkIns: SerializedCheckIn[] = [];
+      const errors: Event[] = [];
+      const runner = createRunner(__dirname, 'scenario-telemetry.mjs')
+        .withEnv({ INSTRUMENT_BEFORE_RUN: instrumentBeforeRun })
+        .unordered()
+        .expectN(4, { check_in: checkIn => void checkIns.push(checkIn) })
+        .expect({ event: event => void errors.push(event) });
+      const spansByJob = ['ok-job', 'failing-job'].map(jobName =>
+        runner.collectStreamedSpans(spans =>
+          spans.some(
+            span =>
+              span.is_segment &&
+              span.name === 'process crons-telemetry' &&
+              span.attributes['messaging.bullmq.job.name']?.value === jobName,
+          ),
+        ),
+      );
+
+      await runner.start().completed();
+
+      const [okSpans, failingSpans] = await Promise.all(spansByJob);
+      const okSegment = okSpans!.find(span => span.is_segment)!;
+      const failingSegment = failingSpans!.find(span => span.is_segment)!;
+
+      expect(okSegment.status).toBe('ok');
+      expect(failingSegment.status).toBe('error');
+      expect(okSpans!.some(span => span.name === 'work ok-job' && span.parent_span_id === okSegment.span_id)).toBe(
+        true,
+      );
+      expect(
+        failingSpans!.some(span => span.name === 'work failing-job' && span.parent_span_id === failingSegment.span_id),
+      ).toBe(true);
+
+      const summary = (slug: string): unknown =>
+        checkIns
+          .filter(checkIn => checkIn.monitor_slug === slug)
+          .map(checkIn => ({
+            status: checkIn.status,
+            check_in_id: checkIn.check_in_id,
+            trace_id: checkIn.contexts?.trace?.trace_id,
+          }));
+      const okCheckInId = checkIns.find(checkIn => checkIn.monitor_slug === 'ok-scheduler')?.check_in_id;
+      const failingCheckInId = checkIns.find(checkIn => checkIn.monitor_slug === 'failing-scheduler')?.check_in_id;
+
+      expect(summary('ok-scheduler')).toEqual([
+        { status: 'in_progress', check_in_id: okCheckInId, trace_id: okSegment.trace_id },
+        { status: 'ok', check_in_id: okCheckInId, trace_id: okSegment.trace_id },
+      ]);
+      expect(summary('failing-scheduler')).toEqual([
+        { status: 'in_progress', check_in_id: failingCheckInId, trace_id: failingSegment.trace_id },
+        { status: 'error', check_in_id: failingCheckInId, trace_id: failingSegment.trace_id },
+      ]);
+
+      expect(errors).toHaveLength(1);
+      expect(errors[0]?.exception?.values?.[0]?.value).toBe('Error in scheduled job');
+      expect(errors[0]?.contexts?.trace?.trace_id).toBe(failingSegment.trace_id);
+    },
+  );
+});