diff --git a/server/utils/queue.js b/server/utils/queue.js index ffc4a741..2414ad70 100644 --- a/server/utils/queue.js +++ b/server/utils/queue.js @@ -1,10 +1,13 @@ // @flow import Queue from "bull"; import Redis from "ioredis"; +import { snakeCase } from "lodash"; import { client, subscriber } from "../redis"; +import * as metrics from "../utils/metrics"; export function createQueue(name: string) { - return new Queue(name, { + const prefix = `queue.${snakeCase(name)}`; + const queue = new Queue(name, { createClient(type) { switch (type) { case "client": @@ -16,4 +19,27 @@ export function createQueue(name: string) { } }, }); + + queue.on("stalled", () => { + metrics.increment(`${prefix}.jobs.stalled`); + }); + + queue.on("completed", () => { + metrics.increment(`${prefix}.jobs.completed`); + }); + + queue.on("error", () => { + metrics.increment(`${prefix}.jobs.errored`); + }); + + queue.on("failed", () => { + metrics.increment(`${prefix}.jobs.failed`); + }); + + setInterval(async () => { + metrics.gauge(`${prefix}.count`, await queue.count()); + metrics.gauge(`${prefix}.delayed_count`, await queue.getDelayedCount()); + }, 5 * 1000); + + return queue; }