fix(uptime): отличать ошибки опроса от измеренных потерь Ping
This commit is contained in:
Denozordec committed
1 parent
ca23284486
commit
8fb46b1552
35 files changed
+1582
-78
No files matched your search
@@ -0,0 +1,3 @@
|
||||
-- A failed command is an attempt, not a measured packet-loss percentage.
|
||||
ALTER TABLE uptime_probe_samples ADD COLUMN IF NOT EXISTS measurement_known BOOLEAN NOT NULL DEFAULT true;
|
||||
ALTER TABLE uptime_probe_samples ADD COLUMN IF NOT EXISTS observation_error TEXT;
|
||||
@@ -15,7 +15,7 @@
|
||||
"db:studio": "drizzle-kit studio",
|
||||
"db:migrate-from-sqlite": "tsx src/scripts/migrate-sqlite-to-pg.ts",
|
||||
"facts:rebuild": "tsx src/scripts/rebuild-flow-facts.ts",
|
||||
"test:probes": "tsx src/routes/probes.test.ts && tsx src/routes/probes.cancel.test.ts",
|
||||
"test:probes": "tsx src/routes/probes.test.ts && tsx src/routes/probes.cancel.test.ts && tsx src/services/ping-measurement.test.ts && tsx src/services/uptime-probe-observations.test.ts && tsx src/services/ping-measurement-sql.test.ts",
|
||||
"test:exec": "tsx src/services/mikrotik-logs.test.ts && tsx src/routes/exec.test.ts && tsx src/plugins/auth.exec.test.ts",
|
||||
"test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts && tsx src/plugins/auth.ospf-optimizer.test.ts",
|
||||
"test:wireguard": "npx tsx src/services/wireguard-config.test.ts && npx tsx src/services/wireguard-ipv6.test.ts && npx tsx --test src/services/wireguard-import.test.ts src/services/wireguard-import-apply.test.ts src/services/wireguard-import-revision.test.ts src/services/wireguard-delete.test.ts && npx tsx src/routes/wireguard-import.test.ts",
|
||||
|
||||
@@ -365,6 +365,8 @@ export const uptimeProbeSamples = pgTable("uptime_probe_samples", {
|
||||
rttMs: integer("rtt_ms"),
|
||||
lossPct: integer("loss_pct").notNull().default(0),
|
||||
status: text("status", { enum: ["up", "warn", "down"] }).notNull().default("down"),
|
||||
measurementKnown: boolean("measurement_known").notNull().default(true),
|
||||
observationError: text("observation_error"),
|
||||
}, (t) => [
|
||||
primaryKey({ columns: [t.probeId, t.sampledAt] }),
|
||||
])
|
||||
|
||||
@@ -241,6 +241,7 @@ const uptimeRoutes: FastifyPluginAsyncZod = async (app) => {
|
||||
if (!Number.isFinite(serverId)) return reply.status(400).send({ error: "Invalid server id" })
|
||||
const srv = (await db.select().from(servers).where(eq(servers.id, serverId)).limit(1))[0]
|
||||
if (!srv) return reply.status(404).send({ error: "Server not found" })
|
||||
if (!srv.enabled) return reply.status(409).send({ error: "Сервер отключён: включите его перед опросом интерфейсов" })
|
||||
try {
|
||||
const rows = await MikrotikClient.fromServer(srv).getInterfaces()
|
||||
const interfaces = rows
|
||||
@@ -271,6 +272,7 @@ const uptimeRoutes: FastifyPluginAsyncZod = async (app) => {
|
||||
if (!Number.isFinite(serverId)) return reply.status(400).send({ error: "Invalid server id" })
|
||||
const srv = (await db.select().from(servers).where(eq(servers.id, serverId)).limit(1))[0]
|
||||
if (!srv) return reply.status(404).send({ error: "Server not found" })
|
||||
if (!srv.enabled) return reply.status(409).send({ error: "Сервер отключён: включите его перед опросом интерфейсов" })
|
||||
try {
|
||||
const client = MikrotikClient.fromServer(srv)
|
||||
const [ifaces, ips] = await Promise.all([
|
||||
@@ -322,8 +324,9 @@ const uptimeRoutes: FastifyPluginAsyncZod = async (app) => {
|
||||
|
||||
const probes = probeRows.map((p) => {
|
||||
const rows = probeSamples.filter((r) => r.probeId === p.id)
|
||||
const last = rows[rows.length - 1]
|
||||
const series = toSeries(rows.map((r) => r.rttMs ?? 0))
|
||||
const observations=buildUptimeProbeObservations(rows.filter(r=>Date.parse(r.sampledAt)<=Date.parse(rangeEndAt)))
|
||||
const last = observations.samples.at(-1)
|
||||
const series = observations.samples.map(r => r.rtt)
|
||||
return {
|
||||
id: p.id,
|
||||
srcServerId: String(p.srcServerId),
|
||||
@@ -331,13 +334,13 @@ const uptimeRoutes: FastifyPluginAsyncZod = async (app) => {
|
||||
name: p.name,
|
||||
target: p.target,
|
||||
filter: p.probeFilter || "—",
|
||||
rtt: last?.rttMs ?? null,
|
||||
loss: last?.lossPct ?? 100,
|
||||
status: (last?.status ?? "down") as "up" | "warn" | "down",
|
||||
rtt: last?.rtt ?? null,
|
||||
loss: last?.loss ?? null,
|
||||
status: last?.status ?? "unknown",
|
||||
series,
|
||||
enabled: p.enabled,
|
||||
showOnDashboard: p.showOnDashboard,
|
||||
observations:buildUptimeProbeObservations(rows.filter(r=>Date.parse(r.sampledAt)<=Date.parse(rangeEndAt))),
|
||||
observations,
|
||||
}
|
||||
})
|
||||
|
||||
|
||||
@@ -67,6 +67,7 @@ assert.equal(run({ resources: [row("uptime_resources", [resource("online")]), ro
|
||||
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, [])
|
||||
for (const malformed of [null, {}, {probeId:""}]) assert.deepEqual(run({ ping: [row("uptime_ping", [malformed]), ...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, [])
|
||||
@@ -77,7 +78,8 @@ 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.deepEqual(run({ ping: [row("uptime_ping", [{ ...probe(), ok: false, status: "down", rttMs: null, lossPct: 100 }]), row("uptime_ping", [probe()], -15_000)] }).probes, [])
|
||||
assert.equal(run({ ping: [row("uptime_ping", [{ ...probe(), ok: true, 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).
|
||||
|
||||
@@ -31,6 +31,7 @@ function parseRun(row: CollectorRunRow, job: string, field: "servers" | "probes"
|
||||
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
|
||||
if (field === "probes" && value[field].some(row => !record(row) || typeof row.probeId !== "string" || !row.probeId.trim())) 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) }
|
||||
@@ -104,9 +105,8 @@ function probeSignals(config: CollectorSourceConfig, rows: CollectorRunRow[], no
|
||||
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"
|
||||
|| typeof row.status !== "string" || !["up", "warn", "down"].includes(row.status) || row.ok !== true
|
||||
|| !(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
|
||||
}
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { createServer } from "node:http"
|
||||
import Fastify from "fastify"
|
||||
import { validatorCompiler, serializerCompiler } from "@fastify/type-provider-zod"
|
||||
import { pool, closePool } from "../db/index.js"
|
||||
import { initDatabase } from "../db/bootstrap.js"
|
||||
import { collectPingProbesOnce, collectPingForProbeIds, readProbeSamplesSince } from "./uptime-collector.js"
|
||||
import { buildUptimeProbeObservations } from "./uptime-probe-observations.js"
|
||||
import { buildSignalSnapshotFromCollectors } from "./alert-engine/signals-from-collectors.js"
|
||||
import uptimeRoutes from "../routes/uptime.js"
|
||||
|
||||
const database = process.env.MMAPP_OWN_SQL_DB
|
||||
if (!database) console.log("ping measurement SQL: skipped; explicit owned database required")
|
||||
else {
|
||||
assert(/^mm_accept_166_\d+$/.test(database));assert.equal(new URL(process.env.DATABASE_URL!).pathname,"/"+database)
|
||||
assert.equal((await pool.query("SELECT current_database() AS name")).rows[0].name,database)
|
||||
const prefix="own-measure173-"+Date.now()+"-",probeId=prefix+"probe"
|
||||
let mode:"deny"|"empty"|"loss"|"zero"="deny",requests=0,serverId:number|undefined,cases=0
|
||||
const http=createServer((req,res)=>{requests++;res.setHeader("Content-Type","application/json");if(mode==="deny"){res.statusCode=401;res.end(JSON.stringify({error:401,message:"Unauthorized"}))}else res.end(JSON.stringify(mode==="empty"?[]:[{sent:"4",received:mode==="loss"?"0":"4","packet-loss":mode==="loss"?"100":"0","avg-rtt":"0ms"}]))})
|
||||
await new Promise<void>(resolve=>http.listen(0,"127.0.0.1",resolve))
|
||||
const address=http.address();assert(address&&typeof address==="object")
|
||||
const app=Fastify();app.setValidatorCompiler(validatorCompiler);app.setSerializerCompiler(serializerCompiler);await app.register(uptimeRoutes)
|
||||
try {
|
||||
await initDatabase()
|
||||
serverId=Number((await pool.query("INSERT INTO servers(name,host,port,use_ssl,enabled,username,password,updated_at) VALUES($1,'127.0.0.1',$2,false,true,'fixture','fixture',$3) RETURNING id",[prefix+"server",address.port,new Date(Date.now()-3_600_000).toISOString()])).rows[0].id)
|
||||
await pool.query("INSERT INTO uptime_settings(id,enabled,resources_enabled,ping_enabled,interval_sec,probe_interval_sec) VALUES(1,true,false,true,15,15) ON CONFLICT(id) DO UPDATE SET enabled=true,resources_enabled=false,ping_enabled=true,interval_sec=15,probe_interval_sec=15")
|
||||
await pool.query("INSERT INTO uptime_probes(id,src_server_id,name,target,interval_sec,updated_at) VALUES($1,$2,$3,'192.0.2.1',1,$4)",[probeId,serverId,prefix+"probe",new Date(Date.now()-3_600_000).toISOString()])
|
||||
const since=new Date(Date.now()-3_600_000).toISOString()
|
||||
const latest=async()=>{const rows=await readProbeSamplesSince(since);return buildUptimeProbeObservations(rows.filter(r=>r.probeId===probeId)).samples.at(-1)!}
|
||||
const snapshot=await collectPingProbesOnce();assert.equal(snapshot.probes[0].ok,false);assert.equal(snapshot.probes[0].status,"unknown");assert.equal(snapshot.probes[0].lossPct,null)
|
||||
let sample=await latest();assert.equal(sample.status,"unknown");assert.equal(sample.loss,null);assert("error" in sample);cases++
|
||||
await pool.query("INSERT INTO scheduler_runs(id,job_key,started_at,finished_at,status,result_json) VALUES($1,'uptime_ping',$2,$3,'error',$4::jsonb)",[prefix+"run",snapshot.sampledAt,new Date().toISOString(),JSON.stringify(snapshot)])
|
||||
assert.deepEqual((await buildSignalSnapshotFromCollectors()).probes,[])
|
||||
let response=await app.inject({method:"GET",url:"/uptime/overview?range=1h"});assert.equal(response.statusCode,200);let dto=response.json().probes.find((p:{id:string})=>p.id===probeId);assert.equal(dto.status,"unknown");assert.equal(dto.loss,null);cases++
|
||||
mode="empty";assert.equal((await collectPingForProbeIds([probeId])).polled,1);sample=await latest();assert.equal(sample.status,"unknown");assert.equal(sample.loss,null);cases++
|
||||
mode="loss";await collectPingForProbeIds([probeId]);sample=await latest();assert.equal(sample.status,"down");assert.equal(sample.loss,100);assert.equal(sample.rtt,null);cases++
|
||||
mode="zero";await collectPingForProbeIds([probeId]);sample=await latest();assert.equal(sample.status,"up");assert.equal(sample.loss,0);assert.equal(sample.rtt,0);cases++
|
||||
response=await app.inject({method:"GET",url:"/uptime/overview?range=1h"});dto=response.json().probes.find((p:{id:string})=>p.id===probeId);assert.equal(dto.status,"up");assert.equal(dto.loss,0);cases++
|
||||
// Reclassify exact legacy scheduled failures through retained provenance, without guessing real loss.
|
||||
const oldTime=new Date(Date.now()-120_000).toISOString()
|
||||
await pool.query("INSERT INTO uptime_probe_samples(probe_id,sampled_at,status,loss_pct) VALUES($1,$2,'down',100)",[probeId,oldTime])
|
||||
await pool.query("INSERT INTO scheduler_runs(id,job_key,started_at,finished_at,status,result_json) VALUES($1,'uptime_ping',$2,$2,'error',$3::jsonb)",[prefix+"legacy",oldTime,JSON.stringify({v:1,job:"uptime_ping",sampledAt:oldTime,probes:[{probeId,ok:false,error:"legacy API denial"}]})])
|
||||
const old=(await readProbeSamplesSince(since)).find(r=>r.probeId===probeId&&Date.parse(r.sampledAt)===Date.parse(oldTime))!;assert.equal(old.measurementKnown,false);cases++
|
||||
await pool.query("UPDATE servers SET enabled=false WHERE id=$1",[serverId]);const before=requests
|
||||
for(const url of[`/uptime/sources/${serverId}/interfaces`,`/uptime/speed-test/endpoints/${serverId}/interfaces`]){response=await app.inject({method:"GET",url});assert.equal(response.statusCode,409)}
|
||||
assert.equal((await collectPingForProbeIds([probeId])).polled,0);assert.equal(requests,before);cases++
|
||||
console.log(`ping measurement owned SQL: ${cases} cases; real HTTP client/401/empty/manual/loss/zero/API/provenance/disabled guards passed`)
|
||||
}finally{
|
||||
await app.close();await new Promise<void>(resolve=>http.close(()=>resolve()))
|
||||
await pool.query("DELETE FROM scheduler_runs WHERE id LIKE $1",[prefix+"%"])
|
||||
await pool.query("DELETE FROM uptime_probe_samples WHERE probe_id=$1",[probeId]);await pool.query("DELETE FROM uptime_probes WHERE id=$1",[probeId])
|
||||
if(serverId!==undefined)await pool.query("DELETE FROM servers WHERE id=$1",[serverId])
|
||||
await closePool()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { parsePing } from "./uptime-collector.js"
|
||||
for (const rows of [[], [{}], [{status:"not enough permissions"}], [{sent:"0",received:"0","packet-loss":"0","avg-rtt":"0ms"}], [{sent:"4",received:"5","packet-loss":"0"}], [{sent:"4bad",received:"4","packet-loss":"0"}]]) {
|
||||
const result=parsePing(rows)
|
||||
assert.equal(result.measured,false);assert.equal(result.avgRtt,null)
|
||||
}
|
||||
const down=parsePing([{sent:"4",received:"0","packet-loss":"100","avg-rtt":"0ms"}])
|
||||
assert.equal(down.measured,true);assert.equal(down.status,"down");assert.equal(down.loss,100);assert.equal(down.avgRtt,null)
|
||||
const up=parsePing([{sent:"4",received:"4","packet-loss":"0","avg-rtt":"0ms"}])
|
||||
assert.equal(up.measured,true);assert.equal(up.status,"up");assert.equal(up.loss,0);assert.equal(up.avgRtt,0)
|
||||
const partial=parsePing([{time:"500us"},{status:"timeout"}])
|
||||
assert.equal(partial.measured,true);assert.equal(partial.loss,50)
|
||||
console.log("ping measurement: empty/unsupported is not a measurement; zero and real 100 percent loss preserved")
|
||||
@@ -2,7 +2,7 @@ import { replacePingProbes } from "./ping-probe-replacement.js"
|
||||
import type { PingProbeWrite } from "@mmapp/contracts"
|
||||
import { and, asc, eq, gte, inArray, max, sql } from "drizzle-orm"
|
||||
import { listEnabledMonitoringServers } from "./monitoring-server-catalog.js"
|
||||
import { db } from "../db/index.js"
|
||||
import { db, dbAll } from "../db/index.js"
|
||||
import {
|
||||
uptimeProbeSamples,
|
||||
uptimeProbes,
|
||||
@@ -76,7 +76,7 @@ export function parsePing(results: Array<{ time?: string; status?: string; sent?
|
||||
.filter((v): v is number => v != null && Number.isFinite(v))
|
||||
const avgFromReplies = rtts.length ? Math.round(rtts.reduce((a, b) => a + b, 0) / rtts.length) : null
|
||||
const avgFromSummary = parsePingTimeMs(sum?.time) ?? parsePingTimeMs(sum?.["avg-rtt"])
|
||||
const avgRtt = avgFromReplies ?? (avgFromSummary != null ? Math.round(avgFromSummary) : null)
|
||||
let avgRtt = avgFromReplies ?? (avgFromSummary != null ? Math.round(avgFromSummary) : null)
|
||||
const sent = Number.parseInt(String(sum?.sent ?? ""), 10)
|
||||
const received = Number.parseInt(String(sum?.received ?? ""), 10)
|
||||
const sentFallback = okReplies.length + timedOutReplies.length
|
||||
@@ -86,13 +86,17 @@ export function parsePing(results: Array<{ time?: string; status?: string; sent?
|
||||
const lossRaw = String(sum?.["packet-loss"] ?? "")
|
||||
const parsedLoss = Number.parseInt(lossRaw.replace("%", ""), 10)
|
||||
const lossSource = Number.isFinite(parsedLoss) ? parsedLoss : computedLoss
|
||||
const validCounts = (sum?.sent === undefined || (/^\d+$/.test(String(sum.sent).trim()) && Number.isSafeInteger(sent) && sent > 0))
|
||||
&& (sum?.received === undefined || (/^\d+$/.test(String(sum.received).trim()) && Number.isSafeInteger(received) && received >= 0 && (sum?.sent === undefined || received <= sent)))
|
||||
const measured = validCounts && typeof lossSource === "number" && Number.isFinite(lossSource) && lossSource >= 0 && lossSource <= 100
|
||||
const loss =
|
||||
typeof lossSource === "number" && Number.isFinite(lossSource)
|
||||
? Math.max(0, Math.min(100, lossSource))
|
||||
: 100
|
||||
const status: "up" | "warn" | "down" =
|
||||
loss >= 100 ? "down" : (loss > 1 || (avgRtt ?? 0) > PING_PROBE_WARN_RTT_MS ? "warn" : "up")
|
||||
return { avgRtt, loss, status }
|
||||
if (loss === 100 || !measured) avgRtt = null
|
||||
return { avgRtt, loss, status, measured }
|
||||
}
|
||||
|
||||
/** Disabled excludes every server type from resource and ping collection. */
|
||||
@@ -269,6 +273,7 @@ export async function collectPingProbesOnce(): Promise<PingRunSnapshot> {
|
||||
const client = MikrotikClient.fromServer(src)
|
||||
const results = await client.ping(p.target, 4, p.srcInterface || undefined)
|
||||
const parsed = parsePing(results)
|
||||
if (!parsed.measured) throw new Error("RouterOS не вернул подтверждённые результаты Ping")
|
||||
await db.insert(uptimeProbeSamples).values({
|
||||
probeId: p.id,
|
||||
sampledAt: now,
|
||||
@@ -295,6 +300,8 @@ export async function collectPingProbesOnce(): Promise<PingRunSnapshot> {
|
||||
rttMs: null,
|
||||
lossPct: 100,
|
||||
status: "down",
|
||||
measurementKnown: false,
|
||||
observationError: err instanceof Error ? err.message : String(err),
|
||||
})
|
||||
snapshot.probes.push({
|
||||
probeId: p.id,
|
||||
@@ -305,8 +312,8 @@ export async function collectPingProbesOnce(): Promise<PingRunSnapshot> {
|
||||
srcInterface: p.srcInterface || "",
|
||||
ok: false,
|
||||
rttMs: null,
|
||||
lossPct: 100,
|
||||
status: "down",
|
||||
lossPct: null,
|
||||
status: "unknown",
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
})
|
||||
}
|
||||
@@ -367,6 +374,7 @@ export async function collectPingForProbeIds(probeIds: string[]): Promise<{ poll
|
||||
const client = MikrotikClient.fromServer(src)
|
||||
const results = await client.ping(p.target, 4, p.srcInterface || undefined)
|
||||
const parsed = parsePing(results)
|
||||
if (!parsed.measured) throw new Error("RouterOS не вернул подтверждённые результаты Ping")
|
||||
await db.insert(uptimeProbeSamples).values({
|
||||
probeId: p.id,
|
||||
sampledAt: now,
|
||||
@@ -374,13 +382,15 @@ export async function collectPingForProbeIds(probeIds: string[]): Promise<{ poll
|
||||
lossPct: parsed.loss,
|
||||
status: parsed.status,
|
||||
})
|
||||
} catch {
|
||||
} catch (err) {
|
||||
await db.insert(uptimeProbeSamples).values({
|
||||
probeId: p.id,
|
||||
sampledAt: now,
|
||||
rttMs: null,
|
||||
lossPct: 100,
|
||||
status: "down",
|
||||
measurementKnown: false,
|
||||
observationError: err instanceof Error ? err.message : String(err),
|
||||
})
|
||||
}
|
||||
polled += 1
|
||||
@@ -482,9 +492,26 @@ export async function replaceProbes(rows: PingProbeWrite[]) {
|
||||
}
|
||||
|
||||
export async function readProbeSamplesSince(sinceIso: string) {
|
||||
return await db.select().from(uptimeProbeSamples)
|
||||
const rows = await db.select().from(uptimeProbeSamples)
|
||||
.where(gte(uptimeProbeSamples.sampledAt, sinceIso))
|
||||
.orderBy(asc(uptimeProbeSamples.sampledAt))
|
||||
if (!rows.length) return rows
|
||||
// Old scheduled failures can be classified only when their exact snapshot provenance still exists.
|
||||
const failures = await dbAll<{probeId: string | null; sampledAt: string | null; error: string | null}>(`
|
||||
SELECT probe->>'probeId' AS "probeId", result_json->>'sampledAt' AS "sampledAt", probe->>'error' AS error
|
||||
FROM scheduler_runs CROSS JOIN LATERAL jsonb_array_elements(
|
||||
CASE WHEN jsonb_typeof(result_json->'probes')='array' THEN result_json->'probes' ELSE '[]'::jsonb END) probe
|
||||
WHERE job_key='uptime_ping' AND status IN ('ok','error') AND finished_at >= ?
|
||||
AND result_json->'v'='1'::jsonb AND result_json->>'job'='uptime_ping' AND probe->'ok'='false'::jsonb`, [sinceIso])
|
||||
const bySample = new Map<string, string>()
|
||||
for (const failure of failures) {
|
||||
if (!failure.probeId || !failure.sampledAt || !Number.isFinite(Date.parse(failure.sampledAt))) continue
|
||||
bySample.set(`${failure.probeId}\0${Date.parse(failure.sampledAt)}`, failure.error ?? "Результат измерения не подтверждён")
|
||||
}
|
||||
return rows.map(row => {
|
||||
const error = bySample.get(`${row.probeId}\0${Date.parse(row.sampledAt)}`)
|
||||
return error === undefined ? row : {...row,measurementKnown:false,observationError:error}
|
||||
})
|
||||
}
|
||||
|
||||
export async function readResourceSamplesSince(sinceIso: string, serverId: number) {
|
||||
|
||||
@@ -2,5 +2,6 @@ import assert from "node:assert/strict"
|
||||
import {buildUptimeProbeObservations} from "./uptime-probe-observations.js"
|
||||
assert.deepEqual(buildUptimeProbeObservations([]),{samples:[]})
|
||||
const good={sampledAt:"2026-10-03T00:00:00Z",rttMs:0,lossPct:0,status:"up" as const},failed={sampledAt:"2026-10-03T00:01:00Z",rttMs:null,lossPct:100,status:"down" as const}
|
||||
assert.deepEqual(buildUptimeProbeObservations([failed,good]),{samples:[{sampledAt:good.sampledAt,rtt:0,loss:0,status:"up"},{sampledAt:failed.sampledAt,rtt:null,loss:100,status:"down"}]});assert.equal(good.rttMs,0)
|
||||
assert.deepEqual(buildUptimeProbeObservations([failed,good]),{samples:[{sampledAt:new Date(good.sampledAt).toISOString(),rtt:0,loss:0,status:"up"},{sampledAt:new Date(failed.sampledAt).toISOString(),rtt:null,loss:100,status:"down"}]});assert.equal(good.rttMs,0)
|
||||
assert.deepEqual(buildUptimeProbeObservations([{...failed,measurementKnown:false,observationError:"API denied"}]).samples,[{sampledAt:new Date(failed.sampledAt).toISOString(),rtt:null,loss:null,status:"unknown",error:"API denied"}])
|
||||
console.log("uptime probe observations tests ok")
|
||||
@@ -1,4 +1,6 @@
|
||||
import type {UptimeProbeSampleRow} from "../db/schema.js"
|
||||
type Sample=Pick<UptimeProbeSampleRow,"sampledAt"|"rttMs"|"lossPct"|"status">
|
||||
type Sample=Pick<UptimeProbeSampleRow,"sampledAt"|"rttMs"|"lossPct"|"status">&Partial<Pick<UptimeProbeSampleRow,"measurementKnown"|"observationError">>
|
||||
/** Additive metadata; no samples means no proof of loss or failure. */
|
||||
export function buildUptimeProbeObservations(rows:readonly Sample[]){return {samples:rows.map(row=>({sampledAt:row.sampledAt,rtt:row.rttMs,loss:row.lossPct,status:row.status})).sort((a,b)=>Date.parse(a.sampledAt)-Date.parse(b.sampledAt))}}
|
||||
export function buildUptimeProbeObservations(rows:readonly Sample[]){return {samples:rows.map(row=>row.measurementKnown===false
|
||||
? {sampledAt:new Date(row.sampledAt).toISOString(),rtt:null,loss:null,status:"unknown" as const,error:row.observationError??"Результат измерения не подтверждён"}
|
||||
: {sampledAt:new Date(row.sampledAt).toISOString(),rtt:row.rttMs,loss:row.lossPct,status:row.status}).sort((a,b)=>Date.parse(a.sampledAt)-Date.parse(b.sampledAt))}}
|
||||
@@ -128,6 +128,7 @@ export async function executeBandwidthSpeedTestInner(params: SpeedTestRunParams)
|
||||
try {
|
||||
const pingRows = await srcClient.ping(initialAddress, 4, srcInterface || undefined, { interval: "1s" })
|
||||
const pr = parsePing(pingRows as Parameters<typeof parsePing>[0])
|
||||
if (!pr.measured) throw new Error("RouterOS не вернул подтверждённые результаты контрольного Ping")
|
||||
afterBtPing = { rttMs: pr.avgRtt, lossPct: pr.loss, error: null }
|
||||
} catch (pe) {
|
||||
afterBtPing = {
|
||||
|
||||
@@ -65,8 +65,8 @@ export interface PingProbeSnapshot {
|
||||
srcInterface: string
|
||||
ok: boolean
|
||||
rttMs: number | null
|
||||
lossPct: number
|
||||
status: "up" | "warn" | "down"
|
||||
lossPct: number | null
|
||||
status: "up" | "warn" | "down" | "unknown"
|
||||
error?: string
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user