Merge pull request OneUptime#1651 from OneUptime/bull-board-ui

feat: Add Bull Board for monitoring and managing queues
yonasBSD · Aug 20, 2024 · abde6ee · abde6ee
2 parents 222027f + 9aae86c
commit abde6ee
Show file tree

Hide file tree

Showing 4 changed files with 161 additions and 1,061 deletions.
diff --git a/App/FeatureSet/Workers/Index.ts b/App/FeatureSet/Workers/Index.ts
@@ -58,7 +58,7 @@ import AnalyticsTableManagement from "./Utils/AnalyticsDatabase/TableManegement"
 import RunDatabaseMigrations from "./Utils/DataMigration";
 import JobDictionary from "./Utils/JobDictionary";
 import { PromiseVoidFunction } from "Common/Types/FunctionTypes";
-import { QueueJob, QueueName } from "Common/Server/Infrastructure/Queue";
+import Queue, { QueueJob, QueueName } from "Common/Server/Infrastructure/Queue";
 import QueueWorker from "Common/Server/Infrastructure/QueueWorker";
 import FeatureSet from "Common/Server/Types/FeatureSet";
 import logger from "Common/Server/Utils/Logger";
@@ -69,6 +69,10 @@ import "./Jobs/Probe/UpdateConnectionStatus";
 
 // Telemetry Monitors.
 import "./Jobs/TelemetryMonitor/MonitorTelemetryMonitor";
+import Express, { ExpressApplication } from "Common/Server/Utils/Express";
+import ClusterKeyAuthorization from "Common/Server/Middleware/ClusterKeyAuthorization";
+
+const app: ExpressApplication = Express.getExpressApp();
 
 const WorkersFeatureSet: FeatureSet = {
   init: async (): Promise<void> => {
@@ -99,6 +103,13 @@ const WorkersFeatureSet: FeatureSet = {
         },
         { concurrency: 100 },
       );
+
+      // attach bull board to the app
+      app.use(
+        Queue.getInspectorRoute(),
+        ClusterKeyAuthorization.isAuthorizedServiceMiddleware,
+        Queue.getQueueInspectorRouter(),
+      );
     } catch (err) {
       logger.error("App Init Failed:");
       logger.error(err);

diff --git a/Common/Server/Infrastructure/Queue.ts b/Common/Server/Infrastructure/Queue.ts
@@ -1,7 +1,16 @@
-import { RedisHostname, RedisPassword, RedisPort } from "../EnvironmentConfig";
+import {
+  ClusterKey,
+  RedisHostname,
+  RedisPassword,
+  RedisPort,
+} from "../EnvironmentConfig";
 import Dictionary from "Common/Types/Dictionary";
 import { JSONObject } from "Common/Types/JSON";
 import { Queue as BullQueue, Job, JobsOptions } from "bullmq";
+import { ExpressAdapter } from "@bull-board/express";
+import { createBullBoard } from "@bull-board/api";
+import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
+import { ExpressRouter } from "../Utils/Express";
 
 export enum QueueName {
   Workflow = "Workflow",
@@ -51,6 +60,32 @@ export default class Queue {
     await this.getQueue(queueName).removeRepeatableByKey(jobId);
   }
 
+  public static getInspectorRoute(): string {
+    return "/api/inspect/queue/:clusterKey";
+  }
+
+  public static getQueueInspectorRouter(): ExpressRouter {
+    const serverAdapter = new ExpressAdapter();
+
+    createBullBoard({
+      queues: [
+        ...Object.values(QueueName).map((queueName) => {
+          return new BullMQAdapter(this.getQueue(queueName));
+        }),
+      ],
+      serverAdapter: serverAdapter,
+    });
+
+    serverAdapter.setBasePath(
+      this.getInspectorRoute().replace(
+        "/:clusterKey",
+        "/" + ClusterKey.toString(),
+      ),
+    );
+
+    return serverAdapter.getRouter();
+  }
+
   public static async addJob(
     queueName: QueueName,
     jobId: string,