fix(alerts): учитывать свежесть и настройки источников оповещений
This commit is contained in:
Denozordec committed
1 parent
8ffb8b5f00
commit
db7105119d
19 files changed
+1614
-233
No files matched your search
@@ -8,7 +8,7 @@
|
||||
"dev": "tsx watch src/index.ts",
|
||||
"build": "npm run build -w @mmapp/contracts && tsc",
|
||||
"test:alerts-telegram": "tsx src/services/alerts-outbox-retry.test.ts && tsx src/services/alerts-outbox-overview.test.ts && tsx src/services/alert-engine/outbox-execution.test.ts && tsx src/services/alerts-telegram-delivery.test.ts && tsx src/services/alerts-revision.test.ts && tsx src/services/alerts-overview.test.ts && tsx ../packages/contracts/src/alerts-normalization.test.ts && tsx src/services/alerts-config-execution.test.ts && tsx src/services/alerts-telegram-settings.test.ts && tsx src/services/alerts-telegram-send.test.ts && tsx src/routes/alerts.test.ts && tsx src/plugins/auth.alerts.test.ts",
|
||||
"test:alert-engine": "tsx src/services/alert-engine/server-signal-identity.test.ts && tsx src/services/alert-engine/groups.test.ts && tsx src/services/alert-engine/evaluate-rule.test.ts && tsx src/services/gre-bgp-snapshot-collector.test.ts && tsx src/services/gre-bgp-snapshot-history.test.ts && tsx src/services/gre-bgp-snapshot-sql.test.ts && tsx src/services/alert-engine/confirm-stability.test.ts",
|
||||
"test:alert-engine": "tsx src/services/alert-engine/server-signal-identity.test.ts && tsx src/services/alert-engine/collector-source-signals.test.ts && tsx src/services/alert-engine/groups.test.ts && tsx src/services/alert-engine/evaluate-rule.test.ts && tsx src/services/gre-bgp-snapshot-collector.test.ts && tsx src/services/gre-bgp-snapshot-history.test.ts && tsx src/services/gre-bgp-snapshot-sql.test.ts && tsx src/services/alert-engine/collector-source-sql.test.ts && tsx src/services/alert-engine/confirm-stability.test.ts",
|
||||
"start": "node dist/index.js",
|
||||
"db:generate": "drizzle-kit generate",
|
||||
"db:migrate": "drizzle-kit migrate",
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { collectSourceSignals, probePeriodMs, type CollectorSourceConfig, type CollectorSourceRuns, type CollectorRunRow } from "./collector-source-signals.js"
|
||||
import { evaluateRule } from "./evaluate-rule.js"
|
||||
import type { ApiAlertRule } from "../alerts-service.js"
|
||||
const now = Date.parse("2026-10-12T10:00:00Z"), iso = (delta = 0) => new Date(now + delta).toISOString()
|
||||
const config: CollectorSourceConfig = {
|
||||
servers: [{ id: 1, name: "one", host: "host1", enabled: true, updatedAt: iso(-900_000) }, { id: 2, name: "two", host: "host2", enabled: true }],
|
||||
probes: [{ id: "slow", name: "slow", target: "target", srcServerId: 1, srcInterface: "ether1", enabled: true, intervalSec: 120 },
|
||||
{ id: "fast", name: "fast", target: "target", srcServerId: 2, srcInterface: "ether2", enabled: true, intervalSec: 0 }],
|
||||
uptime: { enabled: true, resourcesEnabled: true, pingEnabled: true, intervalSec: 15, probeIntervalSec: 15 },
|
||||
rest: { enabled: true, intervalSec: 120 }, traffic: { enabled: true, intervalSec: 30 },
|
||||
}
|
||||
const device = (id = 1) => ({ serverId: id, name: id === 1 ? "one" : "two", host: `host${id}` })
|
||||
const resource = (status = "offline", id = 1) => ({ ...device(id), status })
|
||||
const rest = (ok = false) => ({ ...device(), ok })
|
||||
const traffic = (rx = 0, tx = 0) => ({ ...device(), ok: true, sumRxMbps: rx, sumTxMbps: tx })
|
||||
const probe = (id = "slow") => { const p = config.probes.find(p => p.id === id)!; return { probeId: id, name: p.name, target: p.target, srcServerId: p.srcServerId, srcServerName: p.srcServerId === 1 ? "one" : "two", srcInterface: p.srcInterface, ok: true, status: "up", rttMs: 4, lossPct: 0 } }
|
||||
const row = (job: string, entries: unknown[], delta = 0, extra = {}): CollectorRunRow => ({ finishedAt: iso(delta), resultJson: { v: 1, job, sampledAt: iso(delta), [job === "uptime_ping" ? "probes" : "servers"]: entries, ...extra } })
|
||||
const empty: CollectorSourceRuns = { resources: [], rest: [], ping: [], traffic: [] }
|
||||
const run = (over: Partial<CollectorSourceRuns>, cfg = config) => collectSourceSignals(cfg, { ...empty, ...over }, now)
|
||||
const result = run({ resources: [row("uptime_resources", [resource()])], traffic: [row("traffic", [traffic()])], ping: [row("uptime_ping", [probe()])] })
|
||||
assert.equal(result.servers[0].status, "offline"); assert.equal(result.trafficByServer[0].rxMbps, 0); assert.equal(result.probes[0].rttMs, 4)
|
||||
|
||||
// Stale/future completion AND sample boundaries, errors, empty current and malformed values must not resurrect old success.
|
||||
for (const [field, job, value, period] of [["resources", "uptime_resources", resource(), 15_000], ["rest", "servers_rest_ping", rest(), 120_000], ["traffic", "traffic", traffic(), 30_000]] as const) {
|
||||
const old = row(job, [value], -1_000)
|
||||
const outputs = (rows: CollectorRunRow[]) => { const r = run({ [field]: rows }); return field === "traffic" ? r.trafficByServer : r.servers }
|
||||
for (const current of [
|
||||
{ finishedAt: iso(), resultJson: null }, { finishedAt: iso(), resultJson: "{" },
|
||||
row(job, []), row(job, [value], 0, { skipped: true }), row(job, [value], 0, { fatalError: "failure" }),
|
||||
row(job, [value], -3 * period - 1), row(job, [value], period + 1),
|
||||
{ ...row(job, [value]), finishedAt: iso(-3 * period - 1) }, { ...row(job, [value]), finishedAt: iso(period + 1) },
|
||||
row(job, [value], 0, { sampledAt: "invalid" }), row(job, [value], 0, { sampledAt: 1 }),
|
||||
row(job, [value], 0, { v: 99 }), row(job, [value], 0, { servers: {} }),
|
||||
]) assert.deepEqual(outputs([current, old]), [])
|
||||
assert.equal(outputs([row(job, [value], -3 * period)]).length, 1)
|
||||
assert.equal(outputs([row(job, [value], period)]).length, 1)
|
||||
}
|
||||
assert.deepEqual(run({ resources: [row("uptime_resources", [resource("unknown")]), row("uptime_resources", [resource()])] }).servers, [])
|
||||
assert.deepEqual(run({ resources: [row("uptime_resources", [{ ...resource(), status: ["offline"] }])] }).servers, [])
|
||||
for (const value of [{ ...traffic(), ok: false }, { ...device(), ok: true }, { ...traffic(), sumRxMbps: null }, { ...traffic(), sumRxMbps: -1 }, { ...traffic(), sumTxMbps: "0" }]) {
|
||||
assert.deepEqual(run({ traffic: [row("traffic", [value]), row("traffic", [traffic(100, 100)], -1_000)] }).trafficByServer, [])
|
||||
}
|
||||
// Mixed outcomes keep only independently confirmed current devices.
|
||||
assert.deepEqual(run({ resources: [row("uptime_resources", [resource("unknown"), resource("online", 2)])] }).servers.map(s => s.name), ["two"])
|
||||
assert.deepEqual(run({ resources: [row("uptime_resources", [resource(), resource()])] }).servers, [])
|
||||
assert.deepEqual(run({ resources: [row("uptime_resources", [resource()])] }, { ...config, servers: [] }).servers, [])
|
||||
assert.deepEqual(run({ resources: [row("uptime_resources", [resource()])] }, { ...config, servers: config.servers.map(s => ({ ...s, enabled: false })) }).servers, [])
|
||||
|
||||
// Scheduler semantics: dedicated switches override the legacy master flag; absent settings never seed defaults.
|
||||
assert.deepEqual(run({ resources: [row("uptime_resources", [resource()])], rest: [row("servers_rest_ping", [rest()])], traffic: [row("traffic", [traffic()])], ping: [row("uptime_ping", [probe()])] }, {
|
||||
...config, uptime: { ...config.uptime!, resourcesEnabled: false, pingEnabled: false }, rest: { ...config.rest!, enabled: false }, traffic: { ...config.traffic!, enabled: false },
|
||||
}), { servers: [], probes: [], trafficByServer: [] })
|
||||
assert.equal(run({ resources: [row("uptime_resources", [resource()])] }, { ...config, uptime: { ...config.uptime!, enabled: false } }).servers.length, 1)
|
||||
assert.deepEqual(run({ resources: [row("uptime_resources", [resource()])] }, { ...config, uptime: undefined }).servers, [])
|
||||
assert.equal(run({ resources: [row("uptime_resources", [resource()], -200_000)] }, { ...config, uptime: { ...config.uptime!, intervalSec: 300 } }).servers.length, 1)
|
||||
|
||||
// Disabled or stale REST cannot synthesize recovery from a different resource timeline.
|
||||
assert.equal(run({ resources: [row("uptime_resources", [resource()])], rest: [row("servers_rest_ping", [rest(true)], -361_000)] }).servers[0].status, "offline")
|
||||
const merged = run({ resources: [row("uptime_resources", [resource()])], rest: [row("servers_rest_ping", [rest(true)])] }).servers[0]
|
||||
assert.equal(merged.status, "online"); assert.equal(merged.prevStatus, null); assert.equal(merged.prev2Status, null)
|
||||
const recovered = run({ resources: [row("uptime_resources", [resource("online")]), row("uptime_resources", [resource()], -15_000)] }).servers[0]
|
||||
assert.equal(recovered.prevStatus, "offline")
|
||||
assert.equal(run({ resources: [row("uptime_resources", [resource("online")]), row("uptime_resources", [resource("unknown")], -1_000), row("uptime_resources", [resource()], -15_000)] }).servers[0].prevStatus, null)
|
||||
|
||||
// Slow probes stay available through many fast-only/empty due batches, but not through a global failed run.
|
||||
const due = [row("uptime_ping", [probe("fast")]), row("uptime_ping", [], -15_000), row("uptime_ping", [probe()], -90_000)]
|
||||
assert.deepEqual(run({ ping: due }).probes.map(p => p.key).sort(), ["fast → target", "slow → target"])
|
||||
assert.deepEqual(run({ ping: [{ finishedAt: iso(), resultJson: null }, ...due] }).probes, [])
|
||||
assert.deepEqual(run({ ping: [row("uptime_ping", [probe()], -360_001)] }).probes, [])
|
||||
assert.equal(run({ ping: [row("uptime_ping", [probe()], -360_000)] }).probes.length, 1)
|
||||
assert.deepEqual(run({ ping: [row("uptime_ping", [probe()], 15_001)] }).probes, [])
|
||||
for (const changed of [{ ...probe(), target: "old-target" }, { ...probe(), srcInterface: "old-interface" }, { ...probe(), srcServerId: 2 }, { ...probe(), status: "unknown" }, { ...probe(), rttMs: -1 }, { ...probe(), lossPct: 101 }]) {
|
||||
assert.deepEqual(run({ ping: [row("uptime_ping", [changed]), row("uptime_ping", [probe()], -15_000)] }).probes, [])
|
||||
}
|
||||
assert.deepEqual(run({ ping: [row("uptime_ping", [probe(), probe()])] }).probes, [])
|
||||
assert.deepEqual(run({ ping: [row("uptime_ping", [probe()])] }, { ...config, probes: [{ ...config.probes[0], enabled: false }] }).probes, [])
|
||||
assert.deepEqual(run({ ping: [row("uptime_ping", [probe()])] }, { ...config, probes: [config.probes[0], { ...config.probes[0], id: "duplicate" }] }).probes, [])
|
||||
assert.deepEqual(run({ ping: [row("uptime_ping", [probe()], -1_000)] }, { ...config, servers: [{ ...config.servers[0], updatedAt: iso() }] }).probes, [])
|
||||
assert.equal(run({ ping: [row("uptime_ping", [{ ...probe(), ok: false, status: "down", rttMs: null, lossPct: 100 }])] }).probes[0].status, "down")
|
||||
assert.equal(probePeriodMs(config, { ...config.probes[0], intervalSec: 1 }), 15_000)
|
||||
|
||||
// Suppressed measurements do not fire real evaluation paths (offline, loss, low traffic).
|
||||
const missing = { sampledAt: iso(), ...run({ resources: [row("uptime_resources", [resource()], -46_000)], traffic: [row("traffic", [{ ...device(), ok: false }])], ping: [{ finishedAt: iso(), resultJson: null }] }), greTunnels: [], bgpSessions: [] }
|
||||
const positive = { sampledAt: iso(), ...result, probes: [{ ...result.probes[0], status: "down" as const, rttMs: null, lossPct: 100 }], greTunnels: [], bgpSessions: [] }
|
||||
for (const [type, target, condition] of [["server", "one", "offline"], ["loss", "slow → target", "> 5%"], ["traffic", "one", "RX < 1"]] as const) {
|
||||
const rule = { id: type, name: type, type, targets: [target], conditions: [condition], enabled: true } as unknown as ApiAlertRule
|
||||
assert.equal(evaluateRule(rule, missing), null)
|
||||
assert.ok(evaluateRule(rule, positive))
|
||||
}
|
||||
console.log("collector source guards: settings/identity/current/unknown/time boundaries/due-only probes/history/evaluation passed")
|
||||
@@ -0,0 +1,129 @@
|
||||
import { createServerSignalIdentityGuard, type AlertServerIdentity } from "./server-signal-identity.js"
|
||||
import type { ProbeSignal, ServerSignal, TrafficServerSignal } from "./types.js"
|
||||
|
||||
export interface CollectorRunRow { resultJson: unknown; finishedAt: Date | string }
|
||||
export interface ProbeIdentity {
|
||||
id: string; name: string; target: string; srcServerId: number; srcInterface: string
|
||||
enabled: boolean; intervalSec: number; updatedAt?: Date | string
|
||||
}
|
||||
export interface CollectorSourceConfig {
|
||||
servers: Array<AlertServerIdentity & { updatedAt?: Date | string }>
|
||||
probes: ProbeIdentity[]
|
||||
uptime?: { enabled: boolean; resourcesEnabled: boolean | null; pingEnabled: boolean | null; intervalSec: number; probeIntervalSec: number }
|
||||
rest?: { enabled: boolean; intervalSec: number }
|
||||
traffic?: { enabled: boolean; intervalSec: number }
|
||||
}
|
||||
type JsonRow = Record<string, unknown>
|
||||
interface Run { sampledAt: string; sampledMs: number; finishedMs: number; rows: JsonRow[] }
|
||||
export interface CollectorSourceRuns { resources: CollectorRunRow[]; rest: CollectorRunRow[]; ping: CollectorRunRow[]; traffic: CollectorRunRow[] }
|
||||
const keyOf = (name: string) => name.trim().toLowerCase()
|
||||
const record = (value: unknown): value is JsonRow => value != null && typeof value === "object" && !Array.isArray(value)
|
||||
const finiteNonnegative = (value: unknown): value is number => typeof value === "number" && Number.isFinite(value) && value >= 0
|
||||
export function sourcePeriodMs(seconds: number, minimumSeconds: number): number {
|
||||
return typeof seconds === "number" && Number.isFinite(seconds) && seconds > 0 ? Math.max(minimumSeconds, seconds) * 1000 : NaN
|
||||
}
|
||||
export function probePeriodMs(config: CollectorSourceConfig, probe: ProbeIdentity): number {
|
||||
const global = config.uptime?.probeIntervalSec ?? NaN
|
||||
return Math.max(sourcePeriodMs(global, 1), probe.intervalSec > 0 ? sourcePeriodMs(probe.intervalSec, 0) : sourcePeriodMs(global, 5))
|
||||
}
|
||||
function parseRun(row: CollectorRunRow, job: string, field: "servers" | "probes"): Run | null {
|
||||
let value = row.resultJson
|
||||
if (typeof value === "string") { try { value = JSON.parse(value) } catch { return null } }
|
||||
if (!record(value) || value.v !== 1 || value.job !== job || value.skipped === true || value.fatalError
|
||||
|| typeof value.sampledAt !== "string" || !Array.isArray(value[field])) return null
|
||||
const sampledMs = Date.parse(value.sampledAt), finishedMs = new Date(row.finishedAt).getTime()
|
||||
if (!Number.isFinite(sampledMs) || !Number.isFinite(finishedMs)) return null
|
||||
return { sampledAt: value.sampledAt, sampledMs, finishedMs, rows: value[field].filter(record) }
|
||||
}
|
||||
function fresh(run: Run, period: number, now: number, future = period): boolean {
|
||||
return Number.isFinite(period) && period > 0 && [run.sampledMs, run.finishedMs].every(t => t >= now - 3 * period && t <= now + future)
|
||||
}
|
||||
function afterConfig(run: Run, ...times: Array<Date | string | undefined>): boolean {
|
||||
return times.every(t => t === undefined || (Number.isFinite(new Date(t).getTime()) && run.sampledMs >= new Date(t).getTime()))
|
||||
}
|
||||
function identityRows(config: CollectorSourceConfig, rows: JsonRow[]) {
|
||||
return createServerSignalIdentityGuard(config.servers)(rows as unknown as Array<{ serverId: number; name: string; host: string }>)
|
||||
}
|
||||
function serverSignals(config: CollectorSourceConfig, rows: CollectorRunRow[], rest: boolean, period: number, now: number): ServerSignal[] {
|
||||
const runs = rows.map(row => parseRun(row, rest ? "servers_rest_ping" : "uptime_resources", "servers"))
|
||||
const current = runs[0]
|
||||
if (!current || !fresh(current, period, now)) return []
|
||||
const status = (row: JsonRow): ServerSignal["status"] | null => rest
|
||||
? (typeof row.ok === "boolean" ? row.ok ? "online" : "offline" : null)
|
||||
: (typeof row.status === "string" && ["online", "offline", "degraded"].includes(row.status) ? row.status as ServerSignal["status"] : null)
|
||||
const out: ServerSignal[] = []
|
||||
for (const row of identityRows(config, current.rows)) {
|
||||
const server = config.servers.find(s => Number(s.id) === row.serverId)!
|
||||
if (!afterConfig(current, server.updatedAt)) continue
|
||||
const states: ServerSignal["status"][] = []
|
||||
for (const run of runs) {
|
||||
if (!run || !fresh(run, period, now) || !afterConfig(run, server.updatedAt)) break
|
||||
const matches = run.rows.filter(r => r.serverId === row.serverId)
|
||||
if (!matches.length) continue
|
||||
if (matches.length !== 1 || !identityRows(config, matches).length) break
|
||||
const state = status(matches[0])
|
||||
if (state === null) break
|
||||
states.push(state)
|
||||
if (states.length === 3) break
|
||||
}
|
||||
if (states.length) out.push({ name: row.name, status: states[0], prevStatus: states[1] ?? null, prev2Status: states[2] ?? null, sampledAt: current.sampledAt })
|
||||
}
|
||||
return out
|
||||
}
|
||||
function trafficSignals(config: CollectorSourceConfig, rows: CollectorRunRow[], now: number): TrafficServerSignal[] {
|
||||
if (config.traffic?.enabled !== true) return []
|
||||
const run = rows[0] && parseRun(rows[0], "traffic", "servers")
|
||||
if (!run || !fresh(run, sourcePeriodMs(config.traffic.intervalSec, 5), now)) return []
|
||||
const out: TrafficServerSignal[] = []
|
||||
for (const row of identityRows(config, run.rows)) {
|
||||
const raw = row as unknown as JsonRow
|
||||
const server = config.servers.find(s => Number(s.id) === row.serverId)!
|
||||
if (raw.ok !== true || !finiteNonnegative(raw.sumRxMbps) || !finiteNonnegative(raw.sumTxMbps) || !afterConfig(run, server.updatedAt)) continue
|
||||
out.push({ serverName: row.name, rxMbps: raw.sumRxMbps, txMbps: raw.sumTxMbps, sampledAt: run.sampledAt })
|
||||
}
|
||||
return out
|
||||
}
|
||||
function probeSignals(config: CollectorSourceConfig, rows: CollectorRunRow[], now: number): ProbeSignal[] {
|
||||
if ((config.uptime?.pingEnabled ?? config.uptime?.enabled) !== true) return []
|
||||
const runs = rows.map(row => parseRun(row, "uptime_ping", "probes"))
|
||||
const probes = config.probes.filter(p => p.enabled && config.servers.some(s => s.enabled === true && Number(s.id) === p.srcServerId))
|
||||
const label = (p: ProbeIdentity) => `${p.name.trim()} → ${p.target.trim()}`
|
||||
const counts = new Map<string, number>()
|
||||
for (const probe of probes) counts.set(label(probe), (counts.get(label(probe)) ?? 0) + 1)
|
||||
const out: ProbeSignal[] = []
|
||||
for (const probe of probes) {
|
||||
if (!probe.name.trim() || !probe.target.trim() || counts.get(label(probe)) !== 1) continue
|
||||
const source = config.servers.find(s => Number(s.id) === probe.srcServerId)!
|
||||
const period = probePeriodMs(config, probe), future = sourcePeriodMs(config.uptime!.probeIntervalSec, 1)
|
||||
for (const run of runs) {
|
||||
// Invalid completed runs are barriers; a valid due-only batch may omit this probe.
|
||||
if (!run || !fresh(run, period, now, future) || !afterConfig(run, probe.updatedAt, source.updatedAt)) break
|
||||
const matches = run.rows.filter(r => r.probeId === probe.id)
|
||||
if (!matches.length) continue
|
||||
if (matches.length !== 1) break
|
||||
const row = matches[0]
|
||||
if (row.name !== probe.name || row.target !== probe.target || row.srcServerId !== probe.srcServerId
|
||||
|| row.srcInterface !== probe.srcInterface || row.srcServerName !== (source.name || source.host)
|
||||
|| typeof row.status !== "string" || !["up", "warn", "down"].includes(row.status) || typeof row.ok !== "boolean"
|
||||
|| !(row.rttMs === null || finiteNonnegative(row.rttMs)) || !finiteNonnegative(row.lossPct) || row.lossPct > 100) break
|
||||
if (row.ok === false && (row.status !== "down" || row.rttMs !== null || row.lossPct !== 100)) break
|
||||
out.push({ key: label(probe), status: row.status as ProbeSignal["status"], rttMs: row.rttMs as number | null, lossPct: row.lossPct, sampledAt: run.sampledAt })
|
||||
break
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
/** Missing/unknown data is never a fabricated zero or offline. */
|
||||
export function collectSourceSignals(config: CollectorSourceConfig, runs: CollectorSourceRuns, now = Date.now()) {
|
||||
const resourcesOn = (config.uptime?.resourcesEnabled ?? config.uptime?.enabled) === true
|
||||
const resources = resourcesOn ? serverSignals(config, runs.resources, false, sourcePeriodMs(config.uptime!.intervalSec, 5), now) : []
|
||||
const rest = config.rest?.enabled === true ? serverSignals(config, runs.rest, true, sourcePeriodMs(config.rest.intervalSec, 10), now) : []
|
||||
const combined = new Map<string, ServerSignal>()
|
||||
for (const signal of [...resources, ...rest]) {
|
||||
const previous = combined.get(keyOf(signal.name))
|
||||
// A fresh success confirms reachability; use its own history, never invent cross-source recovery.
|
||||
if (!previous || (signal.status === "online" && previous.status !== "online")
|
||||
|| ((signal.status === "online") === (previous.status === "online") && Date.parse(signal.sampledAt) >= Date.parse(previous.sampledAt))) combined.set(keyOf(signal.name), signal)
|
||||
}
|
||||
return { servers: [...combined.values()].sort((a, b) => a.name.localeCompare(b.name)), probes: probeSignals(config, runs.ping, now), trafficByServer: trafficSignals(config, runs.traffic, now) }
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { pool, closePool } from "../../db/index.js"
|
||||
import { initDatabase } from "../../db/bootstrap.js"
|
||||
import { buildSignalSnapshotFromCollectors } from "./signals-from-collectors.js"
|
||||
import { evaluateRule } from "./evaluate-rule.js"
|
||||
import type { ApiAlertRule } from "../alerts-service.js"
|
||||
|
||||
const database = process.env.MMAPP_OWN_SQL_DB
|
||||
if (!database) console.log("collector sources SQL: skipped; explicit owned database required")
|
||||
else {
|
||||
assert(/^mm_accept_166_\d+$/.test(database))
|
||||
assert.equal(decodeURIComponent(new URL(process.env.DATABASE_URL!).pathname.slice(1)), database)
|
||||
assert.equal((await pool.query("SELECT current_database() AS name")).rows[0].name, database)
|
||||
const prefix = "own-sources170-" + Date.now() + "-", now = Date.now(), iso = (delta = 0) => new Date(now + delta).toISOString()
|
||||
let sequence = 0, cases = 0
|
||||
const ids: number[] = []
|
||||
const snapshot = (job: string, rows: unknown[], delta = -1_000) => ({ v: 1, job, sampledAt: iso(delta), [job === "uptime_ping" ? "probes" : "servers"]: rows })
|
||||
const insert = async (job: string, payload: unknown, delta = -1_000, status = "ok") => {
|
||||
await pool.query("INSERT INTO scheduler_runs(id,job_key,started_at,finished_at,status,error,result_json) VALUES($1,$2,$3,$3,$4,$5,$6::jsonb)", [prefix + sequence++, job, iso(delta), status, status === "error" ? "owned failure" : null, payload === null ? null : JSON.stringify(payload)])
|
||||
}
|
||||
const reset = () => pool.query("DELETE FROM scheduler_runs WHERE id LIKE $1", [prefix + "%"])
|
||||
const build = () => buildSignalSnapshotFromCollectors(() => now)
|
||||
try {
|
||||
await initDatabase()
|
||||
for (const index of [1, 2]) ids.push(Number((await pool.query("INSERT INTO servers(name,host,enabled,updated_at) VALUES($1,$2,true,$3) RETURNING id", [prefix + index, `own-host-${index}`, iso(-900_000)])).rows[0].id))
|
||||
await pool.query("INSERT INTO uptime_settings(id,enabled,resources_enabled,ping_enabled,interval_sec,probe_interval_sec) VALUES(1,true,true,true,15,15) ON CONFLICT(id) DO UPDATE SET enabled=true,resources_enabled=true,ping_enabled=true,interval_sec=15,probe_interval_sec=15")
|
||||
await pool.query("INSERT INTO servers_api_ping_settings(id,enabled,interval_sec) VALUES(1,false,120) ON CONFLICT(id) DO UPDATE SET enabled=false,interval_sec=120")
|
||||
await pool.query("INSERT INTO traffic_settings(id,enabled,interval_sec) VALUES(1,true,30) ON CONFLICT(id) DO UPDATE SET enabled=true,interval_sec=30")
|
||||
for (const [index, interval] of [[0, 120], [1, 0]]) await pool.query("INSERT INTO uptime_probes(id,src_server_id,src_interface,name,target,enabled,interval_sec,updated_at) VALUES($1,$2,'ether1',$3,'own-target',true,$4,$5)", [prefix + "probe" + index, ids[index], prefix + "probe" + index, interval, iso(-900_000)])
|
||||
const server = (index = 0) => ({ serverId: ids[index], name: prefix + (index + 1), host: `own-host-${index + 1}` })
|
||||
const resource = (status = "offline", index = 0) => ({ ...server(index), status })
|
||||
const traffic = { ...server(), ok: true, sumRxMbps: 0, sumTxMbps: 12 }
|
||||
const probe = (index = 0) => ({ probeId: prefix + "probe" + index, name: prefix + "probe" + index, target: "own-target", srcServerId: ids[index], srcServerName: prefix + (index + 1), srcInterface: "ether1", ok: true, status: "up", rttMs: 5, lossPct: 0 })
|
||||
|
||||
await insert("uptime_resources", snapshot("uptime_resources", [resource("online")], -15_000), -15_000)
|
||||
await insert("uptime_resources", snapshot("uptime_resources", [resource(), resource("online", 1)]), -1_000, "error")
|
||||
await insert("traffic", snapshot("traffic", [traffic]))
|
||||
await insert("uptime_ping", snapshot("uptime_ping", [probe()], -90_000), -90_000)
|
||||
// More than the old 24-tick history; the slow probe must still survive due-only batches.
|
||||
for (let index = 30; index > 0; index--) await insert("uptime_ping", snapshot("uptime_ping", [probe(1)], -index * 1_000), -index * 1_000)
|
||||
let signals = await build()
|
||||
assert.equal(signals.servers.length, 2); assert.equal(signals.servers[0].prevStatus, "online")
|
||||
assert.equal(signals.trafficByServer[0].rxMbps, 0); assert.equal(signals.probes.length, 2); cases++
|
||||
const offlineRule = { id: "owned", name: "owned", type: "server", targets: [server().name], conditions: ["offline"], enabled: true } as ApiAlertRule
|
||||
assert.ok(evaluateRule(offlineRule, signals))
|
||||
|
||||
await pool.query("UPDATE uptime_settings SET resources_enabled=false,ping_enabled=false WHERE id=1")
|
||||
await pool.query("UPDATE traffic_settings SET enabled=false WHERE id=1")
|
||||
signals = await build(); assert.deepEqual(signals.servers, []); assert.deepEqual(signals.probes, []); assert.deepEqual(signals.trafficByServer, []); cases++
|
||||
await pool.query("UPDATE uptime_settings SET resources_enabled=true,ping_enabled=true WHERE id=1")
|
||||
await pool.query("UPDATE traffic_settings SET enabled=true WHERE id=1")
|
||||
await pool.query("UPDATE servers SET enabled=false WHERE id=$1", [ids[0]])
|
||||
signals = await build(); assert.equal(signals.servers.length, 1); assert.equal(signals.probes.length, 1); assert.deepEqual(signals.trafficByServer, []); cases++
|
||||
await pool.query("UPDATE servers SET enabled=true WHERE id=$1", [ids[0]])
|
||||
await pool.query("UPDATE servers SET name=$1 WHERE id=$2", [server().name, ids[1]])
|
||||
signals = await build(); assert.deepEqual(signals.servers, []); assert.deepEqual(signals.trafficByServer, []); cases++
|
||||
await pool.query("UPDATE servers SET name=$1 WHERE id=$2", [server(1).name, ids[1]])
|
||||
|
||||
for (const broken of [null, "{", { ...snapshot("uptime_resources", []), servers: {} }, snapshot("uptime_resources", [])]) {
|
||||
await reset(); await insert("uptime_resources", snapshot("uptime_resources", [resource()], -15_000), -15_000)
|
||||
await insert("uptime_resources", broken, -1_000, "error")
|
||||
signals = await build(); assert.deepEqual(signals.servers, []); assert.equal(evaluateRule(offlineRule, signals), null); cases++
|
||||
}
|
||||
for (const [delta, field] of [[-46_000, "sample"], [16_000, "sample"], [-46_000, "finish"], [16_000, "finish"]] as const) {
|
||||
await reset(); await insert("uptime_resources", snapshot("uptime_resources", [resource()], field === "sample" ? delta : -1_000), field === "finish" ? delta : -1_000)
|
||||
assert.deepEqual((await build()).servers, []); cases++
|
||||
}
|
||||
await reset(); await insert("uptime_resources", snapshot("uptime_resources", [resource("unknown"), resource("online", 1)]), -1_000, "error")
|
||||
signals = await build(); assert.equal(signals.servers.length, 1); assert.equal(signals.servers[0].name, server(1).name); cases++
|
||||
await reset(); await insert("traffic", snapshot("traffic", [traffic], -30_000), -30_000)
|
||||
await insert("traffic", snapshot("traffic", [{ ...server(), ok: false }]), -1_000, "error")
|
||||
assert.deepEqual((await build()).trafficByServer, []); cases++
|
||||
await reset(); await insert("uptime_ping", snapshot("uptime_ping", [probe()], -15_000), -15_000)
|
||||
await insert("uptime_ping", null, -1_000, "error")
|
||||
assert.deepEqual((await build()).probes, []); cases++
|
||||
await reset(); await insert("uptime_ping", snapshot("uptime_ping", [probe()], -15_000), -15_000)
|
||||
await pool.query("UPDATE uptime_probes SET updated_at=$1 WHERE id=$2", [iso(), probe().probeId])
|
||||
assert.deepEqual((await build()).probes, []); cases++
|
||||
console.log(`collector sources owned SQL: ${cases} cases; real settings/identity/error/time/due-only/unknown/reconfiguration/evaluation passed`)
|
||||
} finally {
|
||||
await reset()
|
||||
await pool.query("DELETE FROM uptime_probes WHERE id LIKE $1", [prefix + "%"])
|
||||
await pool.query("DELETE FROM servers WHERE id = ANY($1::bigint[])", [ids])
|
||||
assert.equal((await pool.query("SELECT count(*)::int AS n FROM scheduler_runs WHERE id LIKE $1", [prefix + "%"])).rows[0].n, 0)
|
||||
assert.equal((await pool.query("SELECT count(*)::int AS n FROM servers WHERE id=ANY($1::bigint[])", [ids])).rows[0].n, 0)
|
||||
await closePool()
|
||||
}
|
||||
}
|
||||
@@ -20,6 +20,8 @@ assert.deepEqual(guard([sample(1), sample(1)]), [])
|
||||
assert.deepEqual(guard([{ ...sample(1), name: " chr " }]), [{ ...sample(1), name: " chr " }])
|
||||
assert.deepEqual(createServerSignalIdentityGuard([{ ...server(1), id: "1" }])([sample(1)]), [sample(1)])
|
||||
assert.deepEqual(createServerSignalIdentityGuard([{ ...server(1), id: "9007199254740993" }])([sample(1)]), [])
|
||||
assert.deepEqual(createServerSignalIdentityGuard([server(1, "")])([sample(1, "host-1")]), [sample(1, "host-1")])
|
||||
assert.deepEqual(createServerSignalIdentityGuard([server(1, ""), server(2, "host-1")])([sample(1, "host-1")]), [])
|
||||
// Filtering is immutable, and checks every run against the same current inventory.
|
||||
const rows = [sample(1), sample(2)]
|
||||
const before = JSON.stringify(rows)
|
||||
|
||||
@@ -8,19 +8,20 @@ export interface AlertServerIdentity {
|
||||
interface SnapshotServerIdentity { serverId: number; name: string; host: string }
|
||||
|
||||
const nameKey = (name: string) => name.trim().toLowerCase()
|
||||
const displayName = (server: AlertServerIdentity) => server.name || server.host
|
||||
|
||||
/** Rules currently address servers by name. Ambiguous names must never select an arbitrary device. */
|
||||
export function createServerSignalIdentityGuard(servers: AlertServerIdentity[]) {
|
||||
const active = servers.filter(s => s.enabled === true)
|
||||
const nameCounts = new Map<string, number>()
|
||||
for (const server of active) {
|
||||
const key = nameKey(server.name)
|
||||
const key = nameKey(displayName(server))
|
||||
nameCounts.set(key, (nameCounts.get(key) ?? 0) + 1)
|
||||
}
|
||||
const eligible = new Map<number, AlertServerIdentity>()
|
||||
for (const server of active) {
|
||||
const id = Number(server.id)
|
||||
const key = nameKey(server.name)
|
||||
const key = nameKey(displayName(server))
|
||||
if (Number.isSafeInteger(id) && id > 0 && key && nameCounts.get(key) === 1) eligible.set(id, server)
|
||||
}
|
||||
return <T extends SnapshotServerIdentity>(rows: T[]): T[] => {
|
||||
@@ -30,7 +31,7 @@ export function createServerSignalIdentityGuard(servers: AlertServerIdentity[])
|
||||
if (!row || counts.get(row.serverId) !== 1) return false
|
||||
const server = eligible.get(row.serverId)
|
||||
return server != null && typeof row.name === "string" && typeof row.host === "string"
|
||||
&& nameKey(row.name) === nameKey(server.name) && row.host.trim() === server.host.trim()
|
||||
&& nameKey(row.name) === nameKey(displayName(server)) && row.host.trim() === server.host.trim()
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -2,226 +2,40 @@ import { dbAll } from "../../db/index.js"
|
||||
import { collectBgpSignalsFromRuns } from "./bgp-signals.js"
|
||||
import { collectGreSignalsFromRuns } from "./gre-signals.js"
|
||||
import { readGreBgpRunSnapshots } from "../gre-bgp-snapshot-history.js"
|
||||
import { createServerSignalIdentityGuard, type AlertServerIdentity } from "./server-signal-identity.js"
|
||||
import type {
|
||||
PingRunSnapshot,
|
||||
ResourcesRunSnapshot,
|
||||
SchedulerRunSnapshot,
|
||||
ServersRestPingRunSnapshot,
|
||||
TrafficRunSnapshot,
|
||||
} from "../../types/scheduler-run-snapshot.js"
|
||||
import type {
|
||||
ProbeSignal,
|
||||
ServerSignal,
|
||||
SignalSnapshot,
|
||||
TrafficServerSignal,
|
||||
} from "./types.js"
|
||||
import { collectSourceSignals, probePeriodMs, type CollectorRunRow, type CollectorSourceConfig } from "./collector-source-signals.js"
|
||||
import type { SignalSnapshot } from "./types.js"
|
||||
|
||||
function normalizeNameKey(v: string): string {
|
||||
return v.trim().toLowerCase()
|
||||
async function readRuns(job: string, limit: number, since?: string): Promise<CollectorRunRow[]> {
|
||||
return dbAll<CollectorRunRow>(`SELECT result_json AS "resultJson", finished_at AS "finishedAt"
|
||||
FROM scheduler_runs WHERE job_key = ? AND status IN ('ok', 'error')
|
||||
${since ? "AND finished_at >= ?" : ""}
|
||||
ORDER BY finished_at DESC, id DESC LIMIT ?`, since ? [job, since, limit] : [job, limit])
|
||||
}
|
||||
|
||||
function safeParseRunSnapshot(raw: unknown): SchedulerRunSnapshot | null {
|
||||
const obj = typeof raw === "string" ? (() => {
|
||||
try { return JSON.parse(raw) as unknown } catch { return null }
|
||||
})() : raw
|
||||
if (!obj || typeof obj !== "object") return null
|
||||
return obj as SchedulerRunSnapshot
|
||||
}
|
||||
|
||||
async function readLatestOkRunSnapshots(jobKey: string, limit = 32): Promise<SchedulerRunSnapshot[]> {
|
||||
const rows = await dbAll<{ resultJson: unknown }>(
|
||||
`SELECT result_json AS "resultJson"
|
||||
FROM scheduler_runs
|
||||
WHERE job_key = ? AND status = 'ok' AND result_json IS NOT NULL
|
||||
ORDER BY finished_at DESC
|
||||
LIMIT ?`,
|
||||
[jobKey, limit],
|
||||
)
|
||||
const out: SchedulerRunSnapshot[] = []
|
||||
for (const row of rows) {
|
||||
const parsed = safeParseRunSnapshot(row.resultJson)
|
||||
if (!parsed || parsed.job !== jobKey) continue
|
||||
out.push(parsed)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
type TriState = "online" | "offline" | "degraded"
|
||||
|
||||
function timelineToServerSignal(name: string, sampledAt: string, statuses: TriState[]): ServerSignal {
|
||||
/** Read-only: do not seed defaults or load credentials while building alert facts. */
|
||||
export async function buildSignalSnapshotFromCollectors(clock: () => number = Date.now): Promise<SignalSnapshot> {
|
||||
const now = clock()
|
||||
const [servers, probes, uptime, rest, traffic] = await Promise.all([
|
||||
dbAll<CollectorSourceConfig["servers"][number]>("SELECT id, name, host, enabled, updated_at AS \"updatedAt\" FROM servers"),
|
||||
dbAll<CollectorSourceConfig["probes"][number]>(`SELECT id, name, target, src_server_id AS "srcServerId", src_interface AS "srcInterface",
|
||||
enabled, interval_sec AS "intervalSec", updated_at AS "updatedAt" FROM uptime_probes`),
|
||||
dbAll<NonNullable<CollectorSourceConfig["uptime"]>>(`SELECT enabled, resources_enabled AS "resourcesEnabled", ping_enabled AS "pingEnabled",
|
||||
interval_sec AS "intervalSec", probe_interval_sec AS "probeIntervalSec" FROM uptime_settings WHERE id = 1`),
|
||||
dbAll<NonNullable<CollectorSourceConfig["rest"]>>('SELECT enabled, interval_sec AS "intervalSec" FROM servers_api_ping_settings WHERE id = 1'),
|
||||
dbAll<NonNullable<CollectorSourceConfig["traffic"]>>('SELECT enabled, interval_sec AS "intervalSec" FROM traffic_settings WHERE id = 1'),
|
||||
])
|
||||
const config: CollectorSourceConfig = { servers, probes, uptime: uptime[0], rest: rest[0], traffic: traffic[0] }
|
||||
const periods = probes.filter(p => p.enabled).map(p => probePeriodMs(config, p)).filter(Number.isFinite)
|
||||
const probeWindowMs = 3 * Math.max(5_000, ...periods)
|
||||
const [resourceRuns, restRuns, pingRuns, trafficRuns, greBgpRuns] = await Promise.all([
|
||||
readRuns("uptime_resources", 16), readRuns("servers_rest_ping", 16),
|
||||
// Due-only batches: cover the longest configured TTL rather than only the last 24 ticks.
|
||||
readRuns("uptime_ping", 4096, new Date(now - probeWindowMs).toISOString()),
|
||||
readRuns("traffic", 8), readGreBgpRunSnapshots(undefined, clock),
|
||||
])
|
||||
return {
|
||||
name,
|
||||
status: statuses[0] ?? "offline",
|
||||
prevStatus: statuses[1] ?? null,
|
||||
prev2Status: statuses[2] ?? null,
|
||||
sampledAt,
|
||||
}
|
||||
}
|
||||
|
||||
function collectServerSignalsFromResourceRuns(runs: ResourcesRunSnapshot[]): Map<string, ServerSignal> {
|
||||
const byKey = new Map<
|
||||
string,
|
||||
{ name: string; sampledAt: string; statuses: TriState[] }
|
||||
>()
|
||||
for (const run of runs) {
|
||||
for (const s of run.servers ?? []) {
|
||||
const name = String(s.name ?? "").trim()
|
||||
if (!name) continue
|
||||
const key = normalizeNameKey(name)
|
||||
const rec = byKey.get(key) ?? { name, sampledAt: run.sampledAt, statuses: [] }
|
||||
if (rec.statuses.length < 3) rec.statuses.push(s.status === "online" ? "online" : "offline")
|
||||
if (!byKey.has(key)) rec.sampledAt = run.sampledAt
|
||||
byKey.set(key, rec)
|
||||
}
|
||||
}
|
||||
const out = new Map<string, ServerSignal>()
|
||||
for (const [key, rec] of byKey) out.set(key, timelineToServerSignal(rec.name, rec.sampledAt, rec.statuses))
|
||||
return out
|
||||
}
|
||||
|
||||
function collectServerSignalsFromRestRuns(runs: ServersRestPingRunSnapshot[]): Map<string, ServerSignal> {
|
||||
const byKey = new Map<
|
||||
string,
|
||||
{ name: string; sampledAt: string; statuses: TriState[] }
|
||||
>()
|
||||
for (const run of runs) {
|
||||
for (const s of run.servers ?? []) {
|
||||
const name = String(s.name ?? "").trim()
|
||||
if (!name) continue
|
||||
const key = normalizeNameKey(name)
|
||||
const rec = byKey.get(key) ?? { name, sampledAt: run.sampledAt, statuses: [] }
|
||||
if (rec.statuses.length < 3) rec.statuses.push(s.ok ? "online" : "offline")
|
||||
if (!byKey.has(key)) rec.sampledAt = run.sampledAt
|
||||
byKey.set(key, rec)
|
||||
}
|
||||
}
|
||||
const out = new Map<string, ServerSignal>()
|
||||
for (const [key, rec] of byKey) out.set(key, timelineToServerSignal(rec.name, rec.sampledAt, rec.statuses))
|
||||
return out
|
||||
}
|
||||
|
||||
function mergeResourceAndRestServer(resource: ServerSignal, rest: ServerSignal): ServerSignal {
|
||||
const rUp = resource.status === "online"
|
||||
const tUp = rest.status === "online"
|
||||
if (tUp && !rUp) {
|
||||
const rBad = resource.status === "offline" || resource.status === "degraded"
|
||||
if (rBad) {
|
||||
return {
|
||||
name: rest.name,
|
||||
status: "online",
|
||||
prevStatus: "online",
|
||||
prev2Status: resource.status,
|
||||
sampledAt: rest.sampledAt,
|
||||
}
|
||||
}
|
||||
return rest
|
||||
}
|
||||
if (rUp && !tUp) return resource
|
||||
if (tUp && rUp) {
|
||||
if (rest.sampledAt >= resource.sampledAt) {
|
||||
if (resource.prevStatus === "offline" || resource.prevStatus === "degraded") {
|
||||
return {
|
||||
name: rest.name,
|
||||
status: "online",
|
||||
prevStatus: "online",
|
||||
prev2Status: resource.prevStatus === "degraded" ? "degraded" : "offline",
|
||||
sampledAt: rest.sampledAt,
|
||||
}
|
||||
}
|
||||
if (
|
||||
resource.prevStatus === "online" &&
|
||||
resource.prev2Status != null &&
|
||||
(resource.prev2Status === "offline" || resource.prev2Status === "degraded")
|
||||
) {
|
||||
return {
|
||||
name: rest.name,
|
||||
status: "online",
|
||||
prevStatus: "online",
|
||||
prev2Status: resource.prev2Status,
|
||||
sampledAt: rest.sampledAt,
|
||||
}
|
||||
}
|
||||
}
|
||||
return rest.sampledAt >= resource.sampledAt ? rest : resource
|
||||
}
|
||||
return rest.sampledAt >= resource.sampledAt ? rest : resource
|
||||
}
|
||||
|
||||
function collectProbeSignalsFromRuns(runs: PingRunSnapshot[]): ProbeSignal[] {
|
||||
const out = new Map<string, ProbeSignal>()
|
||||
for (const run of runs) {
|
||||
for (const p of run.probes ?? []) {
|
||||
const key = `${String(p.name ?? "").trim()} → ${String(p.target ?? "").trim()}`
|
||||
if (!key.trim() || out.has(key)) continue
|
||||
out.set(key, {
|
||||
key,
|
||||
status: p.status,
|
||||
rttMs: p.rttMs ?? null,
|
||||
lossPct: p.lossPct ?? null,
|
||||
sampledAt: run.sampledAt,
|
||||
})
|
||||
}
|
||||
}
|
||||
return [...out.values()]
|
||||
}
|
||||
|
||||
function collectTrafficSignalsFromRuns(runs: TrafficRunSnapshot[]): TrafficServerSignal[] {
|
||||
const out = new Map<string, TrafficServerSignal>()
|
||||
for (const run of runs) {
|
||||
for (const s of run.servers ?? []) {
|
||||
const name = String(s.name ?? "").trim()
|
||||
if (!name) continue
|
||||
const key = normalizeNameKey(name)
|
||||
if (out.has(key)) continue
|
||||
out.set(key, {
|
||||
serverName: name,
|
||||
rxMbps: Number(s.sumRxMbps ?? 0),
|
||||
txMbps: Number(s.sumTxMbps ?? 0),
|
||||
sampledAt: run.sampledAt,
|
||||
})
|
||||
}
|
||||
}
|
||||
return [...out.values()]
|
||||
}
|
||||
|
||||
/** Собирает сигналы alert_engine напрямую из результатов collector jobs (`scheduler_runs.result_json`). */
|
||||
export async function buildSignalSnapshotFromCollectors(): Promise<SignalSnapshot> {
|
||||
const sampledAt = new Date().toISOString()
|
||||
// Read only identity/configuration; credentials are not needed by the alert engine.
|
||||
const currentServers = await dbAll<AlertServerIdentity>("SELECT id, name, host, enabled FROM servers")
|
||||
const guardServerRows = createServerSignalIdentityGuard(currentServers)
|
||||
const resourceRuns = await readLatestOkRunSnapshots("uptime_resources", 16) as ResourcesRunSnapshot[]
|
||||
const restRuns = await readLatestOkRunSnapshots("servers_rest_ping", 16) as ServersRestPingRunSnapshot[]
|
||||
const pingRuns = await readLatestOkRunSnapshots("uptime_ping", 24) as PingRunSnapshot[]
|
||||
const trafficRuns = await readLatestOkRunSnapshots("traffic", 8) as TrafficRunSnapshot[]
|
||||
const greBgpRuns = await readGreBgpRunSnapshots()
|
||||
|
||||
const resourceByKey = collectServerSignalsFromResourceRuns(resourceRuns.map(run => ({ ...run, servers: guardServerRows(run.servers ?? []) })))
|
||||
const restByKey = collectServerSignalsFromRestRuns(restRuns.map(run => ({ ...run, servers: guardServerRows(run.servers ?? []) })))
|
||||
const serverKeys = new Set([...resourceByKey.keys(), ...restByKey.keys()])
|
||||
const servers: ServerSignal[] = []
|
||||
for (const key of serverKeys) {
|
||||
const resource = resourceByKey.get(key)
|
||||
const rest = restByKey.get(key)
|
||||
if (!rest) {
|
||||
if (resource) servers.push(resource)
|
||||
continue
|
||||
}
|
||||
if (!resource) {
|
||||
servers.push(rest)
|
||||
continue
|
||||
}
|
||||
servers.push(mergeResourceAndRestServer(resource, rest))
|
||||
}
|
||||
servers.sort((a, b) => a.name.localeCompare(b.name, undefined, { sensitivity: "base" }))
|
||||
|
||||
return {
|
||||
sampledAt,
|
||||
servers,
|
||||
probes: collectProbeSignalsFromRuns(pingRuns),
|
||||
trafficByServer: collectTrafficSignalsFromRuns(trafficRuns.map(run => ({ ...run, servers: guardServerRows(run.servers ?? []) }))),
|
||||
greTunnels: collectGreSignalsFromRuns(greBgpRuns),
|
||||
bgpSessions: collectBgpSignalsFromRuns(greBgpRuns),
|
||||
sampledAt: new Date(now).toISOString(),
|
||||
...collectSourceSignals(config, { resources: resourceRuns, rest: restRuns, ping: pingRuns, traffic: trafficRuns }, now),
|
||||
greTunnels: collectGreSignalsFromRuns(greBgpRuns), bgpSessions: collectBgpSignalsFromRuns(greBgpRuns),
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user