fix(traffic): исключить ложные нули из суммы трафика
This commit is contained in:
1 parent
8fb46b1552
commit
4527e5537c
12 files changed
+280
-27
No files matched your search
@@ -5,7 +5,8 @@ import { decodeTrafficSampleFlags, encodeTrafficFlags } from "../db/traffic-flag
|
||||
import type { TrafficRunSnapshot } from "../types/scheduler-run-snapshot.js"
|
||||
import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
|
||||
import { MikrotikClient } from "./mikrotik.js"
|
||||
import { bpsToMbps, rateBpsFromDelta, shouldIncludeIface } from "./traffic-rate.js"
|
||||
import { bpsToMbps, shouldIncludeIface } from "./traffic-rate.js"
|
||||
import { observedTrafficRate, parseTrafficCounter } from "./traffic-observation.js"
|
||||
import { rememberServerIfaces } from "./traffic-flow-ifindex.js"
|
||||
|
||||
interface RosIfaceTraffic {
|
||||
@@ -40,16 +41,8 @@ function sampleRate(
|
||||
rxBytes: number,
|
||||
txBytes: number,
|
||||
nowMs: number,
|
||||
): { rxBps: number; txBps: number } {
|
||||
const prev = prevWave.get(key)
|
||||
const prevMs = prev ? Date.parse(prev.sampledAt) : NaN
|
||||
const rxBps = prev && Number.isFinite(prevMs)
|
||||
? (rateBpsFromDelta(prev.rxBytes, rxBytes, prevMs, nowMs) ?? 0)
|
||||
: 0
|
||||
const txBps = prev && Number.isFinite(prevMs)
|
||||
? (rateBpsFromDelta(prev.txBytes, txBytes, prevMs, nowMs) ?? 0)
|
||||
: 0
|
||||
return { rxBps, txBps }
|
||||
): { rxBps: number; txBps: number; measured: boolean } {
|
||||
return observedTrafficRate(prevWave.get(key), rxBytes, txBytes, nowMs)
|
||||
}
|
||||
|
||||
export interface TrafficCollectorState {
|
||||
@@ -61,8 +54,7 @@ export interface TrafficCollectorState {
|
||||
let collecting = false
|
||||
|
||||
function toNum(raw: unknown): number {
|
||||
const n = Number.parseFloat(String(raw ?? "0"))
|
||||
return Number.isFinite(n) ? Math.max(0, Math.round(n)) : 0
|
||||
return parseTrafficCounter(raw)
|
||||
}
|
||||
|
||||
async function getSettingsRow() {
|
||||
@@ -139,14 +131,18 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
|
||||
const nowMs = Date.parse(now)
|
||||
let sumRxMbps = 0
|
||||
let sumTxMbps = 0
|
||||
let includedInterfaces = 0
|
||||
let totalsMeasured = true
|
||||
const rows = ifaces.map((i) => {
|
||||
const interfaceName = i.name ?? "unknown"
|
||||
const rxBytes = toNum(i["rx-byte"])
|
||||
const txBytes = toNum(i["tx-byte"])
|
||||
const running = (i.running ?? "false") === "true"
|
||||
const disabled = (i.disabled ?? "false") === "true"
|
||||
const { rxBps, txBps } = sampleRate(prevWave, waveKey(interfaceName), rxBytes, txBytes, nowMs)
|
||||
const { rxBps, txBps, measured } = sampleRate(prevWave, waveKey(interfaceName), rxBytes, txBytes, nowMs)
|
||||
if (shouldIncludeIface(interfaceName, running, disabled)) {
|
||||
includedInterfaces++
|
||||
totalsMeasured &&= measured
|
||||
sumRxMbps += bpsToMbps(rxBps)
|
||||
sumTxMbps += bpsToMbps(txBps)
|
||||
}
|
||||
@@ -203,8 +199,10 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
|
||||
host: srv.host,
|
||||
ok: true,
|
||||
interfaces: ifaces.length,
|
||||
sumRxMbps: Math.round(sumRxMbps * 1000) / 1000,
|
||||
sumTxMbps: Math.round(sumTxMbps * 1000) / 1000,
|
||||
...(includedInterfaces > 0 && totalsMeasured ? {
|
||||
sumRxMbps: Math.round(sumRxMbps * 1000) / 1000,
|
||||
sumTxMbps: Math.round(sumTxMbps * 1000) / 1000,
|
||||
} : {}),
|
||||
})
|
||||
} catch (err) {
|
||||
snapshot.servers.push({
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { observedTrafficRate, parseTrafficCounter } from "./traffic-observation.js"
|
||||
import { collectSourceSignals } from "./alert-engine/collector-source-signals.js"
|
||||
import fs from "node:fs"
|
||||
import vm from "node:vm"
|
||||
import ts from "typescript"
|
||||
import { bpsToMbps, shouldIncludeIface } from "./traffic-rate.js"
|
||||
|
||||
const now = Date.parse("2026-10-12T12:00:00Z"), iso = new Date(now).toISOString()
|
||||
const previous = { rxBytes: 1000, txBytes: 1000, sampledAt: new Date(now - 30_000).toISOString() }
|
||||
for (const value of [undefined, null, "", "123bad", "1.5", "-1", -1, NaN, Infinity, Number.MAX_SAFE_INTEGER + 1, {}, " 10 "]) {
|
||||
assert.throws(() => parseTrafficCounter(value), /некорректный счётчик/)
|
||||
}
|
||||
for (const value of [0, "0", "1000", Number.MAX_SAFE_INTEGER]) assert.equal(parseTrafficCounter(value), Number(value))
|
||||
const cases = [
|
||||
observedTrafficRate(undefined, 1000, 1000, now),
|
||||
observedTrafficRate(previous, 10, 10, now),
|
||||
observedTrafficRate(previous, 10, 1100, now),
|
||||
observedTrafficRate({ ...previous, sampledAt: iso }, 1100, 1100, now),
|
||||
observedTrafficRate({ ...previous, sampledAt: "invalid" }, 1100, 1100, now),
|
||||
observedTrafficRate(previous, 1000, 1000, now),
|
||||
observedTrafficRate(previous, 3_751_000, 3_751_000, now),
|
||||
]
|
||||
assert.deepEqual(cases.map(r => r.measured), [false, false, false, false, false, true, true])
|
||||
assert.equal(cases[5].rxBps, 0); assert.equal(cases[6].rxBps, 1_000_000)
|
||||
for (const rate of cases) {
|
||||
const row = { serverId: 1, name: "audit", host: "audit.example", ok: true,
|
||||
...(rate.measured ? { sumRxMbps: rate.rxBps / 1_000_000, sumTxMbps: rate.txBps / 1_000_000 } : {}) }
|
||||
const signals = collectSourceSignals({ servers: [{ id: 1, name: "audit", host: "audit.example", enabled: true }], probes: [], traffic: { enabled: true, intervalSec: 30 } },
|
||||
{ resources: [], rest: [], ping: [], traffic: [{ finishedAt: iso, resultJson: { v: 1, job: "traffic", sampledAt: iso, servers: [row] } }] }, now)
|
||||
assert.equal(signals.trafficByServer.length, Number(rate.measured))
|
||||
}
|
||||
console.log("traffic observation counters/rates/source confidence tests ok")
|
||||
|
||||
// Execute the actual collector body with in-memory DB/HTTP boundaries, not a copied sum algorithm.
|
||||
const collector = fs.readFileSync(new URL("./traffic-collector.ts", import.meta.url), "utf8")
|
||||
const ast = ts.createSourceFile("traffic-collector.ts", collector, ts.ScriptTarget.Latest, true)
|
||||
const functions = ast.statements.filter(n => ts.isFunctionDeclaration(n) && n.name &&
|
||||
["waveKey", "sampleRate", "toNum", "collectTrafficOnce"].includes(n.name.text)).map(n => n.getText(ast)).join("\n")
|
||||
assert.equal(ast.statements.filter(n => ts.isFunctionDeclaration(n) && n.name && n.name.text === "collectTrafficOnce").length, 1)
|
||||
let inserted: unknown[] = [], wave = new Map<string, typeof previous>(), counters: unknown = "1000"
|
||||
let included = true
|
||||
const context = vm.createContext({
|
||||
Date, observedTrafficRate, parseTrafficCounter, bpsToMbps, shouldIncludeIface,
|
||||
SCHEDULER_RUN_SNAPSHOT_VERSION: 1, servers: {}, trafficSamples: {}, trafficSettings: {}, eq: () => ({}),
|
||||
getSettingsRow: async () => ({}), readPreviousWave: async () => wave, rememberServerIfaces: () => {},
|
||||
encodeTrafficFlags: () => 1,
|
||||
MikrotikClient: { fromServer: () => ({ get: async (path: string) => path === "/interface" ?
|
||||
[{ name: "ether1", running: String(included), disabled: "false", "rx-byte": counters, "tx-byte": counters }] : [] }) },
|
||||
db: {
|
||||
select: () => ({ from: () => ({ where: async () => [{ id: 1, name: "audit", host: "audit.example", enabled: true }] }) }),
|
||||
insert: () => ({ values: async (rows: unknown[]) => { inserted.push(...rows) } }),
|
||||
update: () => ({ set: () => ({ where: async () => {} }) }),
|
||||
},
|
||||
})
|
||||
vm.runInContext(ts.transpileModule("let collecting=false;\n" + functions.replace("export async function", "async function") +
|
||||
"\nglobalThis.run=collectTrafficOnce;", { compilerOptions: { target: ts.ScriptTarget.ES2022 } }).outputText, context)
|
||||
let snapshot = await context.run()
|
||||
assert.equal(snapshot.servers[0].ok, true); assert.equal(snapshot.servers[0].sumRxMbps, undefined); assert.equal(inserted.length, 1)
|
||||
wave = new Map([["ether1\0", { ...previous, sampledAt: new Date(Date.now() - 30_000).toISOString() }]])
|
||||
snapshot = await context.run(); assert.equal(snapshot.servers[0].sumRxMbps, 0)
|
||||
counters = "10"; snapshot = await context.run(); assert.equal(snapshot.servers[0].sumRxMbps, undefined)
|
||||
counters = "bad"; inserted = []; snapshot = await context.run()
|
||||
assert.equal(snapshot.servers[0].ok, false); assert.equal(inserted.length, 0); assert.match(snapshot.servers[0].error, /некорректный счётчик/)
|
||||
counters = "1000"; included = false; snapshot = await context.run(); assert.equal(snapshot.servers[0].sumRxMbps, undefined)
|
||||
console.log("actual traffic collector first-wave/zero/reset/invalid/excluded totals tests ok")
|
||||
@@ -0,0 +1,22 @@
|
||||
import { rateBpsFromDelta } from "./traffic-rate.js"
|
||||
|
||||
/** RouterOS byte counters must be exact nonnegative integers, never fallback zero. */
|
||||
export function parseTrafficCounter(raw: unknown): number {
|
||||
if (typeof raw !== "number" && (typeof raw !== "string" || !/^\d+$/.test(raw))) {
|
||||
throw new Error("RouterOS вернул некорректный счётчик трафика")
|
||||
}
|
||||
const n = Number(raw)
|
||||
if (!Number.isSafeInteger(n) || n < 0) throw new Error("RouterOS вернул некорректный счётчик трафика")
|
||||
return n
|
||||
}
|
||||
|
||||
export function observedTrafficRate(
|
||||
previous: { rxBytes: number; txBytes: number; sampledAt: string } | undefined,
|
||||
rxBytes: number, txBytes: number, nowMs: number,
|
||||
): { rxBps: number; txBps: number; measured: boolean } {
|
||||
const prevMs = previous ? Date.parse(previous.sampledAt) : NaN
|
||||
const rx = previous ? rateBpsFromDelta(previous.rxBytes, rxBytes, prevMs, nowMs) : null
|
||||
const tx = previous ? rateBpsFromDelta(previous.txBytes, txBytes, prevMs, nowMs) : null
|
||||
// Stored rates retain the legacy numeric shape; observed totals require both deltas.
|
||||
return { rxBps: rx ?? 0, txBps: tx ?? 0, measured: rx !== null && tx !== null }
|
||||
}
|
||||
Reference in new issue
Block a user