Merge branch 'master' into tremor-graphs-refactor

Author: simlarsen

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> => {
@@ -97,7 +101,14 @@ const WorkersFeatureSet: FeatureSet = {
             await funcToRun();
           }
         },
-        { concurrency: 10 },
+        { concurrency: 100 },
+      );
+
+      // attach bull board to the app
+      app.use(
+        Queue.getInspectorRoute(),
+        ClusterKeyAuthorization.isAuthorizedServiceMiddleware,
+        Queue.getQueueInspectorRouter(),
       );
     } catch (err) {
       logger.error("App Init Failed:");

App/FeatureSet/Workers/Jobs/IncomingRequestMonitor/CheckHeartbeat.ts

@@ -17,7 +17,12 @@ RunCron(
   "IncomingRequestMonitor:CheckHeartbeat",
   { schedule: EVERY_THIRTY_SECONDS, runOnStartup: false },
   async () => {
-    logger.debug("Checking IncomingRequestMonitor:CheckHeartbeat");
+    logger.debug(
+      "Checking IncomingRequestMonitor:CheckHeartbeat at " +
+        OneUptimeDate.getDateAsLocalFormattedString(
+          OneUptimeDate.getCurrentDate(),
+        ),
+    );
 
     const newIncomingRequestMonitors: Array<Monitor> =
       await MonitorService.findBy({
@@ -87,6 +92,14 @@ RunCron(
 
     for (const monitor of totalIncomingRequestMonitors) {
       try {
+        logger.debug(
+          `Processing incoming request monitor: ${monitor.id?.toString()}`,
+        );
+
+        if (!monitor.monitorSteps) {
+          logger.debug("Monitor has no steps. Skipping...");
+          continue;
+        }
 
         await MonitorService.updateOneById({
           id: monitor.id!,
@@ -99,10 +112,9 @@ RunCron(
           },
         });
 
-        if (!monitor.monitorSteps) {
-          logger.debug("Monitor has no steps. Skipping...");
-          continue;
-        }
+        logger.debug(
+          `Updated incoming request monitor heartbeat checked at: ${monitor.id?.toString()}`,
+        );
 
         const processRequest: boolean = shouldProcessRequest(monitor);
 
@@ -124,7 +136,15 @@ RunCron(
           onlyCheckForIncomingRequestReceivedAt: true,
         };
 
+        logger.debug(
+          `Processing incoming request monitor: ${monitor.id?.toString()}`,
+        );
+
         await MonitorResourceUtil.monitorResource(incomingRequest);
+
+        logger.debug(
+          `Processed incoming request monitor: ${monitor.id?.toString()}`,
+        );
       } catch (error) {
         logger.error(
           `Error while processing incoming request monitor: ${monitor.id?.toString()}`,

App/Index.ts

@@ -9,7 +9,7 @@ import Workers from "./FeatureSet/Workers/Index";
 import Workflow from "./FeatureSet/Workflow/Index";
 import { PromiseVoidFunction } from "Common/Types/FunctionTypes";
 import { ClickhouseAppInstance } from "Common/Server/Infrastructure/ClickhouseDatabase";
-import { PostgresAppInstance } from "Common/Server/Infrastructure/PostgresDatabase";
+import PostgresAppInstance from "Common/Server/Infrastructure/PostgresDatabase";
 import Redis from "Common/Server/Infrastructure/Redis";
 import InfrastructureStatus from "Common/Server/Infrastructure/Status";
 import logger from "Common/Server/Utils/Logger";

Common/Server/Infrastructure/PostgresDatabase.ts

@@ -8,23 +8,23 @@ export type DatabaseSourceOptions = DataSourceOptions;
 export type DatabaseSource = DataSource;
 
 export default class Database {
-  protected dataSourceOptions: DataSourceOptions | null = null;
-  protected dataSource: DataSource | null = null;
+  protected static dataSourceOptions: DataSourceOptions | null = null;
+  protected static dataSource: DataSource | null = null;
 
-  public getDatasourceOptions(): DataSourceOptions {
+  public static getDatasourceOptions(): DataSourceOptions {
     this.dataSourceOptions = DatabaseDataSourceOptions;
     return this.dataSourceOptions;
   }
 
-  public getDataSource(): DataSource | null {
+  public static getDataSource(): DataSource | null {
     return this.dataSource;
   }
 
-  public isConnected(): boolean {
+  public static isConnected(): boolean {
     return Boolean(this.dataSource);
   }
 
-  public async connect(): Promise<DataSource> {
+  public static async connect(): Promise<DataSource> {
     let retry: number = 0;
 
     const dataSourceOptions: DataSourceOptions = this.getDatasourceOptions();
@@ -67,14 +67,14 @@ export default class Database {
     }
   }
 
-  public async disconnect(): Promise<void> {
+  public static async disconnect(): Promise<void> {
     if (this.dataSource) {
       await this.dataSource.destroy();
       this.dataSource = null;
     }
   }
 
-  public async checkConnnectionStatus(): Promise<boolean> {
+  public static async checkConnnectionStatus(): Promise<boolean> {
     // check popstgres connection to see if it is still alive
 
     try {
@@ -94,31 +94,29 @@ export default class Database {
     }
   }
 
-  public async dropDatabase(): Promise<void> {
+  public static async dropDatabase(): Promise<void> {
     await dropDatabase({
       options: this.getDatasourceOptions(),
     });
     this.dataSource = null;
     this.dataSourceOptions = null;
   }
 
-  public async createDatabase(): Promise<void> {
+  public static async createDatabase(): Promise<void> {
     await createDatabase({
       options: this.getDatasourceOptions(),
       ifNotExist: true,
     });
   }
 
-  public async createAndConnect(): Promise<void> {
+  public static async createAndConnect(): Promise<void> {
     await this.createDatabase();
     await this.connect();
   }
 
-  public async disconnectAndDropDatabase(): Promise<void> {
+  public static async disconnectAndDropDatabase(): Promise<void> {
     // Drop the database. Since this is the in-mem db, it will be destroyed.
     await this.disconnect();
     await this.dropDatabase();
   }
 }
-
-export const PostgresAppInstance: Database = new Database();

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,

Common/Server/Infrastructure/Status.ts

@@ -1,6 +1,6 @@
 // This class checks the status of all the datasources.
 import { ClickhouseAppInstance } from "./ClickhouseDatabase";
-import { PostgresAppInstance } from "./PostgresDatabase";
+import PostgresAppInstance from "./PostgresDatabase";
 import Redis from "./Redis";
 import DatabaseNotConnectedException from "Common/Types/Exception/DatabaseNotConnectedException";
 

Common/Server/Services/DatabaseService.ts

@@ -1,5 +1,5 @@
 import { AppApiHostname, EncryptionSecret } from "../EnvironmentConfig";
-import { PostgresAppInstance } from "../Infrastructure/PostgresDatabase";
+import PostgresAppInstance from "../Infrastructure/PostgresDatabase";
 import ClusterKeyAuthorization from "../Middleware/ClusterKeyAuthorization";
 import CountBy from "../Types/Database/CountBy";
 import CreateBy from "../Types/Database/CreateBy";

Common/Server/Utils/Telemetry.ts

@@ -5,7 +5,6 @@ import {
   Histogram,
   MetricOptions,
 } from "@opentelemetry/api/build/src/metrics/Metric";
-import { getNodeAutoInstrumentations } from "@opentelemetry/auto-instrumentations-node";
 import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-http";
 import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-proto";
 import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";
@@ -152,7 +151,11 @@ export default class Telemetry {
       const nodeSdkConfiguration: Partial<opentelemetry.NodeSDKConfiguration> =
         {
           idGenerator: new AWSXRayIdGenerator(),
-          instrumentations: hasHeaders ? [getNodeAutoInstrumentations()] : [],
+          instrumentations: hasHeaders
+            ? [
+                // Add instrumentations here
+              ]
+            : [],
           resource: this.getResource({
             serviceName: data.serviceName,
           }),

Common/Tests/Server/TestingUtils/__mocks__/TestDatabase.mock.ts

@@ -1,7 +1,6 @@
 import getTestDataSourceOptions from "../Postgres/TestDataSourceOptions";
-import {
+import PostgresAppInstance, {
   DatabaseSourceOptions,
-  PostgresAppInstance,
 } from "../../../../Server/Infrastructure/PostgresDatabase";
 import Redis from "../../../../Server/Infrastructure/Redis";
 import getTestRedisConnectionOptions from "../Redis/TestRedisOptions";