From 0694bd06d1b1a867de2359e8ec4907e12a9aa35a Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 5 May 2025 14:01:17 -0700 Subject: [PATCH] use redis service --- server/index.js | 29 +++++++++++++++++------------ server/service/jobQueue.js | 16 ++++++---------- server/service/redisService.js | 1 + 3 files changed, 24 insertions(+), 22 deletions(-) diff --git a/server/index.js b/server/index.js index 520e5cab2..f8a82edff 100755 --- a/server/index.js +++ b/server/index.js @@ -81,7 +81,9 @@ import ServiceRegistry from "./service/serviceRegistry.js"; import MongoDB from "./db/mongo/MongoDB.js"; +// Redis Service and dependencies import IORedis from "ioredis"; +import RedisService from "./service/redisService.js"; import TranslationService from "./service/translationService.js"; import languageMiddleware from "./middleware/languageMiddleware.js"; @@ -113,21 +115,16 @@ const shutdown = async () => { method: "shutdown", }); // flush Redis - const settings = - ServiceRegistry.get(SettingsService.SERVICE_NAME).getSettings() || {}; + const redisService = ServiceRegistry.get(RedisService.SERVICE_NAME); + await redisService.flushall(); - const { redisUrl } = settings; - const redis = new IORedis(redisUrl, { maxRetriesPerRequest: null }); - - logger.info({ message: "Flushing Redis" }); - await redis.flushall(); - logger.info({ message: "Redis flushed" }); process.exit(1); }, SHUTDOWN_TIMEOUT); try { server.close(); await ServiceRegistry.get(JobQueue.SERVICE_NAME).obliterate(); await ServiceRegistry.get(MongoDB.SERVICE_NAME).disconnect(); + await ServiceRegistry.get(RedisService.SERVICE_NAME).flushall(); logger.info({ message: "Graceful shutdown complete" }); process.exit(0); } catch (error) { @@ -185,7 +182,13 @@ const startApp = async () => { stringService ); - const jobQueue = new JobQueue( + const redisService = await RedisService.createInstance({ + logger, + IORedis, + SettingsService: settingsService, + }); + + const jobQueue = new JobQueue({ db, statusService, networkService, @@ -194,8 +197,9 @@ const startApp = async () => { stringService, logger, Queue, - Worker - ); + Worker, + redisService, + }); // Register services ServiceRegistry.register(JobQueue.SERVICE_NAME, jobQueue); @@ -207,6 +211,7 @@ const startApp = async () => { ServiceRegistry.register(StatusService.SERVICE_NAME, statusService); ServiceRegistry.register(NotificationService.SERVICE_NAME, notificationService); ServiceRegistry.register(TranslationService.SERVICE_NAME, translationService); + ServiceRegistry.register(RedisService.SERVICE_NAME, redisService); await translationService.initialize(); @@ -343,7 +348,7 @@ const startApp = async () => { app.use("/api/v1/queue", verifyJWT, queueRoutes.getRouter()); app.use("/api/v1/distributed-uptime", distributedUptimeRoutes.getRouter()); app.use("/api/v1/status-page", statusPageRoutes.getRouter()); - app.use("/api/v1/notifications", verifyJWT, notificationRoutes.getRouter()); + app.use("/api/v1/notifications", verifyJWT, notificationRoutes.getRouter()); app.use("/api/v1/diagnostic", verifyJWT, diagnosticRoutes.getRouter()); app.use("/api/v1/health", (req, res) => { res.json({ diff --git a/server/service/jobQueue.js b/server/service/jobQueue.js index 5674c7524..8f16f910d 100755 --- a/server/service/jobQueue.js +++ b/server/service/jobQueue.js @@ -1,5 +1,3 @@ -import IORedis from "ioredis"; - const QUEUE_NAMES = ["uptime", "pagespeed", "hardware", "distributed"]; const SERVICE_NAME = "JobQueue"; const JOBS_PER_WORKER = 5; @@ -18,7 +16,7 @@ const getSchedulerId = (monitor) => `scheduler:${monitor.type}:${monitor._id}`; class NewJobQueue { static SERVICE_NAME = SERVICE_NAME; - constructor( + constructor({ db, statusService, networkService, @@ -27,16 +25,14 @@ class NewJobQueue { stringService, logger, Queue, - Worker - ) { - const settings = settingsService.getSettings() || {}; - const { redisUrl } = settings; - const connection = new IORedis(redisUrl, { maxRetriesPerRequest: null }); + Worker, + redisService, + }) { + this.connection = redisService.getConnection(); this.queues = {}; this.workers = {}; this.lastJobProcessedTime = {}; - this.connection = connection; this.db = db; this.networkService = networkService; this.statusService = statusService; @@ -47,7 +43,7 @@ class NewJobQueue { this.stringService = stringService; QUEUE_NAMES.forEach((name) => { - const q = new Queue(name, { connection }); + const q = new Queue(name, { connection: this.connection }); this.lastJobProcessedTime[q.name] = Date.now(); q.on("error", (error) => { this.logger.error({ diff --git a/server/service/redisService.js b/server/service/redisService.js index 021011fe7..d2d76002c 100644 --- a/server/service/redisService.js +++ b/server/service/redisService.js @@ -31,6 +31,7 @@ class RedisService { return Math.min(times * 100, 2000); }, }); + await new Promise((resolve, reject) => { let errorOccurred = false;