fix(scheduler): отделить результат запуска от сбоев журналирования
This commit is contained in:
1 parent
49b8a1a19c
commit
63b2f75012
9 files changed
+317
-129
No files matched your search
@@ -24,9 +24,10 @@
|
||||
"test:recursive-routes": "tsx src/routes/recursive-routes.test.ts && tsx src/services/recursive-route-catalog.test.ts",
|
||||
"test:config-sync": "tsx src/services/config-apply-plan.test.ts && tsx src/services/entity-snapshots.test.ts",
|
||||
"test:ros-file": "tsx src/services/ros-file-contents.test.ts",
|
||||
"test:scheduler": "tsx src/services/scheduler-recorded-run.test.ts && tsx src/routes/scheduler.test.ts",
|
||||
"test:backups": "tsx src/services/backup-scheduler-run.test.ts && tsx src/services/backup-restore.test.ts && tsx src/services/backup-deletion.test.ts && tsx src/services/backup-dto.test.ts && tsx src/services/backup-storage-update.test.ts && tsx src/services/backup-job-execution.test.ts && tsx src/routes/backups-operations.test.ts && tsx src/services/s3-backup-client.test.ts",
|
||||
"test:live-maps": "tsx src/services/ospf-route-parse.test.ts && tsx src/services/vxlan-live.test.ts && tsx src/services/containers-live.test.ts && tsx src/services/firewall-poll.test.ts && tsx src/services/gre-catalog-poll.test.ts && tsx src/routes/filters-gre-catalog.test.ts && tsx src/routes/bgp-catalog.test.ts && tsx src/services/bgp-catalog-parse.test.ts && tsx src/routes/ospf-catalog.test.ts && tsx src/services/ospf-observed-values.test.ts",
|
||||
"test": "npm run test:probes && npm run test:exec && npm run test:alert-engine && npm run test:auth && npm run test:wireguard && npm run test:traffic-rate && npm run test:traffic-flow && npm run test:users && npm run test:pg && npm run test:backups && npm run test:live-maps && npm run test:ospf-optimizer && npm run test:recursive-routes && npm run test:config-sync && npm run test:ros-file && npm run test:containers && npm run test:certificates",
|
||||
"test": "npm run test:probes && npm run test:exec && npm run test:alert-engine && npm run test:auth && npm run test:wireguard && npm run test:traffic-rate && npm run test:traffic-flow && npm run test:users && npm run test:pg && npm run test:backups && npm run test:scheduler && npm run test:live-maps && npm run test:ospf-optimizer && npm run test:recursive-routes && npm run test:config-sync && npm run test:ros-file && npm run test:containers && npm run test:certificates",
|
||||
"test:geoip": "tsx src/services/traffic-flow-geoip.test.ts",
|
||||
"test:ospf-optimizer": "tsx src/services/ospf-optimizer-identity.test.ts && tsx src/services/ospf-optimizer-metrics.test.ts && tsx src/services/ospf-optimizer-plan.test.ts && tsx src/routes/ospf-optimizer.test.ts",
|
||||
"test:containers": "tsx src/services/containers-state.test.ts && tsx src/routes/containers-state.test.ts",
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { test } from "node:test"
|
||||
import Fastify from "fastify"
|
||||
import { createSchedulerRoutes } from "./scheduler.js"
|
||||
import type { SchedulerRunReceipt } from "../services/scheduler-recorded-run.js"
|
||||
const receipt: SchedulerRunReceipt = { status: "ok", error: null, result: { v: 1, job: "backups", sampledAt: "2026-10-01T00:00:00Z", due: true, created: 1, failures: 0, pruned: 0 } }
|
||||
async function setup(patch: Partial<Parameters<typeof createSchedulerRoutes>[0]> = {}) {
|
||||
const app = Fastify(); let calls = 0
|
||||
await app.register(createSchedulerRoutes({ status: async () => ({ jobs: [] }), runs: async () => ({ runs: [] }), refresh: async () => {}, run: async () => { calls++; return receipt }, ...patch }))
|
||||
return { app, calls: () => calls }
|
||||
}
|
||||
test("manual partial receipt is not ok and retains confirmed backup result", async () => {
|
||||
const { app } = await setup({ run: async () => ({ ...receipt, status: "error", error: "S3 unavailable" }) })
|
||||
try { const response = await app.inject({ method: "POST", url: "/scheduler/jobs/backups/run-now" }); assert.equal(response.statusCode, 200); assert.equal(response.json().ok, false); assert.equal(response.json().result.result.created, 1); assert.equal(response.json().result.error, "S3 unavailable") } finally { await app.close() }
|
||||
})
|
||||
test("status refresh failure never relabels confirmed execution or runs the job twice", async () => {
|
||||
const { app, calls } = await setup({ status: async () => { throw Error("status unavailable") } })
|
||||
try { const response = await app.inject({ method: "POST", url: "/scheduler/jobs/backups/run-now" }); assert.equal(response.statusCode, 200); assert.equal(response.json().ok, true); assert.equal(response.json().status, null); assert.equal(response.json().statusError, "status unavailable"); assert.equal(response.json().result.result.created, 1); assert.equal(calls(), 1) } finally { await app.close() }
|
||||
})
|
||||
test("invalid key and running conflict preserve their existing HTTP contracts", async () => {
|
||||
const { app, calls } = await setup()
|
||||
try { assert.equal((await app.inject({ method: "POST", url: "/scheduler/jobs/invalid/run-now" })).statusCode, 400); assert.equal(calls(), 0) } finally { await app.close() }
|
||||
const conflict = await setup({ run: async () => { throw Object.assign(Error("running"), { statusCode: 409 }) } })
|
||||
try { assert.equal((await conflict.app.inject({ method: "POST", url: "/scheduler/jobs/backups/run-now" })).statusCode, 409) } finally { await conflict.app.close() }
|
||||
})
|
||||
test("collector or journal rejection remains explicit without a fabricated receipt", async () => {
|
||||
const { app } = await setup({ run: async () => { throw Error("journal not confirmed; retry may repeat operations") } })
|
||||
try { const response = await app.inject({ method: "POST", url: "/scheduler/jobs/backups/run-now" }); assert.equal(response.statusCode, 500); assert.match(response.json().error, /retry may repeat/); assert.equal(response.json().result, undefined) } finally { await app.close() }
|
||||
})
|
||||
@@ -12,9 +12,15 @@ function parseJobKey(raw: string | undefined): typeof JOB_KEYS[number] | null {
|
||||
return (JOB_KEYS as readonly string[]).includes(s) ? (s as typeof JOB_KEYS[number]) : null
|
||||
}
|
||||
|
||||
const schedulerRoutes: FastifyPluginAsyncZod = async (app) => {
|
||||
export function createSchedulerRoutes(deps: {
|
||||
status: typeof getSchedulerStatus
|
||||
runs: typeof listSchedulerRuns
|
||||
refresh: typeof refreshScheduler
|
||||
run: typeof runSchedulerJobNow
|
||||
}): FastifyPluginAsyncZod {
|
||||
return async (app) => {
|
||||
app.get("/scheduler/status", async (_req, reply) => {
|
||||
return reply.send(await getSchedulerStatus())
|
||||
return reply.send(await deps.status())
|
||||
})
|
||||
|
||||
app.get("/scheduler/runs", async (req, reply) => {
|
||||
@@ -25,20 +31,24 @@ const schedulerRoutes: FastifyPluginAsyncZod = async (app) => {
|
||||
}
|
||||
const limit = Math.min(200, Math.max(1, Number.parseInt(String(q.limit ?? "50"), 10) || 50))
|
||||
const offset = Math.max(0, Number.parseInt(String(q.offset ?? "0"), 10) || 0)
|
||||
return reply.send(await listSchedulerRuns({ jobKey: jobKey ?? undefined, limit, offset }))
|
||||
return reply.send(await deps.runs({ jobKey: jobKey ?? undefined, limit, offset }))
|
||||
})
|
||||
|
||||
app.post("/scheduler/refresh", async (_req, reply) => {
|
||||
await refreshScheduler()
|
||||
return reply.send({ ok: true, status: await getSchedulerStatus() })
|
||||
await deps.refresh()
|
||||
return reply.send({ ok: true, status: await deps.status() })
|
||||
})
|
||||
|
||||
app.post("/scheduler/jobs/:jobKey/run-now", async (req, reply) => {
|
||||
const jobKey = parseJobKey(String((req.params as { jobKey?: string }).jobKey ?? ""))
|
||||
if (!jobKey) return reply.status(400).send({ error: "Некорректный jobKey" })
|
||||
try {
|
||||
await runSchedulerJobNow(jobKey)
|
||||
return reply.send({ ok: true, status: await getSchedulerStatus() })
|
||||
const result = await deps.run(jobKey)
|
||||
let status: Awaited<ReturnType<typeof getSchedulerStatus>> | null = null
|
||||
let statusError: string | undefined
|
||||
try { status = await deps.status() }
|
||||
catch (error) { statusError = error instanceof Error ? error.message : String(error) }
|
||||
return reply.send({ ok: result.status === "ok", result, status, ...(statusError ? { statusError } : {}) })
|
||||
} catch (e) {
|
||||
const statusCode = (e as Error & { statusCode?: number }).statusCode
|
||||
if (statusCode === 409) {
|
||||
@@ -50,4 +60,6 @@ const schedulerRoutes: FastifyPluginAsyncZod = async (app) => {
|
||||
})
|
||||
}
|
||||
|
||||
export default schedulerRoutes
|
||||
}
|
||||
|
||||
export default createSchedulerRoutes({ status: getSchedulerStatus, runs: listSchedulerRuns, refresh: refreshScheduler, run: runSchedulerJobNow })
|
||||
@@ -0,0 +1,71 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { test } from "node:test"
|
||||
import type { AppendEventBody } from "@mmapp/contracts/events"
|
||||
import type { SchedulerRunSnapshot, BackupsRunSnapshot } from "../types/scheduler-run-snapshot.js"
|
||||
import { runRecordedSchedulerJob, SchedulerRunJournalError, schedulerSnapshotError, type SchedulerRunRecord } from "./scheduler-recorded-run.js"
|
||||
const backup = (patch: Partial<BackupsRunSnapshot> = {}): BackupsRunSnapshot => ({ v: 1, job: "backups", sampledAt: "2026-10-01T00:00:00.000Z", due: true, created: 2, failures: 0, pruned: 1, errors: [], ...patch })
|
||||
function harness(snapshot: SchedulerRunSnapshot = backup()) {
|
||||
const records: SchedulerRunRecord[] = [], events: AppendEventBody[] = [], reported: unknown[] = []
|
||||
let collections = 0, hooks = 0
|
||||
const deps = { jobKey: snapshot.job, collect: async () => { collections++; return snapshot }, appendRun: async (row: SchedulerRunRecord) => { records.push(row) }, event: async (body: AppendEventBody) => { events.push(body) }, emitOkEvent: true, afterCollect: () => { hooks++ }, reportJournalError: (error: unknown) => { reported.push(error) } }
|
||||
return { deps, records, events, reported, counts: () => ({ collections, hooks }) }
|
||||
}
|
||||
test("event failure keeps one successful run and the confirmed snapshot", async () => {
|
||||
const h = harness(); h.deps.event = async () => { throw Error("event DB offline") }
|
||||
const result = await runRecordedSchedulerJob(h.deps)
|
||||
assert.equal(result.status, "ok"); assert.equal(result.eventError, "event DB offline")
|
||||
assert.equal(h.records.length, 1); assert.equal(h.records[0].status, "ok"); assert.equal(h.records[0].result, result.result)
|
||||
assert.deepEqual(h.counts(), { collections: 1, hooks: 1 }); assert.equal(h.reported.length, 1)
|
||||
})
|
||||
test("collector exception records one error; event failure never replaces the original", async () => {
|
||||
const h = harness(), original = Error("collector failed")
|
||||
h.deps.collect = async () => { throw original }; h.deps.event = async () => { throw Error("event failed") }
|
||||
await assert.rejects(runRecordedSchedulerJob(h.deps), error => error === original)
|
||||
assert.equal(h.records.length, 1); assert.equal(h.records[0].error, "collector failed"); assert.equal(h.records[0].result, null)
|
||||
assert.equal(h.counts().hooks, 0); assert.equal(h.reported.length, 1)
|
||||
})
|
||||
test("run journal failure never appends a second row or invokes the collector again", async () => {
|
||||
const h = harness(); let writes = 0
|
||||
h.deps.appendRun = async () => { writes++; throw Error("lost DB acknowledgment") }
|
||||
await assert.rejects(runRecordedSchedulerJob(h.deps), error => {
|
||||
assert.ok(error instanceof SchedulerRunJournalError); assert.equal(error.receipt.status, "ok"); assert.equal((error.receipt.result as BackupsRunSnapshot).created, 2)
|
||||
assert.match(error.message, /Повторный запуск может повторить операции/); return true
|
||||
})
|
||||
assert.equal(writes, 1); assert.equal(h.events.length, 0); assert.equal(h.counts().collections, 1)
|
||||
})
|
||||
test("collector and run journal failure retain both causes and never fabricate a saved run", async () => {
|
||||
const h = harness(); let writes = 0
|
||||
h.deps.collect = async () => { throw Error("setup failure") }; h.deps.appendRun = async () => { writes++; throw Error("journal failure") }
|
||||
await assert.rejects(runRecordedSchedulerJob(h.deps), error => {
|
||||
assert.ok(error instanceof SchedulerRunJournalError); assert.equal(error.receipt.status, "error"); assert.equal(error.receipt.error, "setup failure"); assert.equal((error.cause as Error).message, "journal failure"); return true
|
||||
}); assert.equal(writes, 1)
|
||||
})
|
||||
test("returned fatal snapshot is an error with the original snapshot, never a success event", async () => {
|
||||
const snapshot = backup({ created: 0, fatalError: "settings unavailable" }), h = harness(snapshot)
|
||||
const receipt = await runRecordedSchedulerJob(h.deps)
|
||||
assert.equal(receipt.status, "error"); assert.equal(receipt.error, "settings unavailable"); assert.equal(h.records[0].result, snapshot)
|
||||
assert.equal(h.events[0].level, "critical"); assert.equal(h.events[0].eventType, "scheduler.job.failed")
|
||||
})
|
||||
test("partial backups, renew and alert errors are explicit while confirmed results remain intact", async () => {
|
||||
const h = harness(backup({ uploadFailures: 1, errors: ["S3 unavailable"], eventErrors: ["start event unavailable"], metadataError: "metadata unavailable" }))
|
||||
const receipt = await runRecordedSchedulerJob(h.deps)
|
||||
assert.equal(receipt.status, "error"); assert.match(receipt.error!, /ограничениями.*S3 unavailable.*start event unavailable.*metadata unavailable/)
|
||||
assert.equal((receipt.result as BackupsRunSnapshot).created, 2); assert.equal(h.events[0].level, "warning")
|
||||
assert.match(schedulerSnapshotError({ v: 1, job: "certificates_renew", sampledAt: "now", checked: 2, renewed: 1, skippedTargets: 0, errors: ["renew failed"] })!, /renew failed/)
|
||||
assert.match(schedulerSnapshotError({ v: 1, job: "alert_engine", sampledAt: "now", rulesChecked: 1, standaloneFires: 0, groupFires: 0, skippedNoTelegram: false, errors: ["send failed"] })!, /send failed/)
|
||||
})
|
||||
test("valid offline/down observations are not collector errors", () => {
|
||||
assert.equal(schedulerSnapshotError({ v: 1, job: "traffic", sampledAt: "now", servers: [{ serverId: 1, name: "router", host: "host", ok: false, error: "offline" }] }), null)
|
||||
assert.equal(schedulerSnapshotError({ v: 1, job: "uptime_ping", sampledAt: "now", probes: [{ probeId: "probe", name: "probe", target: "host", srcServerId: 1, srcServerName: "router", srcInterface: "ether1", ok: false, rttMs: null, lossPct: 100, status: "down" }] }), null)
|
||||
})
|
||||
test("skipped/not-due snapshots preserve their reason and do not announce completed work", async () => {
|
||||
for (const snapshot of [backup({ due: false }), backup({ skipped: true })]) {
|
||||
const h = harness(snapshot); const receipt = await runRecordedSchedulerJob(h.deps)
|
||||
assert.equal(receipt.result, snapshot); assert.equal(h.events.length, 0); assert.equal(h.records.length, 1)
|
||||
}
|
||||
})
|
||||
test("post-collection hook failure is reported separately from the recorded outcome", async () => {
|
||||
const h = harness(); h.deps.afterCollect = () => { throw Error("hook failed") }
|
||||
const receipt = await runRecordedSchedulerJob(h.deps)
|
||||
assert.equal(receipt.status, "ok"); assert.equal(h.records.length, 1); assert.equal(h.reported.length, 1)
|
||||
})
|
||||
@@ -0,0 +1,82 @@
|
||||
import type { AppendEventBody } from "@mmapp/contracts/events"
|
||||
import type { SchedulerRunSnapshot } from "../types/scheduler-run-snapshot.js"
|
||||
|
||||
export interface SchedulerRunReceipt {
|
||||
status: "ok" | "error"
|
||||
error: string | null
|
||||
result: SchedulerRunSnapshot | null
|
||||
eventError?: string
|
||||
}
|
||||
export interface SchedulerRunRecord extends SchedulerRunReceipt {
|
||||
jobKey: string
|
||||
startedAt: string
|
||||
finishedAt: string
|
||||
durationMs: number
|
||||
}
|
||||
const message = (error: unknown) => error instanceof Error ? error.message : String(error)
|
||||
|
||||
/** Classify execution errors, not observed down/offline states in valid samples. */
|
||||
export function schedulerSnapshotError(snapshot: SchedulerRunSnapshot | null): string | null {
|
||||
if (!snapshot) return null
|
||||
if ("fatalError" in snapshot && snapshot.fatalError) return snapshot.fatalError
|
||||
const errors = "errors" in snapshot ? [...(snapshot.errors ?? [])] : []
|
||||
if (snapshot.job === "backups") {
|
||||
errors.push(...(snapshot.eventErrors ?? []))
|
||||
if (snapshot.metadataError && !errors.some(error => error.includes(snapshot.metadataError!))) errors.push("Метаданные: " + snapshot.metadataError)
|
||||
if (snapshot.failures && !errors.length) errors.push("Ошибки создания: " + snapshot.failures)
|
||||
if (snapshot.uploadFailures && !errors.length) errors.push("Ошибки S3: " + snapshot.uploadFailures)
|
||||
if (snapshot.pruneFailures && !errors.length) errors.push("Ошибки очистки: " + snapshot.pruneFailures)
|
||||
}
|
||||
return errors.length ? "Завершено с ограничениями: " + [...new Set(errors)].join("; ") : null
|
||||
}
|
||||
|
||||
export class SchedulerRunJournalError extends Error {
|
||||
constructor(public readonly receipt: SchedulerRunReceipt, cause: unknown) {
|
||||
super("Результат задачи получен, но запись журнала не подтверждена: " + message(cause) + ". Повторный запуск может повторить операции.", { cause })
|
||||
this.name = "SchedulerRunJournalError"
|
||||
}
|
||||
}
|
||||
|
||||
/** Only collection failures are collection failures; journal failures never create a second run. */
|
||||
export async function runRecordedSchedulerJob(deps: {
|
||||
jobKey: string
|
||||
collect: () => Promise<SchedulerRunSnapshot | undefined>
|
||||
appendRun: (record: SchedulerRunRecord) => Promise<void>
|
||||
event: (body: AppendEventBody) => Promise<void>
|
||||
emitOkEvent: boolean
|
||||
afterCollect?: () => void
|
||||
reportJournalError: (error: unknown) => void
|
||||
now?: () => Date
|
||||
}): Promise<SchedulerRunReceipt> {
|
||||
const now = deps.now ?? (() => new Date())
|
||||
const started = now()
|
||||
let result: SchedulerRunSnapshot | null = null
|
||||
let failed = false, failure: unknown
|
||||
try { result = await deps.collect() ?? null }
|
||||
catch (error) { failed = true; failure = error }
|
||||
const finished = now()
|
||||
const error = failed ? (message(failure) || "Неизвестная ошибка коллектора") : schedulerSnapshotError(result)
|
||||
const receipt: SchedulerRunReceipt = { status: failed || error ? "error" : "ok", error, result }
|
||||
if (!failed && deps.afterCollect) {
|
||||
try { deps.afterCollect() } catch (hookError) { deps.reportJournalError(hookError) }
|
||||
}
|
||||
const record: SchedulerRunRecord = { ...receipt, jobKey: deps.jobKey, startedAt: started.toISOString(), finishedAt: finished.toISOString(), durationMs: Math.max(0, finished.getTime() - started.getTime()) }
|
||||
try { await deps.appendRun(record) }
|
||||
catch (journalError) { deps.reportJournalError(journalError); throw new SchedulerRunJournalError(receipt, journalError) }
|
||||
const skipped = result && (("skipped" in result && result.skipped) || (result.job === "backups" && !result.due))
|
||||
if (error || (deps.emitOkEvent && !skipped)) {
|
||||
try {
|
||||
await deps.event({
|
||||
level: error ? (failed || (result && "fatalError" in result && result.fatalError) ? "critical" : "warning") : "info",
|
||||
eventType: error ? "scheduler.job.failed" : "scheduler.job.ok",
|
||||
sourceModule: "scheduler",
|
||||
title: error ? (failed ? "Ошибка задачи планировщика" : "Задача планировщика завершена с ошибками") : "Задача планировщика завершена",
|
||||
message: deps.jobKey + ": " + (error ?? "выполнено за " + record.durationMs + " мс"),
|
||||
entityType: "job", entityId: deps.jobKey,
|
||||
payload: { startedAt: record.startedAt, finishedAt: record.finishedAt },
|
||||
})
|
||||
} catch (eventError) { receipt.eventError = message(eventError); deps.reportJournalError(eventError) }
|
||||
}
|
||||
if (failed) throw failure
|
||||
return receipt
|
||||
}
|
||||
@@ -54,6 +54,7 @@ import {
|
||||
tryBeginSchedulerJob,
|
||||
} from "./scheduler-running.js"
|
||||
import { appendEvent } from "../modules/events/service/events-service.js"
|
||||
import { runRecordedSchedulerJob, type SchedulerRunReceipt } from "./scheduler-recorded-run.js"
|
||||
|
||||
export const JOB_KEYS = [
|
||||
"traffic",
|
||||
@@ -118,123 +119,70 @@ async function appendSchedulerRun(row: {
|
||||
})
|
||||
}
|
||||
|
||||
async function runSchedulerJobBody(jobKey: SchedulerJobKey): Promise<void> {
|
||||
const startedAt = Date.now()
|
||||
const startedIso = new Date().toISOString()
|
||||
let snapshot: SchedulerRunSnapshot | undefined
|
||||
try {
|
||||
switch (jobKey) {
|
||||
case "traffic":
|
||||
snapshot = await collectTrafficOnce()
|
||||
break
|
||||
case "uptime_resources":
|
||||
snapshot = await collectResourceSamplesOnce()
|
||||
break
|
||||
case "uptime_ping":
|
||||
snapshot = await collectPingProbesOnce()
|
||||
break
|
||||
case "uptime_speed":
|
||||
snapshot = await runScheduledSpeedProbesOnce()
|
||||
break
|
||||
case "servers_rest_ping":
|
||||
snapshot = await collectServersRestPingOnce()
|
||||
break
|
||||
case "gre_bgp":
|
||||
snapshot = await collectGreBgpSnapshotOnce()
|
||||
break
|
||||
case "internet_path":
|
||||
snapshot = await collectInternetPathSnapshotOnce()
|
||||
break
|
||||
case "certificates_renew":
|
||||
snapshot = await collectCertificatesRenewOnce()
|
||||
break
|
||||
case "backups":
|
||||
snapshot = await collectScheduledBackupsOnce()
|
||||
break
|
||||
case "geoip_update":
|
||||
snapshot = await collectGeoipUpdateOnce()
|
||||
break
|
||||
case "alert_engine": {
|
||||
const r = await runAlertEngineOnce()
|
||||
snapshot = {
|
||||
v: SCHEDULER_RUN_SNAPSHOT_VERSION,
|
||||
job: "alert_engine",
|
||||
sampledAt: r.sampledAt,
|
||||
rulesChecked: r.rulesChecked,
|
||||
standaloneFires: r.standaloneFires,
|
||||
groupFires: r.groupFires,
|
||||
skippedNoTelegram: r.skippedNoTelegram,
|
||||
...(r.ruleDiag.length ? { ruleDiag: r.ruleDiag } : {}),
|
||||
...(r.errors.length ? { errors: r.errors } : {}),
|
||||
async function runSchedulerJobBody(jobKey: SchedulerJobKey): Promise<SchedulerRunReceipt> {
|
||||
return runRecordedSchedulerJob({
|
||||
jobKey,
|
||||
collect: async () => {
|
||||
let snapshot: SchedulerRunSnapshot | undefined
|
||||
switch (jobKey) {
|
||||
case "traffic":
|
||||
snapshot = await collectTrafficOnce()
|
||||
break
|
||||
case "uptime_resources":
|
||||
snapshot = await collectResourceSamplesOnce()
|
||||
break
|
||||
case "uptime_ping":
|
||||
snapshot = await collectPingProbesOnce()
|
||||
break
|
||||
case "uptime_speed":
|
||||
snapshot = await runScheduledSpeedProbesOnce()
|
||||
break
|
||||
case "servers_rest_ping":
|
||||
snapshot = await collectServersRestPingOnce()
|
||||
break
|
||||
case "gre_bgp":
|
||||
snapshot = await collectGreBgpSnapshotOnce()
|
||||
break
|
||||
case "internet_path":
|
||||
snapshot = await collectInternetPathSnapshotOnce()
|
||||
break
|
||||
case "certificates_renew":
|
||||
snapshot = await collectCertificatesRenewOnce()
|
||||
break
|
||||
case "backups":
|
||||
snapshot = await collectScheduledBackupsOnce()
|
||||
break
|
||||
case "geoip_update":
|
||||
snapshot = await collectGeoipUpdateOnce()
|
||||
break
|
||||
case "alert_engine": {
|
||||
const r = await runAlertEngineOnce()
|
||||
snapshot = {
|
||||
v: SCHEDULER_RUN_SNAPSHOT_VERSION,
|
||||
job: "alert_engine",
|
||||
sampledAt: r.sampledAt,
|
||||
rulesChecked: r.rulesChecked,
|
||||
standaloneFires: r.standaloneFires,
|
||||
groupFires: r.groupFires,
|
||||
skippedNoTelegram: r.skippedNoTelegram,
|
||||
...(r.ruleDiag.length ? { ruleDiag: r.ruleDiag } : {}),
|
||||
...(r.errors.length ? { errors: r.errors } : {}),
|
||||
}
|
||||
break
|
||||
}
|
||||
break
|
||||
default:
|
||||
throw new Error(`Unknown job: ${jobKey}`)
|
||||
}
|
||||
default:
|
||||
throw new Error(`Unknown job: ${jobKey}`)
|
||||
}
|
||||
const finishedIso = new Date().toISOString()
|
||||
await appendSchedulerRun({
|
||||
jobKey,
|
||||
startedAt: startedIso,
|
||||
finishedAt: finishedIso,
|
||||
status: "ok",
|
||||
error: null,
|
||||
durationMs: Date.now() - startedAt,
|
||||
result: snapshot ?? null,
|
||||
})
|
||||
if (shouldAppendSchedulerOkEvent(jobKey)) {
|
||||
await appendEvent({
|
||||
level: "info",
|
||||
eventType: "scheduler.job.ok",
|
||||
sourceModule: "scheduler",
|
||||
title: "Задача планировщика завершена",
|
||||
message: `${jobKey}: выполнено за ${Date.now() - startedAt} мс`,
|
||||
entityType: "job",
|
||||
entityId: jobKey,
|
||||
payload: {
|
||||
startedAt: startedIso,
|
||||
finishedAt: finishedIso,
|
||||
},
|
||||
})
|
||||
}
|
||||
if (
|
||||
jobKey === "traffic" ||
|
||||
jobKey === "servers_rest_ping" ||
|
||||
jobKey === "uptime_resources" ||
|
||||
jobKey === "uptime_ping" ||
|
||||
jobKey === "uptime_speed" ||
|
||||
jobKey === "internet_path" ||
|
||||
jobKey === "gre_bgp"
|
||||
) {
|
||||
scheduleAlertEngineAfterDataCollectors()
|
||||
}
|
||||
} catch (e) {
|
||||
const msg = e instanceof Error ? e.message : String(e)
|
||||
const finishedIso = new Date().toISOString()
|
||||
await appendSchedulerRun({
|
||||
jobKey,
|
||||
startedAt: startedIso,
|
||||
finishedAt: finishedIso,
|
||||
status: "error",
|
||||
error: msg,
|
||||
durationMs: Date.now() - startedAt,
|
||||
result: snapshot ?? null,
|
||||
})
|
||||
await appendEvent({
|
||||
level: "critical",
|
||||
eventType: "scheduler.job.failed",
|
||||
sourceModule: "scheduler",
|
||||
title: "Ошибка задачи планировщика",
|
||||
message: `${jobKey}: ${msg}`,
|
||||
entityType: "job",
|
||||
entityId: jobKey,
|
||||
payload: {
|
||||
startedAt: startedIso,
|
||||
finishedAt: finishedIso,
|
||||
},
|
||||
})
|
||||
throw e
|
||||
}
|
||||
return snapshot
|
||||
},
|
||||
appendRun: appendSchedulerRun,
|
||||
event: appendEvent,
|
||||
emitOkEvent: shouldAppendSchedulerOkEvent(jobKey),
|
||||
afterCollect: () => {
|
||||
if (["traffic", "servers_rest_ping", "uptime_resources", "uptime_ping", "uptime_speed", "internet_path", "gre_bgp"].includes(jobKey)) scheduleAlertEngineAfterDataCollectors()
|
||||
},
|
||||
reportJournalError: (error) => console.warn("Scheduler journal/hook:", error),
|
||||
})
|
||||
}
|
||||
|
||||
/** Фоновый тик: пропуск, если предыдущий прогон ещё идёт. */
|
||||
@@ -243,14 +191,14 @@ export async function executeSchedulerJob(jobKey: SchedulerJobKey): Promise<void
|
||||
try {
|
||||
await runSchedulerJobBody(jobKey)
|
||||
} catch {
|
||||
/* залогировано в runSchedulerJobBody */
|
||||
/* Collector result is recorded once; unavailable journal is reported separately. */
|
||||
} finally {
|
||||
endSchedulerJob(jobKey)
|
||||
}
|
||||
}
|
||||
|
||||
/** Ручной запуск: 409, если job уже выполняется. */
|
||||
export async function runSchedulerJobNow(jobKey: string): Promise<void> {
|
||||
export async function runSchedulerJobNow(jobKey: string): Promise<SchedulerRunReceipt> {
|
||||
if (!JOB_KEYS.includes(jobKey as SchedulerJobKey)) {
|
||||
throw new Error(`Unknown job key: ${jobKey}`)
|
||||
}
|
||||
@@ -260,7 +208,7 @@ export async function runSchedulerJobNow(jobKey: string): Promise<void> {
|
||||
throw err
|
||||
}
|
||||
try {
|
||||
await runSchedulerJobBody(jobKey as SchedulerJobKey)
|
||||
return await runSchedulerJobBody(jobKey as SchedulerJobKey)
|
||||
} finally {
|
||||
endSchedulerJob(jobKey)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user