fix(traffic): подтверждать свежесть статусов привязанных интерфейсов
This commit is contained in:
1 parent
2b3d583e3b
commit
b4d0733cfa
27 files changed
+1463
-36
No files matched your search
@@ -19,7 +19,7 @@
|
||||
"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",
|
||||
"test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts && tsx src/services/traffic-freshness.test.ts && tsx src/services/traffic-live.test.ts && tsx src/services/traffic-observation-sql.test.ts && tsx src/services/traffic-history-sql.test.ts && tsx src/services/traffic-live-sql.test.ts && tsx src/services/traffic-volume-sql.test.ts",
|
||||
"test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts && tsx src/services/traffic-freshness.test.ts && tsx src/services/traffic-live.test.ts && tsx src/services/traffic-observation-sql.test.ts && tsx src/services/traffic-history-sql.test.ts && tsx src/services/traffic-live-sql.test.ts && tsx src/services/traffic-volume-sql.test.ts && tsx src/services/traffic-binding-status.test.ts && tsx src/services/traffic-status-sql.test.ts",
|
||||
"test:traffic-flow": "tsx src/routes/traffic-flow-query.test.ts && tsx src/services/traffic-flow-month.test.ts && tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-ipv6-pipeline.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-ifindex.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-dest.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-purge.test.ts && tsx src/services/traffic-flow-geoip.test.ts && tsx src/services/traffic-flow-facts.test.ts && tsx src/services/traffic-flow-facts-filter.test.ts && tsx src/services/traffic-flow-facts-rebuild.test.ts && tsx src/services/statistics-aggregate.test.ts",
|
||||
"test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts && tsx src/modules/users/interface-inventory.test.ts && tsx src/routes/users-catalog.test.ts",
|
||||
"test:pg": "tsx src/db/pg-pool.test.ts && tsx src/db/sql-bind.test.ts && tsx src/db/sqlite-json.test.ts && tsx src/db/traffic-flags.test.ts && tsx src/db/pg-schema.test.ts && tsx src/services/config-revisions.test.ts",
|
||||
|
||||
@@ -12,10 +12,11 @@ assert.equal(encodeTrafficFlags(true, true), 3)
|
||||
assert.equal(encodeTrafficFlags(false, false), 0)
|
||||
assert.equal(trafficFlagRunning(1), true)
|
||||
assert.equal(trafficFlagDisabled(1), false)
|
||||
assert.deepEqual(decodeTrafficSampleFlags(1), { running: true, disabled: false, countersKnown: false })
|
||||
assert.deepEqual(decodeTrafficSampleFlags(0), { running: false, disabled: false, countersKnown: false })
|
||||
assert.deepEqual(decodeTrafficSampleFlags(undefined), { running: false, disabled: false, countersKnown: false })
|
||||
assert.deepEqual(decodeTrafficSampleFlags(1), { running: true, disabled: false, countersKnown: false, statusKnown:false })
|
||||
assert.deepEqual(decodeTrafficSampleFlags(0), { running: false, disabled: false, countersKnown: false, statusKnown:false })
|
||||
assert.deepEqual(decodeTrafficSampleFlags(undefined), { running: false, disabled: false, countersKnown: false, statusKnown:false })
|
||||
assert.equal(encodeTrafficFlags(true,false,true),5)
|
||||
assert.deepEqual(decodeTrafficSampleFlags(5), {running:true,disabled:false,countersKnown:true})
|
||||
assert.deepEqual(decodeTrafficSampleFlags(5), {running:true,disabled:false,countersKnown:true,statusKnown:false})
|
||||
assert.deepEqual(decodeTrafficSampleFlags(encodeTrafficFlags(true,false,true,true)),{running:true,disabled:false,countersKnown:true,statusKnown:true})
|
||||
|
||||
console.log("traffic-flags.test.ts: ok")
|
||||
@@ -1,10 +1,12 @@
|
||||
export const TRAFFIC_FLAG_RUNNING = 1
|
||||
export const TRAFFIC_FLAG_DISABLED = 2
|
||||
export const TRAFFIC_FLAG_COUNTERS_KNOWN = 4
|
||||
export const TRAFFIC_FLAG_STATUS_KNOWN = 8
|
||||
|
||||
export function encodeTrafficFlags(running: boolean, disabled: boolean, countersKnown = false): number {
|
||||
export function encodeTrafficFlags(running: boolean, disabled: boolean, countersKnown = false, statusKnown = false): number {
|
||||
return (running ? TRAFFIC_FLAG_RUNNING : 0) | (disabled ? TRAFFIC_FLAG_DISABLED : 0)
|
||||
| (countersKnown ? TRAFFIC_FLAG_COUNTERS_KNOWN : 0)
|
||||
| (statusKnown ? TRAFFIC_FLAG_STATUS_KNOWN : 0)
|
||||
}
|
||||
|
||||
export function trafficFlagRunning(flags: number | null | undefined): boolean {
|
||||
@@ -19,10 +21,12 @@ export function decodeTrafficSampleFlags(flags: number | null | undefined): {
|
||||
running: boolean
|
||||
disabled: boolean
|
||||
countersKnown: boolean
|
||||
statusKnown: boolean
|
||||
} {
|
||||
return {
|
||||
running: trafficFlagRunning(flags),
|
||||
disabled: trafficFlagDisabled(flags),
|
||||
countersKnown: ((Number(flags) || 0) & TRAFFIC_FLAG_COUNTERS_KNOWN) !== 0,
|
||||
statusKnown: ((Number(flags) || 0) & TRAFFIC_FLAG_STATUS_KNOWN) !== 0,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { trafficBindingStatus } from "./traffic-freshness.js"
|
||||
import { parseTrafficFlag } from "./traffic-observation.js"
|
||||
const now=Date.parse("2026-10-12T12:00:00Z"),iso=new Date(now).toISOString()
|
||||
const last={sampledAt:iso,countersKnown:true,statusKnown:true,running:true,disabled:false}
|
||||
const server={enabled:true,updatedAt:iso},settings={enabled:true,intervalSec:30},attempt={serverUpdatedAt:iso,lastAttemptAt:iso,lastSucceededAt:iso,status:"ok",error:null}
|
||||
assert.equal(trafficBindingStatus(last,false,server,settings,attempt,now),"online")
|
||||
assert.equal(trafficBindingStatus({...last,statusKnown:false},false,server,settings,attempt,now),"unknown")
|
||||
assert.equal(trafficBindingStatus({...last,disabled:true},false,server,settings,attempt,now),"offline")
|
||||
assert.equal(trafficBindingStatus(last,true,server,settings,attempt,now),"unknown")
|
||||
assert.equal(trafficBindingStatus({...last,disabled:true},true,server,settings,attempt,now),"offline")
|
||||
for(const status of["error","pending"])assert.equal(trafficBindingStatus(last,false,server,settings,{...attempt,status},now),"unknown")
|
||||
assert.equal(trafficBindingStatus(last,false,server,settings,attempt,now+90_001),"unknown")
|
||||
assert.equal(trafficBindingStatus(last,false,{...server,enabled:false},settings,attempt,now),"unknown")
|
||||
for(const value of[true,"true","yes"])assert.equal(parseTrafficFlag(value),true)
|
||||
for(const value of[false,"false","no"])assert.equal(parseTrafficFlag(value),false)
|
||||
for(const value of[undefined,null,"bad",{},0,1])assert.throws(()=>parseTrafficFlag(value))
|
||||
console.log("binding native status provenance/TTL/peer configuration/boolean guards tests ok")
|
||||
@@ -6,7 +6,7 @@ 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, shouldIncludeIface } from "./traffic-rate.js"
|
||||
import { observedTrafficRate, parseTrafficCounter } from "./traffic-observation.js"
|
||||
import { observedTrafficRate, parseTrafficCounter, parseTrafficFlag } from "./traffic-observation.js"
|
||||
import { rememberServerIfaces } from "./traffic-flow-ifindex.js"
|
||||
import { recordTrafficCollectionAttempt } from "./traffic-collection-state.js"
|
||||
|
||||
@@ -133,16 +133,18 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
|
||||
rememberServerIfaces(srv.id, ifaces)
|
||||
const prevWave = await readPreviousWave(srv.id)
|
||||
const nowMs = Date.parse(now)
|
||||
if (!Array.isArray(ifaces) || ifaces.some(i => typeof i.name !== "string" || !i.name.trim()) || new Set(ifaces.map(i => i.name)).size !== ifaces.length)
|
||||
throw new Error("RouterOS вернул некорректный каталог интерфейсов")
|
||||
let sumRxMbps = 0
|
||||
let sumTxMbps = 0
|
||||
let includedInterfaces = 0
|
||||
let totalsMeasured = true
|
||||
const rows = ifaces.map((i) => {
|
||||
const interfaceName = i.name ?? "unknown"
|
||||
const interfaceName = i.name!
|
||||
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 running = parseTrafficFlag(i.running)
|
||||
const disabled = parseTrafficFlag(i.disabled)
|
||||
const { rxBps, txBps, measured } = sampleRate(prevWave, waveKey(interfaceName), rxBytes, txBytes, nowMs)
|
||||
if (shouldIncludeIface(interfaceName, running, disabled)) {
|
||||
includedInterfaces++
|
||||
@@ -159,7 +161,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
|
||||
txBytes,
|
||||
rxBps: Math.round(rxBps),
|
||||
txBps: Math.round(txBps),
|
||||
flags: encodeTrafficFlags(running, disabled, true),
|
||||
flags: encodeTrafficFlags(running, disabled, true, true),
|
||||
}
|
||||
})
|
||||
try {
|
||||
@@ -170,7 +172,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
|
||||
if (!interfaceName || !peerPublicKey) continue
|
||||
const rxBytes = toNum(p.rx)
|
||||
const txBytes = toNum(p.tx)
|
||||
const disabled = (p.disabled ?? "false") === "true" || p.disabled === "yes"
|
||||
const disabled = parseTrafficFlag(p.disabled)
|
||||
const running = !disabled
|
||||
const { rxBps, txBps } = sampleRate(
|
||||
prevWave,
|
||||
@@ -188,7 +190,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
|
||||
txBytes,
|
||||
rxBps: Math.round(rxBps),
|
||||
txBps: Math.round(txBps),
|
||||
flags: encodeTrafficFlags(running, disabled, true),
|
||||
flags: encodeTrafficFlags(running, disabled, true, true),
|
||||
})
|
||||
}
|
||||
} catch {
|
||||
|
||||
@@ -4,6 +4,16 @@ export type TrafficCurrentState = "current" | "unknown" | "pending" | "error" |
|
||||
export interface TrafficAttempt {
|
||||
serverUpdatedAt: string | Date; lastAttemptAt: string | Date; lastSucceededAt: string | Date | null; status: string; error: string | null
|
||||
}
|
||||
/** Interface running is a native state; enabled WG peer is only configuration, not handshake proof. */
|
||||
export function trafficBindingStatus(last: { sampledAt: string; countersKnown: boolean; statusKnown: boolean; running: boolean; disabled: boolean } | undefined,
|
||||
peer: boolean, server: { enabled: boolean; updatedAt: string | Date },
|
||||
settings: { enabled: boolean; intervalSec: number } | undefined, attempt: TrafficAttempt | undefined, now: number): "online" | "offline" | "unknown" {
|
||||
if (!last?.countersKnown || !last.statusKnown) return "unknown"
|
||||
const observed = applyTrafficFreshness({ rxNow: true, txNow: true, rxSeries: [], txSeries: [], sampledAt: last.sampledAt }, server, settings, attempt, now)
|
||||
if (observed.currentState !== "current") return "unknown"
|
||||
if (last.disabled || !last.running) return "offline"
|
||||
return peer ? "unknown" : "online"
|
||||
}
|
||||
export function mergeTrafficObservations(parts: TrafficObservations[]): TrafficObservations {
|
||||
const oldest = (key: "sampledAt" | "lastAttemptAt" | "lastSucceededAt" | "expiresAt") => {
|
||||
const times = parts.map(p => Date.parse(p[key] ?? ""))
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { observedTrafficRate, parseTrafficCounter } from "./traffic-observation.js"
|
||||
import { observedTrafficRate, parseTrafficCounter, parseTrafficFlag } from "./traffic-observation.js"
|
||||
import { collectSourceSignals } from "./alert-engine/collector-source-signals.js"
|
||||
import fs from "node:fs"
|
||||
import vm from "node:vm"
|
||||
@@ -41,7 +41,7 @@ assert.equal(ast.statements.filter(n => ts.isFunctionDeclaration(n) && n.name &&
|
||||
let inserted: unknown[] = [], wave = new Map<string, typeof previous>(), counters: unknown = "1000"
|
||||
let included = true
|
||||
const context = vm.createContext({
|
||||
Date, observedTrafficRate, parseTrafficCounter, bpsToMbps, shouldIncludeIface,
|
||||
Date, observedTrafficRate, parseTrafficCounter, parseTrafficFlag, bpsToMbps, shouldIncludeIface,
|
||||
SCHEDULER_RUN_SNAPSHOT_VERSION: 1, servers: {}, trafficSamples: {}, trafficSettings: {}, eq: () => ({}),
|
||||
getSettingsRow: async () => ({}), readPreviousWave: async () => wave, rememberServerIfaces: () => {},
|
||||
encodeTrafficFlags: () => 1,
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
import { rateBpsFromDelta } from "./traffic-rate.js"
|
||||
export function parseTrafficFlag(raw: unknown): boolean {
|
||||
if (raw === true || raw === "true" || raw === "yes") return true
|
||||
if (raw === false || raw === "false" || raw === "no") return false
|
||||
throw new Error("RouterOS вернул неизвестное состояние интерфейса")
|
||||
}
|
||||
|
||||
/** RouterOS byte counters must be exact nonnegative integers, never fallback zero. */
|
||||
export function parseTrafficCounter(raw: unknown): number {
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
import assert from "node:assert/strict"
|
||||
import {createServer} from "node:http"
|
||||
import Fastify from "fastify"
|
||||
import {pool,closePool} from "../db/index.js"
|
||||
import {initDatabase} from "../db/bootstrap.js"
|
||||
import {collectTrafficOnce} from "./traffic-collector.js"
|
||||
import trafficRoutes from "../routes/traffic.js"
|
||||
const database=process.env.MMAPP_OWN_SQL_DB
|
||||
if(!database)console.log("traffic status 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-status181-"+Date.now()+"-",userId=prefix+"user";let serverId:number|undefined,cases=0,deny=false,flag:unknown=true,peerDisabled=false
|
||||
const http=createServer((req,res)=>{res.setHeader("Content-Type","application/json");if(deny){res.statusCode=401;return res.end('{"message":"Unauthorized"}')}res.end(JSON.stringify(req.url?.includes("/wireguard/")?[{interface:"wg1","public-key":"fixturekey",rx:"1000",tx:"1000",disabled:peerDisabled}]:[{name:"ether1",running:flag,disabled:false,"rx-byte":"1000","tx-byte":"1000"}]))})
|
||||
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();await app.register(trafficRoutes)
|
||||
const bindings=async()=>{const r=await app.inject({method:"GET",url:"/traffic/bound-interfaces?range=1h"});assert.equal(r.statusCode,200);return r.json().interfaces.filter((i:{userId:string})=>i.userId===userId)}
|
||||
try{
|
||||
await initDatabase();const updated=new Date(Date.now()-3600_000).toISOString()
|
||||
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,updated])).rows[0].id)
|
||||
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")
|
||||
await pool.query("INSERT INTO app_users(id,login,name) VALUES($1,$2,'Status fixture')",[userId,prefix+"login"])
|
||||
for(const[name,key]of[["ether1",""],["wg1","fixturekey"]])await pool.query("INSERT INTO user_interface_bindings(id,user_id,server_id,interface_name,interface_type,peer_public_key) VALUES($1,$2,$3,$4,$5,$6)",[prefix+name,userId,serverId,name,key?"wg":"ether",key])
|
||||
await collectTrafficOnce();let rows=await bindings();assert.equal(rows.find((r:{interfaceName:string})=>r.interfaceName==="ether1").status,"online");assert(rows[0].statusObservedAt&&rows[0].statusExpiresAt);cases++
|
||||
assert.equal(rows.find((r:{interfaceName:string})=>r.interfaceName==="wg1").status,"unknown");cases++
|
||||
peerDisabled=true;await collectTrafficOnce();rows=await bindings();assert.equal(rows.find((r:{interfaceName:string})=>r.interfaceName==="wg1").status,"offline");cases++
|
||||
flag=false;await collectTrafficOnce();rows=await bindings();assert.equal(rows.find((r:{interfaceName:string})=>r.interfaceName==="ether1").status,"offline");cases++
|
||||
flag="bad";let snap=await collectTrafficOnce();assert.equal(snap.servers[0].ok,false);rows=await bindings();assert(rows.every((r:{status:string})=>r.status==="unknown"));cases++
|
||||
flag=true;deny=true;await collectTrafficOnce();rows=await bindings();assert(rows.every((r:{status:string})=>r.status==="unknown"));cases++
|
||||
deny=false;await collectTrafficOnce();await pool.query("UPDATE traffic_samples SET flags=flags & 7 WHERE server_id=$1",[serverId]);rows=await bindings();assert(rows.every((r:{status:string})=>r.status==="unknown"));cases++
|
||||
await collectTrafficOnce();await pool.query("UPDATE traffic_collection_state SET last_attempt_at=$1 WHERE server_id=$2",[new Date(Date.now()-100_000).toISOString(),serverId]);rows=await bindings();assert(rows.every((r:{status:string})=>r.status==="unknown"));cases++
|
||||
await pool.query("UPDATE servers SET enabled=false WHERE id=$1",[serverId]);rows=await bindings();assert(rows.every((r:{status:string})=>r.status==="unknown"));cases++
|
||||
console.log(`traffic status owned SQL: ${cases} cases; first-wave native flags/peer config/invalid/error/legacy/stale/disabled passed`)
|
||||
}finally{await app.close();await new Promise<void>(resolve=>http.close(()=>resolve()));await pool.query("DELETE FROM app_users WHERE id=$1",[userId]);if(serverId!==undefined){await pool.query("DELETE FROM traffic_samples WHERE server_id=$1",[serverId]);await pool.query("DELETE FROM servers WHERE id=$1",[serverId])}await closePool()}
|
||||
}
|
||||
@@ -8,7 +8,7 @@ import {
|
||||
type BuiltTrafficSeries,
|
||||
} from "./traffic-rate.js"
|
||||
import { readTrafficCollectionContext } from "./traffic-collection-state.js"
|
||||
import { applyTrafficFreshness } from "./traffic-freshness.js"
|
||||
import { applyTrafficFreshness, trafficBindingStatus } from "./traffic-freshness.js"
|
||||
import type { TrafficObservations } from "./traffic-rate.js"
|
||||
|
||||
export interface BoundIfaceTrafficDto {
|
||||
@@ -36,7 +36,9 @@ export interface BoundIfaceTrafficDto {
|
||||
txTotal: number
|
||||
rxSeries: number[]
|
||||
txSeries: number[]
|
||||
status: "online" | "offline"
|
||||
status: "online" | "offline" | "unknown"
|
||||
statusObservedAt: string | null
|
||||
statusExpiresAt: string | null
|
||||
}
|
||||
|
||||
export interface UserTrafficDto {
|
||||
@@ -99,9 +101,9 @@ export async function buildUserTrafficList(rangeStartMs: number, rangeEndMs: num
|
||||
const last = [...rows.filter((r) =>
|
||||
r.interfaceName === b.interfaceName && (r.peerPublicKey ?? "") === peerKey,
|
||||
)]
|
||||
.sort((a, c) => a.sampledAt.localeCompare(c.sampledAt))
|
||||
.sort((a, c) => Date.parse(a.sampledAt) - Date.parse(c.sampledAt))
|
||||
.at(-1)
|
||||
const running = Boolean(last?.running) && !last?.disabled
|
||||
const status = trafficBindingStatus(last, !!peerKey, server ?? {enabled:false,updatedAt:""}, context.settings, context.states.get(b.serverId), rangeEndMs)
|
||||
interfaces.push({
|
||||
id: `${b.userId}:${b.serverId}:${b.interfaceName}:${peerKey || "_iface"}`,
|
||||
bindingId: b.id,
|
||||
@@ -118,7 +120,9 @@ export async function buildUserTrafficList(rangeStartMs: number, rangeEndMs: num
|
||||
serverSite: b.serverSite,
|
||||
serverCountry: b.serverCountry,
|
||||
...seriesFromBuilt(built),
|
||||
status: running && server?.enabled ? "online" : "offline",
|
||||
status,
|
||||
statusObservedAt: status !== "unknown" && last ? new Date(last.sampledAt).toISOString() : null,
|
||||
statusExpiresAt: status !== "unknown" ? built.observations?.expiresAt ?? null : null,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user