diff --git a/Server/controllers/queueController.js b/Server/controllers/queueController.js index 813fa9808..cb3e132af 100644 --- a/Server/controllers/queueController.js +++ b/Server/controllers/queueController.js @@ -4,7 +4,7 @@ const SERVICE_NAME = "JobQueue"; const getJobs = async (req, res, next) => { try { - const jobs = await req.jobQueue.getJobs(); + const jobs = await req.jobQueue.getJobStats(); return res.status(200).json({ jobs }); } catch (error) { error.service = SERVICE_NAME; diff --git a/Server/index.js b/Server/index.js index e92a5bb45..1eb5d886b 100644 --- a/Server/index.js +++ b/Server/index.js @@ -19,6 +19,8 @@ const JobQueue = require("./service/jobQueue"); const EmailService = require("./service/emailService"); const PageSpeedService = require("./service/pageSpeedService"); +let cleaningUp = false; + // Need to wrap server setup in a function to handle async nature of JobQueue const startApp = async () => { // ************************** @@ -117,6 +119,11 @@ const startApp = async () => { const pageSpeedService = new PageSpeedService(); const cleanup = async () => { + if (cleaningUp) { + console.log("Already cleaning up"); + return; + } + cleaningUp = true; console.log("Shutting down gracefully"); await jobQueue.obliterate(); console.log("Finished cleanup"); diff --git a/Server/service/jobQueue.js b/Server/service/jobQueue.js index 3fdb91710..12c151abd 100644 --- a/Server/service/jobQueue.js +++ b/Server/service/jobQueue.js @@ -124,6 +124,8 @@ class JobQueue { const load = jobs.length / this.workers.length; return { jobs, load }; } catch (error) { + console.log(error); + throw error; } } @@ -143,8 +145,10 @@ class JobQueue { async scaleWorkers(workerStats) { if (this.workers.length === 0) { // There are no workers, need to add one - const worker = this.createWorker(); - this.workers.push(worker); + for (let i = 0; i < 5; i++) { + const worker = this.createWorker(); + this.workers.push(worker); + } return true; } @@ -168,15 +172,17 @@ class JobQueue { const excessCapacity = workerCapacity - workerStats.jobs.length; // Calculate how many workers to remove const workersToRemove = Math.floor(excessCapacity / JOBS_PER_WORKER); - for (let i = 0; i < workersToRemove; i++) { - const worker = this.workers.pop(); - try { - await worker.close(); - } catch (error) { - // Catch the error instead of throwing it - logger.error(errorMessages.JOB_QUEUE_WORKER_CLOSE, { - service: SERVICE_NAME, - }); + if (this.workers.length > 5) { + for (let i = 0; i < workersToRemove; i++) { + const worker = this.workers.pop(); + try { + await worker.close(); + } catch (error) { + // Catch the error instead of throwing it + logger.error(errorMessages.JOB_QUEUE_WORKER_CLOSE, { + service: SERVICE_NAME, + }); + } } } return true; @@ -196,6 +202,25 @@ class JobQueue { const jobs = await this.queue.getRepeatableJobs(); return jobs; } catch (error) { + console.log(error); + + throw error; + } + } + + async getJobStats() { + try { + const jobs = await this.queue.getJobs(); + const ret = await Promise.all( + jobs.map(async (job) => { + const state = await job.getState(); + return { url: job.data.url, state }; + }) + ); + return { jobs: ret, workers: this.workers.length }; + } catch (error) { + console.log(error); + throw error; } } @@ -210,6 +235,7 @@ class JobQueue { */ async addJob(jobName, payload) { try { + console.log("Adding job", payload.url); // Execute job immediately await this.queue.add(jobName, payload); @@ -221,6 +247,7 @@ class JobQueue { const workerStats = await this.getWorkerStats(); await this.scaleWorkers(workerStats); } catch (error) { + console.log(error); throw error; } } @@ -264,9 +291,12 @@ class JobQueue { await this.queue.removeRepeatableByKey(job.key); await this.queue.remove(job.id); } - this.workers.forEach(async (worker) => { - await worker.close(); - }); + await Promise.all( + this.workers.map(async (worker) => { + await worker.close(); + }) + ); + await this.queue.obliterate(); logger.info(successMessages.JOB_QUEUE_OBLITERATE, { service: SERVICE_NAME,