diff --git a/backend/src/ee/services/gateway-v2/gateway-v2-service.ts b/backend/src/ee/services/gateway-v2/gateway-v2-service.ts index dce5bc299..a2d323790 100644 --- a/backend/src/ee/services/gateway-v2/gateway-v2-service.ts +++ b/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 }; }; diff --git a/backend/src/ee/services/relay/relay-service.ts b/backend/src/ee/services/relay/relay-service.ts index 41fe91bfc..d791e9919 100644 --- a/backend/src/ee/services/relay/relay-service.ts +++ b/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 }; }; diff --git a/backend/src/queue/queue-service.ts b/backend/src/queue/queue-service.ts index 335f25d5f..7f45e3821 100644 --- a/backend/src/queue/queue-service.ts +++ b/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 = [ diff --git a/backend/src/server/routes/index.ts b/backend/src/server/routes/index.ts index a718dd745..e95dd501c 100644 --- a/backend/src/server/routes/index.ts +++ b/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); diff --git a/backend/src/services/health-alert/health-alert-queue.ts b/backend/src/services/health-alert/health-alert-queue.ts new file mode 100644 index 000000000..1d65d0f4d --- /dev/null +++ b/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; + relayService: Pick; +}; + +export type THealthAlertServiceFactory = ReturnType; + +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( + 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 + }; +};