Merge pull request #4661 from Infisical/fix-healthcheck-notif-job

move health alert to queue

backend/src/ee/services/gateway-v2/gateway-v2-service.ts

@@ -2,7 +2,6 @@ import net from "node:net";
 
 import { ForbiddenError } from "@casl/ability";
 import * as x509 from "@peculiar/x509";
-import { CronJob } from "cron";
 
 import { OrgMembershipRole, TRelays } from "@app/db/schemas";
 import { PgSqlLock } from "@app/keystore/keystore";
@@ -891,7 +890,7 @@ export const gatewayV2ServiceFactory = ({
     });
   };
 
-  const $healthcheckNotify = async () => {
+  const healthcheckNotify = async () => {
     const unhealthyGateways = await gatewayV2DAL.find({
       isHeartbeatStale: true
     });
@@ -945,18 +944,6 @@ export const gatewayV2ServiceFactory = ({
     }
   };
 
-  const initializeHealthcheckNotify = async () => {
-    logger.info("Setting up background notification process for gateway v2 health-checks");
-
-    await $healthcheckNotify();
-
-    // run every 5 minutes
-    const job = new CronJob("*/5 * * * *", $healthcheckNotify);
-    job.start();
-
-    return job;
-  };
-
   return {
     listGateways,
     registerGateway,
@@ -965,6 +952,6 @@ export const gatewayV2ServiceFactory = ({
     deleteGatewayById,
     heartbeat,
     getPamSessionKey,
-    initializeHealthcheckNotify
+    healthcheckNotify
   };
 };

backend/src/ee/services/relay/relay-service.ts

@@ -2,7 +2,6 @@ import { isIP } from "node:net";
 
 import { ForbiddenError } from "@casl/ability";
 import * as x509 from "@peculiar/x509";
-import { CronJob } from "cron";
 
 import { OrgMembershipRole, TRelays } from "@app/db/schemas";
 import { PgSqlLock } from "@app/keystore/keystore";
@@ -1209,7 +1208,7 @@ export const relayServiceFactory = ({
     return deletedRelay;
   };
 
-  const $healthcheckNotify = async () => {
+  const healthcheckNotify = async () => {
     const unhealthyRelays = await relayDAL.find({
       isHeartbeatStale: true
     });
@@ -1283,18 +1282,6 @@ export const relayServiceFactory = ({
     }
   };
 
-  const initializeHealthcheckNotify = async () => {
-    logger.info("Setting up background notification process for relay health-checks");
-
-    await $healthcheckNotify();
-
-    // run every 5 minutes
-    const job = new CronJob("*/5 * * * *", $healthcheckNotify);
-    job.start();
-
-    return job;
-  };
-
   return {
     registerRelay,
     getCredentialsForGateway,
@@ -1302,6 +1289,6 @@ export const relayServiceFactory = ({
     getRelays,
     deleteRelay,
     heartbeat,
-    initializeHealthcheckNotify
+    healthcheckNotify
   };
 };

backend/src/queue/queue-service.ts

@@ -76,7 +76,8 @@ export enum QueueName {
   TelemetryAggregatedEvents = "telemetry-aggregated-events",
   DailyReminders = "daily-reminders",
   SecretReminderMigration = "secret-reminder-migration",
-  UserNotification = "user-notification"
+  UserNotification = "user-notification",
+  HealthAlert = "health-alert"
 }
 
 export enum QueueJobs {
@@ -124,7 +125,8 @@ export enum QueueJobs {
   TelemetryAggregatedEvents = "telemetry-aggregated-events",
   DailyReminders = "daily-reminders",
   SecretReminderMigration = "secret-reminder-migration",
-  UserNotification = "user-notification-job"
+  UserNotification = "user-notification-job",
+  HealthAlert = "health-alert"
 }
 
 export type TQueueJobTypes = {
@@ -351,6 +353,10 @@ export type TQueueJobTypes = {
     name: QueueJobs.UserNotification;
     payload: { notifications: TCreateUserNotificationDTO[] };
   };
+  [QueueName.HealthAlert]: {
+    name: QueueJobs.HealthAlert;
+    payload: undefined;
+  };
 };
 
 const SECRET_SCANNING_JOBS = [

backend/src/server/routes/index.ts

@@ -186,6 +186,7 @@ import { folderTreeCheckpointDALFactory } from "@app/services/folder-tree-checkp
 import { folderTreeCheckpointResourcesDALFactory } from "@app/services/folder-tree-checkpoint-resources/folder-tree-checkpoint-resources-dal";
 import { groupProjectDALFactory } from "@app/services/group-project/group-project-dal";
 import { groupProjectServiceFactory } from "@app/services/group-project/group-project-service";
+import { healthAlertServiceFactory } from "@app/services/health-alert/health-alert-queue";
 import { identityDALFactory } from "@app/services/identity/identity-dal";
 import { identityMetadataDALFactory } from "@app/services/identity/identity-metadata-dal";
 import { identityOrgDALFactory } from "@app/services/identity/identity-org-dal";
@@ -1782,6 +1783,7 @@ export const registerRoutes = async (
     identityDAL
   });
 
+  // DAILY
   const dailyResourceCleanUp = dailyResourceCleanUpQueueServiceFactory({
     auditLogDAL,
     queueService,
@@ -1798,6 +1800,12 @@ export const registerRoutes = async (
     keyValueStoreDAL
   });
 
+  const healthAlert = healthAlertServiceFactory({
+    gatewayV2Service,
+    queueService,
+    relayService
+  });
+
   const dailyReminderQueueService = dailyReminderQueueServiceFactory({
     reminderService,
     queueService,
@@ -2210,6 +2218,7 @@ export const registerRoutes = async (
   await telemetryQueue.startTelemetryCheck();
   await telemetryQueue.startAggregatedEventsJob();
   await dailyResourceCleanUp.init();
+  await healthAlert.init();
   await pkiSyncCleanup.init();
   await dailyReminderQueueService.startDailyRemindersJob();
   await dailyReminderQueueService.startSecretReminderMigrationJob();
@@ -2370,16 +2379,6 @@ export const registerRoutes = async (
     cronJobs.push(configSyncJob);
   }
 
-  const gatewayHealthcheckNotifyJob = await gatewayV2Service.initializeHealthcheckNotify();
-  if (gatewayHealthcheckNotifyJob) {
-    cronJobs.push(gatewayHealthcheckNotifyJob);
-  }
-
-  const relayHealthcheckNotifyJob = await relayService.initializeHealthcheckNotify();
-  if (relayHealthcheckNotifyJob) {
-    cronJobs.push(relayHealthcheckNotifyJob);
-  }
-
   const oauthConfigSyncJob = await initializeOauthConfigSync();
   if (oauthConfigSyncJob) {
     cronJobs.push(oauthConfigSyncJob);

backend/src/services/health-alert/health-alert-queue.ts

@@ -0,0 +1,65 @@
+import { TGatewayV2ServiceFactory } from "@app/ee/services/gateway-v2/gateway-v2-service";
+import { TRelayServiceFactory } from "@app/ee/services/relay/relay-service";
+import { getConfig } from "@app/lib/config/env";
+import { logger } from "@app/lib/logger";
+import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
+
+type THealthAlertServiceFactoryDep = {
+  queueService: TQueueServiceFactory;
+  gatewayV2Service: Pick<TGatewayV2ServiceFactory, "healthcheckNotify">;
+  relayService: Pick<TRelayServiceFactory, "healthcheckNotify">;
+};
+
+export type THealthAlertServiceFactory = ReturnType<typeof healthAlertServiceFactory>;
+
+export const healthAlertServiceFactory = ({
+  queueService,
+  gatewayV2Service,
+  relayService
+}: THealthAlertServiceFactoryDep) => {
+  const appCfg = getConfig();
+
+  const init = async () => {
+    if (appCfg.isSecondaryInstance) {
+      return;
+    }
+
+    await queueService.stopRepeatableJob(
+      QueueName.HealthAlert,
+      QueueJobs.HealthAlert,
+      { pattern: "*/5 * * * *", utc: true },
+      QueueName.HealthAlert // job id
+    );
+
+    await queueService.startPg<QueueName.HealthAlert>(
+      QueueJobs.HealthAlert,
+      async () => {
+        try {
+          logger.info(`${QueueName.HealthAlert}: health check alert task started`);
+          await gatewayV2Service.healthcheckNotify();
+          await relayService.healthcheckNotify();
+          logger.info(`${QueueName.HealthAlert}: health check alert task completed`);
+        } catch (error) {
+          logger.error(error, `${QueueName.HealthAlert}: health check alert failed`);
+          throw error;
+        }
+      },
+      {
+        batchSize: 1,
+        workerCount: 1,
+        pollingIntervalSeconds: 60
+      }
+    );
+
+    await queueService.schedulePg(
+      QueueJobs.HealthAlert,
+      "*/5 * * * *", // Schedule to run every 5 minutes
+      undefined,
+      { tz: "UTC" }
+    );
+  };
+
+  return {
+    init
+  };
+};