import { Command, Table } from "./deps.ts"; import { log } from "./deps.ts"; import * as wmill from "./gen/services.gen.ts"; import { pickInstance } from "./instance.ts"; type Data = { count: number; later: number; suspended: number; waiting: number; running: number; rps30s: string; rps5min: string; rps30min: string; rps24h: string; } type GlobalOptions = { instance?: string; baseUrl?: string; }; function createRow(tag: string, data: Record) { if (data[tag]) { return; } else { data[tag] = { count: 0, waiting: 0, later: 0, running: 0, suspended: 0, rps30s: "", rps5min: "", rps30min: "", rps24h: "", } } } async function displayQueues(opts: GlobalOptions, workspace?: string) { const activeInstance = await pickInstance(opts, true); if (activeInstance) { try { const queuedJobs = await wmill.listQueue({workspace: workspace ?? 'admins', allWorkspaces: workspace === undefined, perPage: 100000}); const jobCounts30s = await wmill.countJobsByTag({ horizonSecs: 30, workspaceId: workspace, }); const nowFromDb = new Date(await wmill.getDbClock()); const jobCounts5min = await wmill.countJobsByTag({ horizonSecs: 300, workspaceId: workspace, }); const jobCounts30min = await wmill.countJobsByTag({ horizonSecs: 1800, workspaceId: workspace, }); const jobCounts24h = await wmill.countJobsByTag({ horizonSecs: 86400, workspaceId: workspace, }); const data: Record = {} for (const job of queuedJobs) { createRow(job.tag, data); const scheduledFor = new Date(job.scheduled_for ?? ""); if (job.running) { if (job.suspend) { data[job.tag].suspended += 1; } else { data[job.tag].running += 1; } } else if (scheduledFor <= nowFromDb) { data[job.tag].waiting += 1; } else { data[job.tag].later += 1; } } for (const count of jobCounts30s) { const tag = count.tag; createRow(tag, data); data[tag].rps30s = (count.count / 30).toFixed(3); } for (const count of jobCounts5min) { const tag = count.tag; createRow(tag, data); data[tag].rps5min = (count.count / 300).toFixed(3); } for (const count of jobCounts30min) { const tag = count.tag; createRow(tag, data); data[tag].rps30min = (count.count / 1800).toFixed(3); } for (const count of jobCounts24h) { const tag = count.tag; createRow(tag, data); data[tag].rps24h = (count.count / 86400).toFixed(3); } const table = new Table(); table.header([ "", "Running", "Waiting", "Later", "Suspended", "RPS (30s)", "RPS (5min)", "RPS (30min)", "RPS (24h)", ]); let body = [] for (const tag in data) { body.push([tag, data[tag].running, data[tag].waiting, data[tag].later, data[tag].suspended, data[tag].rps30s, data[tag].rps5min, data[tag].rps30min, data[tag].rps24h]); } table.body(body).render(); } catch (error) { log.error("Failed to fetch queue metrics:", error); } } else { log.info("No active instance found"); log.info("Use 'wmill instance add' to add a new instance"); } } const command = new Command() .description("List all queues with their metrics") .arguments("[workspace:string] the optional workspace to filter by (default to all workspaces)") .option( "--instance [instance]", "Name of the instance to push to, override the active instance" ) .option( "--base-url [baseUrl]", "If used with --token, will be used as the base url for the instance" ) .action(displayQueues as any); export default command;