refactor service dir structure
This commit is contained in:
Alex Holliday committed
1 parent
0b4ec46f79
commit
06fa79e09d
28 files changed
+251
-42
No files matched your search
@@ -0,0 +1,317 @@
|
||||
const QUEUE_NAMES = ["uptime", "pagespeed", "hardware"];
|
||||
const SERVICE_NAME = "JobQueue";
|
||||
const HEALTH_CHECK_INTERVAL = 10 * 60 * 1000; // 10 minutes
|
||||
const QUEUE_LOOKUP = {
|
||||
hardware: "hardware",
|
||||
http: "uptime",
|
||||
ping: "uptime",
|
||||
port: "uptime",
|
||||
docker: "uptime",
|
||||
pagespeed: "pagespeed",
|
||||
};
|
||||
const getSchedulerId = (monitor) => `scheduler:${monitor.type}:${monitor._id}`;
|
||||
|
||||
class JobQueue {
|
||||
static SERVICE_NAME = SERVICE_NAME;
|
||||
|
||||
constructor({ db, jobQueueHelper, logger, stringService }) {
|
||||
this.db = db;
|
||||
this.jobQueueHelper = jobQueueHelper;
|
||||
this.stringService = stringService;
|
||||
this.logger = logger;
|
||||
this.queues = {};
|
||||
this.workers = [];
|
||||
}
|
||||
|
||||
static async create({ db, jobQueueHelper, logger, stringService }) {
|
||||
const instance = new JobQueue({ db, jobQueueHelper, logger, stringService });
|
||||
await instance.init();
|
||||
return instance;
|
||||
}
|
||||
|
||||
async init() {
|
||||
try {
|
||||
await this.initQueues();
|
||||
await this.initWorkers();
|
||||
const monitors = await this.db.getAllMonitors();
|
||||
await Promise.all(
|
||||
monitors
|
||||
.filter((monitor) => monitor.isActive)
|
||||
.map(async (monitor) => {
|
||||
try {
|
||||
await this.addJob(monitor._id, monitor);
|
||||
} catch (error) {
|
||||
this.logger.error({
|
||||
message: error.message,
|
||||
service: SERVICE_NAME,
|
||||
method: "initJobQueue",
|
||||
stack: error.stack,
|
||||
});
|
||||
}
|
||||
})
|
||||
);
|
||||
this.healthCheckInterval = setInterval(async () => {
|
||||
try {
|
||||
const queueHealthChecks = await this.checkQueueHealth();
|
||||
const queueIsStuck = queueHealthChecks.some((healthCheck) => healthCheck.stuck);
|
||||
if (queueIsStuck) {
|
||||
this.logger.warn({
|
||||
message: "Queue is stuck",
|
||||
service: SERVICE_NAME,
|
||||
method: "periodicHealthCheck",
|
||||
details: queueHealthChecks,
|
||||
});
|
||||
await this.flushQueues();
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error({
|
||||
message: error.message,
|
||||
service: SERVICE_NAME,
|
||||
method: "periodicHealthCheck",
|
||||
stack: error.stack,
|
||||
});
|
||||
}
|
||||
}, HEALTH_CHECK_INTERVAL);
|
||||
} catch (error) {
|
||||
this.logger.error({
|
||||
message: error.message,
|
||||
service: SERVICE_NAME,
|
||||
method: "initJobQueue",
|
||||
stack: error.stack,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async initQueues() {
|
||||
const readyPromises = [];
|
||||
|
||||
for (const queueName of QUEUE_NAMES) {
|
||||
const q = this.jobQueueHelper.createQueue(queueName);
|
||||
this.queues[queueName] = q;
|
||||
readyPromises.push(q.waitUntilReady());
|
||||
}
|
||||
await Promise.all(readyPromises);
|
||||
this.logger.info({
|
||||
message: "Queues ready",
|
||||
service: SERVICE_NAME,
|
||||
method: "initQueues",
|
||||
});
|
||||
}
|
||||
|
||||
async initWorkers() {
|
||||
const workerReadyPromises = [];
|
||||
|
||||
for (const queueName of QUEUE_NAMES) {
|
||||
const worker = this.jobQueueHelper.createWorker(queueName, this.queues[queueName]);
|
||||
this.workers.push(worker);
|
||||
workerReadyPromises.push(worker.waitUntilReady());
|
||||
}
|
||||
await Promise.all(workerReadyPromises);
|
||||
this.logger.info({
|
||||
message: "Workers ready",
|
||||
service: SERVICE_NAME,
|
||||
method: "initWorkers",
|
||||
});
|
||||
}
|
||||
|
||||
pauseJob = async (monitor) => {
|
||||
this.deleteJob(monitor);
|
||||
};
|
||||
|
||||
resumeJob = async (monitor) => {
|
||||
this.addJob(monitor._id, monitor);
|
||||
};
|
||||
|
||||
async addJob(jobName, monitor) {
|
||||
this.logger.info({
|
||||
message: `Adding job ${monitor?.url ?? "No URL"}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "addJob",
|
||||
});
|
||||
|
||||
const queueName = QUEUE_LOOKUP[monitor.type];
|
||||
const queue = this.queues[queueName];
|
||||
if (typeof queue === "undefined") {
|
||||
throw new Error(`Queue for ${monitor.type} not found`);
|
||||
}
|
||||
const jobTemplate = {
|
||||
name: jobName,
|
||||
data: monitor,
|
||||
opts: {
|
||||
attempts: 1,
|
||||
backoff: {
|
||||
type: "exponential",
|
||||
delay: 1000,
|
||||
},
|
||||
removeOnComplete: true,
|
||||
removeOnFail: false,
|
||||
timeout: 1 * 60 * 1000,
|
||||
},
|
||||
};
|
||||
|
||||
const schedulerId = getSchedulerId(monitor);
|
||||
await queue.upsertJobScheduler(schedulerId, { every: monitor?.interval ?? 60000 }, jobTemplate);
|
||||
}
|
||||
|
||||
async deleteJob(monitor) {
|
||||
try {
|
||||
const queue = this.queues[QUEUE_LOOKUP[monitor.type]];
|
||||
const schedulerId = getSchedulerId(monitor);
|
||||
const wasDeleted = await queue.removeJobScheduler(schedulerId);
|
||||
|
||||
if (wasDeleted === true) {
|
||||
this.logger.info({
|
||||
message: this.stringService.jobQueueDeleteJob,
|
||||
service: SERVICE_NAME,
|
||||
method: "deleteJob",
|
||||
details: `Deleted job ${monitor._id}`,
|
||||
});
|
||||
return true;
|
||||
} else {
|
||||
this.logger.error({
|
||||
message: this.stringService.jobQueueDeleteJob,
|
||||
service: SERVICE_NAME,
|
||||
method: "deleteJob",
|
||||
details: `Failed to delete job ${monitor._id}`,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
} catch (error) {
|
||||
error.service === undefined ? (error.service = SERVICE_NAME) : null;
|
||||
error.method === undefined ? (error.method = "deleteJob") : null;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async updateJob(monitor) {
|
||||
await this.deleteJob(monitor);
|
||||
await this.addJob(monitor._id, monitor);
|
||||
}
|
||||
|
||||
async getJobs() {
|
||||
try {
|
||||
let stats = {};
|
||||
await Promise.all(
|
||||
QUEUE_NAMES.map(async (name) => {
|
||||
const queue = this.queues[name];
|
||||
const jobs = await queue.getJobs();
|
||||
const ret = await Promise.all(
|
||||
jobs.map(async (job) => {
|
||||
const state = await job.getState();
|
||||
return { url: job.data.url, state, progress: job.progress };
|
||||
})
|
||||
);
|
||||
stats[name] = { jobs: ret };
|
||||
})
|
||||
);
|
||||
return stats;
|
||||
} catch (error) {
|
||||
error.service === undefined ? (error.service = SERVICE_NAME) : null;
|
||||
error.method === undefined ? (error.method = "getJobStats") : null;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async getMetrics() {
|
||||
try {
|
||||
let metrics = {};
|
||||
|
||||
await Promise.all(
|
||||
QUEUE_NAMES.map(async (name) => {
|
||||
const queue = this.queues[name];
|
||||
const [waiting, active, failed, delayed, repeatableJobs] = await Promise.all([
|
||||
queue.getWaitingCount(),
|
||||
queue.getActiveCount(),
|
||||
queue.getFailedCount(),
|
||||
queue.getDelayedCount(),
|
||||
queue.getRepeatableJobs(),
|
||||
]);
|
||||
|
||||
metrics[name] = {
|
||||
waiting,
|
||||
active,
|
||||
failed,
|
||||
delayed,
|
||||
repeatableJobs: repeatableJobs.length,
|
||||
};
|
||||
})
|
||||
);
|
||||
|
||||
return metrics;
|
||||
} catch (error) {
|
||||
this.logger.error({
|
||||
message: error.message,
|
||||
service: SERVICE_NAME,
|
||||
method: "getMetrics",
|
||||
stack: error.stack,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async checkQueueHealth() {
|
||||
const res = [];
|
||||
for (const queueName of QUEUE_NAMES) {
|
||||
const q = this.queues[queueName];
|
||||
await q.waitUntilReady();
|
||||
|
||||
const lastJobProcessedTime = q.lastJobProcessedTime;
|
||||
const currentTime = Date.now();
|
||||
const timeDiff = currentTime - lastJobProcessedTime;
|
||||
|
||||
// Check for jobs
|
||||
const jobCounts = await q.getJobCounts();
|
||||
const hasJobs = Object.values(jobCounts).some((count) => count > 0);
|
||||
|
||||
res.push({
|
||||
queueName,
|
||||
timeSinceLastJob: timeDiff,
|
||||
stuck: hasJobs && timeDiff > 10000,
|
||||
jobCounts,
|
||||
});
|
||||
}
|
||||
return res;
|
||||
}
|
||||
|
||||
async flushQueues() {
|
||||
try {
|
||||
this.logger.warn({
|
||||
message: "Flushing queues",
|
||||
method: "flushQueues",
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
for (const worker of this.workers) {
|
||||
await worker.close();
|
||||
}
|
||||
this.workers = [];
|
||||
|
||||
for (const queue of Object.values(this.queues)) {
|
||||
await queue.obliterate();
|
||||
}
|
||||
this.queue = {};
|
||||
await this.init();
|
||||
return true;
|
||||
} catch (error) {
|
||||
this.logger.warn({
|
||||
message: `${error.message} - Flushing redis manually`,
|
||||
service: SERVICE_NAME,
|
||||
method: "flushQueues",
|
||||
});
|
||||
return await this.jobQueueHelper.flushRedis();
|
||||
}
|
||||
}
|
||||
|
||||
async shutdown() {
|
||||
if (this.healthCheckInterval) {
|
||||
clearInterval(this.healthCheckInterval);
|
||||
this.healthCheckInterval = null;
|
||||
}
|
||||
for (const worker of this.workers) {
|
||||
await worker.close();
|
||||
}
|
||||
for (const queue of Object.values(this.queues)) {
|
||||
await queue.obliterate();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export default JobQueue;
|
||||
Reference in new issue
Block a user