feat(statistics): enhance interface handling and data aggregation
Docker images / prepare-release (push) Successful in 9s
Docker images / backend-test (push) Successful in 2m29s
Docker images / frontend-image (push) Successful in 3m24s
Docker images / updater-image (push) Successful in 44s
Docker images / backend-image (push) Successful in 2m39s
Docker images / notify-webhook (push) Skipped
Docker images / publish-release (push) Successful in 15s
Docker images / prepare-release (push) Successful in 9s
Docker images / backend-test (push) Successful in 2m29s
Docker images / frontend-image (push) Successful in 3m24s
Docker images / updater-image (push) Successful in 44s
Docker images / backend-image (push) Successful in 2m39s
Docker images / notify-webhook (push) Skipped
Docker images / publish-release (push) Successful in 15s
Updated the statistics aggregation service to improve interface resolution and data handling. Introduced new functions for managing interface aliases and collapsing server interface rows, ensuring accurate data representation. Enhanced test coverage for interface resolution and added checks for new functionality. - Implemented `factIfaceAliases` and `collapseServerIfaceRows` for better interface data management. - Updated `resolveIfaceName` to handle additional cases for interface indexing. - Enhanced tests for interface resolution and aggregation logic. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -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