fix(traffic): учитывать состав каждой волны измерений

This commit is contained in:
Denozordec committed 2026-10-13 01:52:50 +07:00
1 parent a8b80d9914
commit 5d19b6a258
17 files changed
+1349 -18

No files matched your search

+1 -1
View File
@@ -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 && tsx src/services/traffic-binding-status.test.ts && tsx src/services/traffic-status-sql.test.ts && tsx src/services/traffic-gap.test.ts && tsx src/services/traffic-identity.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 && tsx src/services/traffic-gap.test.ts && tsx src/services/traffic-identity.test.ts && tsx src/services/traffic-membership.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",
@@ -0,0 +1,12 @@
import assert from 'node:assert/strict'
import {buildTrafficFromSamples} from './traffic-rate.js'
const at=Date.parse('2026-10-13T00:00:00Z')
const sample=(name:string,t:number,bytes:number,known=true)=>({interfaceName:name,sampledAt:new Date(at+t).toISOString(),rxBytes:bytes,txBytes:bytes,rxBps:0,txBps:0,running:true,disabled:false,countersKnown:true,continuityKnown:known,nativeCounterIdentity:name})
const calc=(rows:ReturnType<typeof sample>[])=>buildTrafficFromSamples(rows,at,at+90_000,undefined,undefined,true)
const removed=calc([sample('a',0,0,false),sample('a',30_000,1024),sample('b',0,0,false),sample('b',30_000,1024),sample('b',60_000,2048),sample('b',90_000,3072)])
assert.equal(removed.observations!.rxNow,true);assert.equal(removed.sessions,1);assert.equal(removed.rxTotalGiB,4096/1024**3);assert(removed.observations!.rxSeries.some(Boolean))
const first=calc([sample('a',0,0,false),sample('a',30_000,0),sample('a',60_000,0),sample('new',60_000,0,false)])
assert.equal(first.observations!.rxNow,false);assert.equal(first.sessions,2)
const next=calc([sample('a',0,0,false),sample('a',30_000,0),sample('a',60_000,0),sample('a',90_000,0),sample('new',60_000,0,false),sample('new',90_000,0)])
assert.equal(next.observations!.rxNow,true);assert.equal(next.rxNow,0);assert.equal(next.sessions,2)
console.log('wave membership removal/addition/baseline/zero/history/active count passed')
+20 -5
View File
@@ -147,6 +147,8 @@ export function buildTrafficFromSamples(
const txPoints: Array<{ t: number; v: number }> = []
const byTs = new Map<number, { rx: number; tx: number;rxObserved:boolean;txObserved:boolean;count:number;intervalMs:number }>()
let includedSeries=0,lastAttemptAt:number|null=null
const membership = new Map<number, {count:number; declared:boolean; active:number}>()
let latestMembershipAt = -Infinity
let rxBytesDelta = 0
let txBytesDelta = 0
@@ -170,6 +172,15 @@ export function buildTrafficFromSamples(
const sorted = [...arr].sort((a, b) => Date.parse(a.sampledAt) - Date.parse(b.sampledAt))
const last = sorted[sorted.length - 1]
if (!last) continue
for (const row of sorted) {
const t = Date.parse(row.sampledAt)
if (!Number.isFinite(t) || t < rangeStartMs || t > rangeEndMs) continue
const wave = membership.get(t) ?? {count:0,declared:true,active:0}
wave.count++; wave.declared &&= typeof row.continuityKnown === "boolean"
if (row.running && !row.disabled) wave.active++
membership.set(t,wave)
latestMembershipAt = Math.max(latestMembershipAt,t)
}
// Recorded counter deltas describe the period, independent of the latest link state.
includedSeries++
@@ -183,7 +194,8 @@ export function buildTrafficFromSamples(
const t1 = parseIsoMs(cur.sampledAt)
if(Number.isFinite(Date.parse(cur.sampledAt))&&t1>=rangeStartMs&&t1<=rangeEndMs)lastAttemptAt=Math.max(lastAttemptAt??t1,t1)
const sameIdentity = (prev.nativeCounterIdentity ?? null) === (cur.nativeCounterIdentity ?? null)
const countersKnown = prev.countersKnown !== false && cur.countersKnown !== false && cur.continuityKnown !== false && sameIdentity
const identityPresent = cur.continuityKnown === undefined || typeof cur.nativeCounterIdentity === "string" && cur.nativeCounterIdentity.length > 0
const countersKnown = prev.countersKnown !== false && cur.countersKnown !== false && cur.continuityKnown !== false && identityPresent && sameIdentity
const rxBps = countersKnown ? rateBpsFromDelta(prev.rxBytes, cur.rxBytes, t0, t1) : null
const txBps = countersKnown ? rateBpsFromDelta(prev.txBytes, cur.txBytes, t0, t1) : null
if (t1 >= rangeStartMs && t1 <= rangeEndMs) {
@@ -206,6 +218,9 @@ export function buildTrafficFromSamples(
rxPoints.push({ t, v: v.rx })
txPoints.push({ t, v: v.tx })
}
const expectedAt = (t:number) => membership.get(t)?.declared ? membership.get(t)!.count : includedSeries
const latestWave = membership.get(latestMembershipAt)
if (latestWave?.declared) sessions = latestWave.active
const rxSeries = bucketAvg(rxPoints, rangeStartMs, rangeEndMs)
const txSeries = bucketAvg(txPoints, rangeStartMs, rangeEndMs)
@@ -225,10 +240,10 @@ export function buildTrafficFromSamples(
rxSeries,
txSeries,
...(includeObservations?{observations:{
rxNow:lastAttemptAt!==null&&lastTs===lastAttemptAt&&!!last?.rxObserved&&last.count===includedSeries,
txNow:lastAttemptAt!==null&&lastTs===lastAttemptAt&&!!last?.txObserved&&last.count===includedSeries,
rxSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.rxObserved&&v.count===includedSeries&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1),
txSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.txObserved&&v.count===includedSeries&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1),
rxNow:lastAttemptAt!==null&&lastTs===lastAttemptAt&&!!last?.rxObserved&&last.count===expectedAt(lastTs!),
txNow:lastAttemptAt!==null&&lastTs===lastAttemptAt&&!!last?.txObserved&&last.count===expectedAt(lastTs!),
rxSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.rxObserved&&v.count===expectedAt(t)&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1),
txSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.txObserved&&v.count===expectedAt(t)&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1),
sampledAt:lastAttemptAt===null?null:new Date(lastAttemptAt).toISOString(),
measurementIntervalMs:lastTs===lastAttemptAt?last?.intervalMs??null:null,
}}:{}),
@@ -46,6 +46,15 @@ else {
s=(await get(`/traffic/servers/${serverId}?range=1h`)).server;assert.equal(s.rxTotal,1/1024);assert.equal(s.observations.rxNow,false);assert.equal(s.observations.measurementIntervalMs,120_000);assert(s.observations.error.includes("Интервал"));assert(s.observations.rxSeries.some(Boolean));cases++;
await pool.query("UPDATE traffic_samples SET sampled_at=$1 WHERE server_id=$2 AND sampled_at=$3",[new Date(Date.parse(at)-30_000).toISOString(),serverId,at]);await put("ether1",0,2*1024**2);
s=(await get(`/traffic/servers/${serverId}?range=1h`)).server;assert.equal(s.observations.rxNow,true);assert.equal(s.observations.measurementIntervalMs,30_000);assert.equal(s.rxTotal,2/1024);cases++;
for(const changed of["deleted","renamed"]){
await pool.query("DELETE FROM traffic_samples WHERE server_id=$1",[serverId]);
for(const[name,offset,bytes]of[["ether1",-90_000,0],["ether1",-60_000,1024],["ether2",-90_000,0],["ether2",-60_000,1024],[changed==="renamed"?"newName":"ether2",-30_000,2048],[changed==="renamed"?"newName":"ether2",0,3072]] as const)await put(name,offset,bytes);
await pool.query("UPDATE traffic_samples SET flags=61,native_counter_identity=CASE WHEN interface_name='newName' THEN 'ether2' ELSE interface_name END WHERE server_id=$1",[serverId]);
if(changed==="renamed")await pool.query("UPDATE traffic_samples SET flags=45 WHERE server_id=$1 AND interface_name='newName' AND sampled_at=$2",[serverId,new Date(Date.parse(at)-30_000).toISOString()]);
s=(await get(`/traffic/servers/${serverId}?range=1h`)).server;assert.equal(s.observations.rxNow,true);assert.equal(s.sessions,1);assert.equal(s.rxTotal,(changed==="renamed"?3072:4096)/1024**3);assert(s.observations.rxSeries.some(Boolean));cases++;
}
await put("newInterface",0,0);await pool.query("UPDATE traffic_samples SET flags=45 WHERE server_id=$1 AND interface_name='newInterface'",[serverId]);
s=(await get(`/traffic/servers/${serverId}?range=1h`)).server;assert.equal(s.observations.rxNow,false);assert.equal(s.sessions,2);cases++;
console.log(`traffic volume owned SQL: ${cases} cases; exact bytes/binary unit/server/user/binding/100-part aggregate/reset passed`)
}finally{await app.close();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()}
}