feat(backend): add Bull Board metrics history (#1604)

* feat(backend): add Bull Board metrics history

* docs: add Bull Board metrics changelog entry

* feedback

Author: brendan-kellam

packages/backend/package.json

@@ -22,9 +22,10 @@
         "vitest": "^4.1.4"
     },
     "dependencies": {
-        "@bull-board/api": "6.11.2",
-        "@bull-board/express": "6.11.2",
-        "@bull-board/ui": "6.11.2",
+        "@bull-board/api": "9.0.0",
+        "@bull-board/express": "9.0.0",
+        "@bull-board/metrics": "9.0.0",
+        "@bull-board/ui": "9.0.0",
         "@coderabbitai/bitbucket": "^1.1.3",
         "@gitbeaker/rest": "^40.5.1",
         "@octokit/app": "^16.1.1",

packages/backend/src/api.ts

@@ -1,12 +1,17 @@
 import { createBullBoard } from '@bull-board/api';
-import { BullMQAdapter } from '@bull-board/api/bullMQAdapter.js';
+import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
 import { ExpressAdapter } from '@bull-board/express';
+import {
+    MetricsRecorder,
+    RedisMetricsHistoryProvider,
+} from '@bull-board/metrics';
 import { Octokit } from '@octokit/rest';
 import * as Sentry from "@sentry/node";
 import { PrismaClient, RepoIndexingJobType } from '@sourcebot/db';
 import { createLogger, env, JOB_PRIORITIES } from '@sourcebot/shared';
 import express, { NextFunction, Request, Response } from 'express';
 import 'express-async-errors';
+import type { Redis } from 'ioredis';
 import * as http from "http";
 import z from 'zod';
 import { SINGLE_TENANT_ORG_ID } from './constants.js';
@@ -22,21 +27,41 @@ const PORT = Number(workerApiUrl.port) || (workerApiUrl.protocol === "https:" ?
 
 export class Api {
     private server: http.Server;
+    private metricsRecorder: MetricsRecorder;
+    private metricsHistoryProvider: RedisMetricsHistoryProvider;
 
     constructor(
         promClient: PromClient,
         private prisma: PrismaClient,
         private jobManager: JobManager,
+        redis: Redis,
     ) {
         const app = express();
         app.use(express.json());
         app.use(express.urlencoded({ extended: true }));
 
         const bullBoardAdapter = new ExpressAdapter();
         bullBoardAdapter.setBasePath('/admin/queues');
+        const queueAdapters = jobManager
+            .getQueues()
+            .map(queue => new BullMQAdapter(queue));
+        this.metricsHistoryProvider = new RedisMetricsHistoryProvider({
+            connection: redis,
+        });
+        this.metricsRecorder = new MetricsRecorder({
+            queues: queueAdapters,
+            connection: redis,
+            onLatencyError: (error, queueName) => {
+                logger.error(
+                    `Failed to record BullMQ latency metrics for queue "${queueName}"`,
+                    error,
+                );
+            },
+        });
         createBullBoard({
-            queues: jobManager.getQueues().map(queue => new BullMQAdapter(queue, { readOnlyMode: true })),
+            queues: queueAdapters,
             serverAdapter: bullBoardAdapter,
+            options: { historyProvider: this.metricsHistoryProvider },
         });
         app.use('/admin/queues', bullBoardAdapter.getRouter());
 
@@ -58,6 +83,7 @@ export class Api {
             logger.debug(`API server is running on port ${PORT}`);
             logger.debug(`Bull Board is available at ${workerApiUrl.origin}/admin/queues`);
         });
+        this.metricsRecorder.start();
     }
 
     private async experimental_addGithubRepo(req: Request, res: Response) {
@@ -125,6 +151,9 @@ export class Api {
     }
 
     public async dispose() {
+        this.metricsRecorder.stop();
+        this.metricsHistoryProvider.disconnect();
+
         return new Promise<void>((resolve, reject) => {
             this.server.close((err) => {
                 if (err) reject(err);

packages/backend/src/index.ts

@@ -84,7 +84,7 @@ jobManager.register(repoPermissionSyncWorkload);
 jobManager.register(attachmentPruneWorkload);
 jobManager.register(auditLogPruneWorkload);
 
-const api = new Api(promClient, prisma, jobManager);
+const api = new Api(promClient, prisma, jobManager, redis);
 
 await cleanupOrphanedRepoResources(prisma);
 
@@ -115,11 +115,11 @@ const listenToShutdownSignals = () => {
             logger.info(`Received ${signal}, cleaning up...`);
 
             await configManager.dispose()
+            await api.dispose();
             await jobManager.stop();
 
             await prisma.$disconnect();
             await redis.quit();
-            await api.dispose();
             await shutdownPosthog();
 
             logger.info('All workers shut down gracefully');

packages/backend/src/jobManager.test.ts

@@ -33,6 +33,7 @@ const mocks = vi.hoisted(() => {
         workers: [] as Array<{
             processor: (job: unknown) => Promise<unknown>;
             handlers: Map<string, (...args: unknown[]) => void>;
+            options: unknown;
         }>,
     };
 });
@@ -81,8 +82,9 @@ vi.mock("bullmq", () => ({
         constructor(
             _name: string,
             processor: (job: unknown) => Promise<unknown>,
+            options: unknown,
         ) {
-            this.record = { processor, handlers: new Map() };
+            this.record = { processor, handlers: new Map(), options };
             mocks.workers.push(this.record);
         }
 
@@ -92,6 +94,7 @@ vi.mock("bullmq", () => ({
 
         close = mocks.workerClose;
     },
+    MetricsTime: { ONE_WEEK: 10_080 },
 }));
 
 import { BullMQJobManager } from "./jobManager.js";
@@ -234,6 +237,17 @@ describe("BullMQJobManager lifecycle", () => {
         expect(mocks.producerClose).toHaveBeenCalledOnce();
     });
 
+    test("retains one week of native metrics for history recording", async () => {
+        const manager = new BullMQJobManager({} as Redis);
+        manager.register(createWorkload());
+
+        await manager.start();
+
+        expect(mocks.workers[0].options).toMatchObject({
+            metrics: { maxDataPoints: 10_080 },
+        });
+    });
+
     test("calls onStarted before processing and onCompleted after completion", async () => {
         const calls: string[] = [];
         const workload = createWorkload({

packages/backend/src/jobManager.ts

@@ -10,7 +10,7 @@ import {
     scheduleToMs,
     runWithJobLogContext,
 } from "@sourcebot/shared";
-import { Job, Queue, Worker } from "bullmq";
+import { Job, MetricsTime, Queue, Worker } from "bullmq";
 import { Redis } from "ioredis";
 import { WORKER_STOP_GRACEFUL_TIMEOUT_MS } from "./constants.js";
 import { createExecutionLockRunner } from "./executionLock.js";
@@ -220,6 +220,7 @@ export class BullMQJobManager implements JobManager {
                 connection: this.connection,
                 concurrency,
                 maxStalledCount: 1,
+                metrics: { maxDataPoints: MetricsTime.ONE_WEEK },
                 ...(rateLimit
                     ? {
                           limiter: {