alikia2x/cvsa
1
0
Fork 0
Code

add: metrics for queue

Browse Source
This commit is contained in:
alikia2x (寒寒) 2025-10-30 01:46:34 +08:00
parent 0f22f19572
commit 125a30f0df
3 changed files with 22 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
packages/crawler/metrics/index.ts
View File

@ -20,6 +20,10 @@ export const jobCounter = meter.createCounter("job_count", {
description: "Number of executed BullMQ jobs"
});
export const queueJobsCounter = meter.createGauge("queue_jobs_count", {
description: "Number of jobs in specific BullMQ queue"
});
export const jobDuration = meter.createHistogram("job_duration", {
description: "Execution duration of BullMQ jobs in milliseconds",
unit: "ms"

12
packages/crawler/mq/exec/dispatchMilestoneSnapshots.ts
View File

@ -6,8 +6,10 @@ import { scheduleSnapshot } from "db/snapshotSchedule";
import logger from "@core/log";
import { HOUR, MINUTE, SECOND } from "@core/lib";
import { sql } from "@core/db/dbNew";
import { jobCounter, jobDuration } from "metrics";
export const dispatchMilestoneSnapshotsWorker = async (_job: Job) => {
const start = Date.now();
try {
const videos = await getVideosNearMilestone(sql);
for (const video of videos) {
@ -21,9 +23,17 @@ export const dispatchMilestoneSnapshotsWorker = async (_job: Job) => {
const delay = truncate(scheduledNextSnapshotDelay, minInterval, maxInterval);
const targetTime = now + delay;
await scheduleSnapshot(sql, aid, "milestone", targetTime);
logger.log(`Scheduled milestone snapshot for aid ${aid} in ${(delay / MINUTE).toFixed(2)} mins.`, "mq");
logger.log(
`Scheduled milestone snapshot for aid ${aid} in ${(delay / MINUTE).toFixed(2)} mins.`,
"mq"
);
}
} catch (e) {
logger.error(e as Error, "mq", "fn:dispatchMilestoneSnapshotsWorker");
} finally {
const duration = Date.now() - start;
jobCounter.add(1, { jobName: "dispatchMilestoneSnapshots" });
jobDuration.record(duration, { jobName: "dispatchMilestoneSnapshots" });
}
};

7
packages/crawler/mq/exec/getLatestVideos.ts
View File

@ -1,7 +1,14 @@
import { sql } from "@core/db/dbNew";
import { Job } from "bullmq";
import { queueJobsCounter } from "metrics";
import { SnapshotQueue } from "mq";
import { queueLatestVideos } from "mq/task/queueLatestVideo";
export const getLatestVideosWorker = async (_job: Job): Promise<void> => {
await queueLatestVideos(sql);
const counts = await SnapshotQueue.getJobCounts();
const waiting = counts?.waiting;
const prioritized = counts?.prioritized;
waiting && queueJobsCounter.record(waiting, { queueName: "SnapshotQueue", status: "waiting" });
prioritized && queueJobsCounter.record(prioritized, { queueName: "SnapshotQueue", status: "prioritized" });
};