prehistoric-systems/omnivore
1
0
Fork 0
mirror of https://github.com/omnivore-app/omnivore.git synced 2026-03-11 08:54:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #3430 from omnivore-app/fix/queue-processor-custom-metrics

Browse source 
Expose custom metrics for the queue processor
This commit is contained in:
Jackson Harper 2024-01-25 21:17:49 +08:00 committed by GitHub
commit f3c1b9d9a0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 24 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
packages/api/src/queue-processor.ts
View file

@ -2,7 +2,7 @@
/* eslint-disable @typescript-eslint/restrict-template-expressions */
/* eslint-disable @typescript-eslint/require-await */
/* eslint-disable @typescript-eslint/no-misused-promises */
import { Job, QueueEvents, Worker, Queue } from 'bullmq'
import { Job, QueueEvents, Worker, Queue, JobType } from 'bullmq'
import express, { Express } from 'express'
import { SnakeNamingStrategy } from 'typeorm-naming-strategies'
import { appDataSource } from './data_source'
@ -63,6 +63,26 @@ const main = async () => {
// respond healthy to auto-scaler.
app.get('/_ah/health', (req, res) => res.sendStatus(200))
app.get('/metrics', async (_, res) => {
const queue = await getBackendQueue()
if (!queue) {
res.sendStatus(400)
return
}
let output = ''
const metrics: JobType[] = ['active', 'failed', 'completed', 'prioritized']
const counts = await queue.getJobCounts(...metrics)
console.log('counts: ', counts)
metrics.forEach((metric, idx) => {
output += `# TYPE omnivore_queue_messages_${metric} gauge\n`
output += `omnivore_queue_messages_${metric}{queue="${QUEUE_NAME}"} ${counts[metric]}\n`
})
res.status(200).setHeader('Content-Type', 'text/plain').send(output)
})
const server = app.listen(port, () => {
console.log(`[queue-processor]: started`)
})
@ -78,17 +98,14 @@ const main = async () => {
throw '[queue-processor] error redis is not initialized'
}
const queue = new Queue(QUEUE_NAME, {
connection: workerRedisClient,
})
const worker = new Worker(
QUEUE_NAME,
async (job: Job) => {
switch (job.name) {
case 'refresh-all-feeds': {
const counts = await queue.getJobCounts('wait')
if (counts.wait > 1000) {
const queue = await getBackendQueue()
const counts = await queue?.getJobCounts('prioritized')
if (counts && counts.wait > 1000) {
return
}
return await refreshAllFeeds(appDataSource)