Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
97e43b2335 |
@@ -1,6 +1,7 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { getStatistics, getStatisticsPivot, parseStatisticsPeriod, pivotDimsConflict } from "./statistics-aggregate.js"
|
||||
import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js"
|
||||
import { setRefreshIfacesForTests } from "./traffic-flow-ifaces.js"
|
||||
import { withPgOrSkip } from "../test/pg.js"
|
||||
import { dbQuery } from "../db/index.js"
|
||||
import { ensurePartitionFor } from "../db/partitions.js"
|
||||
@@ -52,6 +53,7 @@ await dbQuery(`
|
||||
|
||||
resetIfaceCacheForTests()
|
||||
rememberServerIfaces(serverId, [{ name: "gre-client", ifindex: "2" }])
|
||||
setRefreshIfacesForTests(async () => {})
|
||||
|
||||
await dbQuery(`
|
||||
INSERT INTO flow_daily_facts (server_id, day, iface, country, service, asn, bytes, packets)
|
||||
@@ -72,6 +74,11 @@ try {
|
||||
assert.ok(unbound)
|
||||
assert.equal(unbound.bytes, 70)
|
||||
assert.ok(all.servers.some((r) => r.id === String(serverId)))
|
||||
const greIface = all.interfaces.find((r) => r.label.includes("gre-client"))
|
||||
assert.ok(greIface)
|
||||
assert.equal(greIface.bytes, 1000)
|
||||
assert.equal(greIface.id, `${serverId}:gre-client`)
|
||||
assert.ok(!all.interfaces.some((r) => /· (?:#)?\d+$/.test(r.label)))
|
||||
|
||||
const sliced = await getStatistics({
|
||||
from: "2026-09-01",
|
||||
@@ -118,6 +125,7 @@ try {
|
||||
assert.equal(hourly.kpis.bytes, 40)
|
||||
assert.ok(hourly.users.some((r) => r.id === "u-stats-1"))
|
||||
} finally {
|
||||
setRefreshIfacesForTests(null)
|
||||
resetIfaceCacheForTests()
|
||||
await dbQuery(`DELETE FROM flow_daily_facts WHERE server_id = $1`, [serverId])
|
||||
await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id = $1`, [serverId])
|
||||
|
||||
@@ -11,10 +11,12 @@ import {
|
||||
type StatisticsQuery,
|
||||
} from "@mmapp/contracts/statistics"
|
||||
import {
|
||||
bindingIfaceAliases,
|
||||
bindingIfaceAliasesAllServers,
|
||||
collapseServerIfaceRows,
|
||||
displayFactIface,
|
||||
expandBindingIfaces,
|
||||
factIfaceAliases,
|
||||
} from "./traffic-flow-ifindex.js"
|
||||
import { refreshServerIfaces } from "./traffic-flow-ifaces.js"
|
||||
|
||||
const TOP_N = 200
|
||||
const HOUR_WINDOW_MS = 48 * 3600_000
|
||||
@@ -92,10 +94,31 @@ interface FilterCtx {
|
||||
}
|
||||
|
||||
function ifaceFilterAliases(iface: string, serverId?: number): string[] {
|
||||
return factIfaceAliases(iface.trim(), serverId)
|
||||
}
|
||||
|
||||
function looksLikeIfIndex(iface: string): boolean {
|
||||
const raw = iface.trim()
|
||||
if (!raw) return []
|
||||
if (serverId != null) return bindingIfaceAliases(serverId, raw)
|
||||
return bindingIfaceAliasesAllServers(raw)
|
||||
return /^\d+$/.test(raw) || /^#\d+$/.test(raw)
|
||||
}
|
||||
|
||||
async function warmIfaceCache(ids: Iterable<number>): Promise<void> {
|
||||
const uniq = [...new Set(ids)].filter((id) => Number.isFinite(id) && id > 0)
|
||||
if (!uniq.length) return
|
||||
await Promise.all(uniq.map((id) => refreshServerIfaces(id)))
|
||||
}
|
||||
|
||||
async function warmBindingIfaceCache(): Promise<void> {
|
||||
const rows = await db.select({ serverId: userInterfaceBindings.serverId }).from(userInterfaceBindings)
|
||||
await warmIfaceCache(rows.map((r) => r.serverId))
|
||||
}
|
||||
|
||||
function canonicalIfaceDimId(id: string): string {
|
||||
const colon = id.indexOf(":")
|
||||
if (colon < 0) return id
|
||||
const sid = Number(id.slice(0, colon))
|
||||
if (!Number.isFinite(sid)) return id
|
||||
return `${sid}:${displayFactIface(sid, id.slice(colon + 1))}`
|
||||
}
|
||||
|
||||
function factWhere(alias: string, grain: "hour" | "day", ctx: FilterCtx): { sql: string; params: unknown[] } {
|
||||
@@ -214,7 +237,7 @@ async function loadBindUserTuples(): Promise<UserBindTuple[]> {
|
||||
const seen = new Set<string>()
|
||||
const out: UserBindTuple[] = []
|
||||
for (const b of binds) {
|
||||
for (const iface of bindingIfaceAliases(b.serverId, b.interfaceName)) {
|
||||
for (const iface of factIfaceAliases(b.interfaceName, b.serverId)) {
|
||||
const k = `${b.userId}\0${b.serverId}\0${iface}`
|
||||
if (seen.has(k)) continue
|
||||
seen.add(k)
|
||||
@@ -281,6 +304,8 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
|
||||
windowSec: 1,
|
||||
})
|
||||
|
||||
await warmBindingIfaceCache()
|
||||
if (query.serverId) await warmIfaceCache([query.serverId])
|
||||
const bindTuples = await loadBindUserTuples()
|
||||
const ctx = await buildFilterCtx(query, period)
|
||||
if (!ctx) return emptyDto(period)
|
||||
@@ -289,12 +314,11 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
|
||||
const timeCol = period.grain === "hour" ? "bucket_at" : "day"
|
||||
const where = factWhere("f", period.grain, ctx)
|
||||
|
||||
const totals = await dbAll<{ bytes: number; packets: number; servers: number; ifaces: number }>(`
|
||||
const totals = await dbAll<{ bytes: number; packets: number; servers: number }>(`
|
||||
SELECT
|
||||
COALESCE(SUM(f.bytes), 0) AS bytes,
|
||||
COALESCE(SUM(f.packets), 0) AS packets,
|
||||
COUNT(DISTINCT f.server_id)::int AS servers,
|
||||
COUNT(DISTINCT (f.server_id::text || ':' || f.iface))::int AS ifaces
|
||||
COUNT(DISTINCT f.server_id)::int AS servers
|
||||
FROM ${table} f
|
||||
WHERE ${where.sql}
|
||||
`, where.params)
|
||||
@@ -302,7 +326,6 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
|
||||
const bytes = Number(totals[0]?.bytes) || 0
|
||||
const packets = Number(totals[0]?.packets) || 0
|
||||
const serverCount = Number(totals[0]?.servers) || 0
|
||||
const ifaceCount = Number(totals[0]?.ifaces) || 0
|
||||
|
||||
const seriesRows = await dbAll<{ t: string; bytes: number }>(`
|
||||
SELECT ${timeCol}::text AS t, SUM(f.bytes) AS bytes
|
||||
@@ -340,12 +363,15 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
|
||||
GROUP BY f.server_id
|
||||
`, where.params)
|
||||
|
||||
const ifaceRows = await dbAll<{ serverId: number; iface: string; bytes: number; packets: number }>(`
|
||||
const ifaceRowsRaw = await dbAll<{ serverId: number; iface: string; bytes: number; packets: number }>(`
|
||||
SELECT f.server_id AS "serverId", f.iface AS iface, SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
|
||||
FROM ${table} f
|
||||
WHERE ${where.sql}
|
||||
GROUP BY f.server_id, f.iface
|
||||
`, where.params)
|
||||
await warmIfaceCache(ifaceRowsRaw.filter((r) => looksLikeIfIndex(r.iface)).map((r) => r.serverId))
|
||||
const ifaceRows = collapseServerIfaceRows(ifaceRowsRaw)
|
||||
const ifaceCount = ifaceRows.length
|
||||
|
||||
let userRows: Array<{ id: string; bytes: number; packets: number }> = []
|
||||
if (bindTuples.length && !ctx.unboundOnly) {
|
||||
@@ -522,6 +548,8 @@ export async function getStatisticsPivot(query: StatisticsPivotQuery): Promise<S
|
||||
if (pivotDimsConflict(query.row, query.col)) return emptyPivot(query)
|
||||
const period = parseStatisticsPeriod(query.from, query.to)
|
||||
if (!period) return emptyPivot(query)
|
||||
await warmBindingIfaceCache()
|
||||
if (query.serverId) await warmIfaceCache([query.serverId])
|
||||
const bindTuples = await loadBindUserTuples()
|
||||
const ctx = await buildFilterCtx(query, period)
|
||||
if (!ctx) return emptyPivot(query)
|
||||
@@ -543,6 +571,25 @@ export async function getStatisticsPivot(query: StatisticsPivotQuery): Promise<S
|
||||
GROUP BY 1, 2
|
||||
`, [...join.params, ...where.params])
|
||||
|
||||
if (query.row === "iface" || query.col === "iface") {
|
||||
const ifaceServerIds: number[] = []
|
||||
for (const r of raw) {
|
||||
for (const dim of [query.row, query.col] as const) {
|
||||
if (dim !== "iface") continue
|
||||
const id = dim === query.row ? String(r.row_id ?? "") : String(r.col_id ?? "")
|
||||
const colon = id.indexOf(":")
|
||||
if (colon < 0) continue
|
||||
const sid = Number(id.slice(0, colon))
|
||||
if (looksLikeIfIndex(id.slice(colon + 1)) && Number.isFinite(sid)) ifaceServerIds.push(sid)
|
||||
}
|
||||
}
|
||||
await warmIfaceCache(ifaceServerIds)
|
||||
for (const r of raw) {
|
||||
if (query.row === "iface") r.row_id = canonicalIfaceDimId(String(r.row_id ?? ""))
|
||||
if (query.col === "iface") r.col_id = canonicalIfaceDimId(String(r.col_id ?? ""))
|
||||
}
|
||||
}
|
||||
|
||||
const metric = query.metric
|
||||
type Acc = { bytes: number; packets: number }
|
||||
const cell = new Map<string, Map<string, Acc>>()
|
||||
@@ -684,7 +731,9 @@ async function loadPivotLabels(
|
||||
if (colon < 0) return id
|
||||
const sid = id.slice(0, colon)
|
||||
const iface = id.slice(colon + 1)
|
||||
return `${serverNames.get(sid) || sid} · ${iface}`
|
||||
const sidNum = Number(sid)
|
||||
const name = Number.isFinite(sidNum) ? displayFactIface(sidNum, iface) : iface
|
||||
return `${serverNames.get(sid) || sid} · ${name}`
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
@@ -22,8 +22,9 @@ rememberServerIfaces(7, [
|
||||
{ ".id": "*A", name: "wg-flow" },
|
||||
{ ".id": "*D", name: "bridge" },
|
||||
])
|
||||
assert.equal(resolveIfaceName(7, "2").name, "ether1")
|
||||
assert.equal(resolveIfaceName(7, "10").name, "wg-flow")
|
||||
assert.equal(resolveIfaceName(7, "2").name, "ether1")
|
||||
assert.equal(resolveIfaceName(7, "#2").name, "ether1")
|
||||
assert.equal(resolveIfaceName(7, "10").name, "wg-flow")
|
||||
assert.equal(resolveIfaceName(7, "13").name, "bridge")
|
||||
assert.equal(resolveIfaceName(7, "0").name, "—")
|
||||
assert.equal(resolveIfaceName(7, "ether1").name, "ether1")
|
||||
|
||||
@@ -3,7 +3,10 @@ import {
|
||||
bindingIfaceAliases,
|
||||
bindingIfaceAliasesAllServers,
|
||||
canonicalFactIface,
|
||||
collapseServerIfaceRows,
|
||||
displayFactIface,
|
||||
expandBindingIfaces,
|
||||
factIfaceAliases,
|
||||
rememberServerIfaces,
|
||||
resetIfaceCacheForTests,
|
||||
resolveIfaceName,
|
||||
@@ -18,12 +21,20 @@ assert.equal(canonicalFactIface(1, "2"), "gre-client")
|
||||
assert.equal(canonicalFactIface(1, "gre-client"), "gre-client")
|
||||
assert.equal(canonicalFactIface(1, "9"), "9")
|
||||
assert.equal(resolveIfaceName(1, "9").name, "#9")
|
||||
assert.equal(resolveIfaceName(1, "2").name, "gre-client")
|
||||
assert.equal(resolveIfaceName(1, "#2").name, "gre-client")
|
||||
assert.equal(displayFactIface(1, "2"), "gre-client")
|
||||
|
||||
const aliases = bindingIfaceAliases(1, "gre-client")
|
||||
assert.ok(aliases.includes("gre-client"))
|
||||
assert.ok(aliases.includes("2"))
|
||||
assert.ok(aliases.includes("#2"))
|
||||
|
||||
const fromIndex = factIfaceAliases("2", 1)
|
||||
assert.ok(fromIndex.includes("gre-client"))
|
||||
assert.ok(fromIndex.includes("2"))
|
||||
assert.ok(fromIndex.includes("#2"))
|
||||
|
||||
const all = bindingIfaceAliasesAllServers("gre-client")
|
||||
assert.ok(all.includes("2"))
|
||||
|
||||
@@ -31,5 +42,17 @@ const expanded = expandBindingIfaces([{ serverId: 1, iface: "gre-client" }])
|
||||
assert.ok(expanded.some((x) => x.iface === "2"))
|
||||
assert.ok(expanded.some((x) => x.iface === "gre-client"))
|
||||
|
||||
const collapsed = collapseServerIfaceRows([
|
||||
{ serverId: 1, iface: "2", bytes: 10, packets: 1 },
|
||||
{ serverId: 1, iface: "gre-client", bytes: 5, packets: 2 },
|
||||
{ serverId: 1, iface: "wan1", bytes: 3, packets: 1 },
|
||||
])
|
||||
assert.equal(collapsed.length, 2)
|
||||
const gre = collapsed.find((r) => r.iface === "gre-client")
|
||||
assert.ok(gre)
|
||||
assert.equal(gre.bytes, 15)
|
||||
assert.equal(gre.packets, 3)
|
||||
assert.ok(collapsed.some((r) => r.iface === "wan1"))
|
||||
|
||||
resetIfaceCacheForTests()
|
||||
console.log("traffic-flow-ifindex.test.ts: ok")
|
||||
|
||||
@@ -36,12 +36,12 @@ export function rememberServerIfaces(serverId: number, rows: RosIfaceIndexRow[])
|
||||
|
||||
export function resolveIfaceName(serverId: number, indexOrName: string): { name: string; index: string } {
|
||||
const trimmed = String(indexOrName ?? "").trim()
|
||||
if (!trimmed || trimmed === "0") return { name: "—", index: trimmed }
|
||||
if (!/^\d+$/.test(trimmed)) return { name: trimmed, index: "" }
|
||||
const idx = Number(trimmed)
|
||||
const name = cache.get(serverId)?.get(idx)
|
||||
if (name) return { name, index: trimmed }
|
||||
return { name: `#${trimmed}`, index: trimmed }
|
||||
const asIndex = trimmed.startsWith("#") && /^\d+$/.test(trimmed.slice(1)) ? trimmed.slice(1) : trimmed
|
||||
if (!asIndex || asIndex === "0") return { name: "—", index: asIndex }
|
||||
if (!/^\d+$/.test(asIndex)) return { name: trimmed, index: "" }
|
||||
const name = cache.get(serverId)?.get(Number(asIndex))
|
||||
if (name) return { name, index: asIndex }
|
||||
return { name: `#${asIndex}`, index: asIndex }
|
||||
}
|
||||
|
||||
/** Имя iface для факта куба: ifIndex→имя, без `#13` при пустом кэше. */
|
||||
@@ -53,18 +53,86 @@ export function canonicalFactIface(serverId: number, inIface: string): string {
|
||||
return name || trimmed
|
||||
}
|
||||
|
||||
function numericIfaceIndex(iface: string): string | null {
|
||||
const raw = String(iface ?? "").trim()
|
||||
if (/^\d+$/.test(raw)) return raw
|
||||
if (raw.startsWith("#") && /^\d+$/.test(raw.slice(1))) return raw.slice(1)
|
||||
return null
|
||||
}
|
||||
|
||||
/** Имя для UI: ifIndex → RouterOS name; `0` → «—»; miss → `#n`. */
|
||||
export function displayFactIface(serverId: number, iface: string): string {
|
||||
return resolveIfaceName(serverId, iface).name
|
||||
}
|
||||
|
||||
/** Склеить факты `2` + `ether1` в одну строку после резолва ifIndex. */
|
||||
export function collapseServerIfaceRows(
|
||||
rows: Array<{ serverId: number; iface: string; bytes: number; packets: number }>,
|
||||
): Array<{ serverId: number; iface: string; bytes: number; packets: number }> {
|
||||
const acc = new Map<string, { serverId: number; iface: string; bytes: number; packets: number }>()
|
||||
for (const r of rows) {
|
||||
const name = displayFactIface(r.serverId, r.iface)
|
||||
const k = `${r.serverId}\0${name}`
|
||||
const prev = acc.get(k)
|
||||
const bytes = Number(r.bytes) || 0
|
||||
const packets = Number(r.packets) || 0
|
||||
if (prev) {
|
||||
prev.bytes += bytes
|
||||
prev.packets += packets
|
||||
} else {
|
||||
acc.set(k, { serverId: r.serverId, iface: name, bytes, packets })
|
||||
}
|
||||
}
|
||||
return [...acc.values()]
|
||||
}
|
||||
|
||||
/** Ключи факта для фильтра: имя, ifIndex и `#n`. */
|
||||
export function factIfaceAliases(iface: string, serverId?: number): string[] {
|
||||
const raw = String(iface ?? "").trim()
|
||||
if (!raw) return []
|
||||
const out = new Set<string>([raw])
|
||||
const idx = numericIfaceIndex(raw)
|
||||
if (idx) {
|
||||
out.add(idx)
|
||||
out.add(`#${idx}`)
|
||||
const n = Number(idx)
|
||||
if (serverId != null) {
|
||||
const name = cache.get(serverId)?.get(n)
|
||||
if (name) out.add(name)
|
||||
} else {
|
||||
for (const map of cache.values()) {
|
||||
const name = map.get(n)
|
||||
if (name) out.add(name)
|
||||
}
|
||||
}
|
||||
}
|
||||
if (serverId != null) {
|
||||
for (const a of bindingIfaceAliases(serverId, raw)) out.add(a)
|
||||
} else {
|
||||
for (const a of bindingIfaceAliasesAllServers(raw)) out.add(a)
|
||||
}
|
||||
return [...out]
|
||||
}
|
||||
|
||||
/** Имя + ifIndex + `#n` — тот же матч, что карта `/traffic`. */
|
||||
export function bindingIfaceAliases(serverId: number, interfaceName: string): string[] {
|
||||
const name = String(interfaceName ?? "").trim()
|
||||
if (!name) return []
|
||||
const out = new Set<string>([name])
|
||||
const map = cache.get(serverId)
|
||||
if (!map) return [...out]
|
||||
for (const [idx, n] of map) {
|
||||
if (n !== name) continue
|
||||
out.add(String(idx))
|
||||
const idx = numericIfaceIndex(name)
|
||||
const canonical = (idx && map?.get(Number(idx))) || name
|
||||
out.add(canonical)
|
||||
if (idx) {
|
||||
out.add(idx)
|
||||
out.add(`#${idx}`)
|
||||
}
|
||||
if (!map) return [...out]
|
||||
for (const [i, n] of map) {
|
||||
if (n !== canonical && n !== name) continue
|
||||
out.add(String(i))
|
||||
out.add(`#${i}`)
|
||||
}
|
||||
return [...out]
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user