fix(statistics): не двоить overlay и транзит в кубе
Docker images / prepare-release (push) Successful in 8s
Docker images / backend-test (push) Successful in 2m18s
Docker images / frontend-image (push) Successful in 3m25s
Docker images / updater-image (push) Successful in 48s
Docker images / backend-image (push) Successful in 3m9s
Docker images / notify-webhook (push) Skipped
Docker images / publish-release (push) Successful in 14s
Docker images / prepare-release (push) Successful in 8s
Docker images / backend-test (push) Successful in 2m18s
Docker images / frontend-image (push) Successful in 3m25s
Docker images / updater-image (push) Successful in 48s
Docker images / backend-image (push) Successful in 3m9s
Docker images / notify-webhook (push) Skipped
Docker images / publish-release (push) Successful in 14s
Co-authored-by: Cursor <[email protected]>
This commit is contained in:
@@ -393,6 +393,15 @@ export default function StatisticsPage() {
|
||||
</Alert>
|
||||
) : null}
|
||||
|
||||
{!slices.serverId && isLive && !emptyCube ? (
|
||||
<Alert>
|
||||
<AlertTitle>Интернет сети</AlertTitle>
|
||||
<AlertDescription>
|
||||
KPI — уникальный payload без overlay и транзита EN. Сумма WAN по серверам не равна интернету сети; для uplink откройте сервер.
|
||||
</AlertDescription>
|
||||
</Alert>
|
||||
) : null}
|
||||
|
||||
<KpiStatGrid
|
||||
aria-label="Сводка трафика"
|
||||
isLoading={loading}
|
||||
@@ -432,7 +441,9 @@ export default function StatisticsPage() {
|
||||
id: "servers",
|
||||
label: "Серверы",
|
||||
value: String(kpis.servers),
|
||||
hint: kpis.ifaces ? `${kpis.ifaces} iface` : undefined,
|
||||
hint: slices.serverId
|
||||
? (kpis.ifaces ? `${kpis.ifaces} iface` : undefined)
|
||||
: "WAN — в слайсе сервера",
|
||||
icon: <ServerIcon />,
|
||||
iconClassName: "text-muted-foreground",
|
||||
},
|
||||
|
||||
@@ -15,7 +15,7 @@
|
||||
"test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts",
|
||||
"test:wireguard": "npx tsx src/services/wireguard-config.test.ts",
|
||||
"test:traffic-rate": "tsx src/services/traffic-rate.test.ts",
|
||||
"test:traffic-flow": "tsx src/services/traffic-flow-parse.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-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/statistics-aggregate.test.ts",
|
||||
"test:traffic-flow": "tsx src/services/traffic-flow-parse.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-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/statistics-aggregate.test.ts",
|
||||
"test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts",
|
||||
"test:pg": "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",
|
||||
"test:backups": "tsx src/services/s3-backup-client.test.ts",
|
||||
|
||||
@@ -2,6 +2,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 { invalidateFlowCatalogCache } from "./traffic-flow-topology.js"
|
||||
import { withPgOrSkip } from "../test/pg.js"
|
||||
import { dbQuery } from "../db/index.js"
|
||||
import { ensurePartitionFor } from "../db/partitions.js"
|
||||
@@ -29,15 +30,25 @@ if (!(await withPgOrSkip())) {
|
||||
}
|
||||
|
||||
const inserted = await dbQuery<{ id: number }>(`
|
||||
INSERT INTO servers (name, host) VALUES ('stats-cube', '127.0.0.1') RETURNING id
|
||||
INSERT INTO servers (name, host, type, wan_uplinks)
|
||||
VALUES ('stats-cube', '127.0.0.1', 'jump-host', '[{"iface":"wan1"}]'::jsonb)
|
||||
RETURNING id
|
||||
`)
|
||||
const serverId = inserted.rows[0]?.id
|
||||
if (serverId == null) throw new Error("no server")
|
||||
|
||||
const enInserted = await dbQuery<{ id: number }>(`
|
||||
INSERT INTO servers (name, host, type)
|
||||
VALUES ('stats-en', '198.51.100.1', 'exit-node')
|
||||
RETURNING id
|
||||
`)
|
||||
const enId = enInserted.rows[0]?.id
|
||||
if (enId == null) throw new Error("no en server")
|
||||
|
||||
await ensurePartitionFor(pool, "flow_daily_facts", "month", new Date("2026-09-01T00:00:00Z"))
|
||||
await ensurePartitionFor(pool, "flow_hour_facts", "day", new Date("2026-09-10T00:00:00Z"))
|
||||
await dbQuery(`DELETE FROM flow_daily_facts WHERE server_id = $1`, [serverId])
|
||||
await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id = $1`, [serverId])
|
||||
await dbQuery(`DELETE FROM flow_daily_facts WHERE server_id IN ($1, $2)`, [serverId, enId])
|
||||
await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id IN ($1, $2)`, [serverId, enId])
|
||||
await dbQuery(`DELETE FROM user_interface_bindings WHERE server_id = $1`, [serverId])
|
||||
await dbQuery(`DELETE FROM app_users WHERE id = 'u-stats-1'`)
|
||||
|
||||
@@ -52,16 +63,29 @@ await dbQuery(`
|
||||
`, [serverId])
|
||||
|
||||
resetIfaceCacheForTests()
|
||||
rememberServerIfaces(serverId, [{ name: "gre-client", ifindex: "2" }])
|
||||
rememberServerIfaces(serverId, [
|
||||
{ name: "gre-client", ifindex: "2" },
|
||||
{ name: "wan1", ifindex: "8" },
|
||||
{ name: "gre-en", ifindex: "9" },
|
||||
])
|
||||
rememberServerIfaces(enId, [
|
||||
{ name: "ether1", ifindex: "2" },
|
||||
{ name: "gre-jh", ifindex: "5" },
|
||||
])
|
||||
setRefreshIfacesForTests(async () => {})
|
||||
invalidateFlowCatalogCache()
|
||||
|
||||
await dbQuery(`
|
||||
INSERT INTO flow_daily_facts (server_id, day, iface, country, service, asn, bytes, packets)
|
||||
VALUES
|
||||
($1, '2026-09-10', '2', 'US', 'https', 15169, 800, 10),
|
||||
($1, '2026-09-10', '2', 'DE', 'dns', 15133, 200, 4),
|
||||
($1, '2026-09-10', 'wan1', 'NL', 'other', 0, 70, 1)
|
||||
`, [serverId])
|
||||
($1, '2026-09-10', 'wan1', 'NL', 'other', 0, 70, 1),
|
||||
($1, '2026-09-10', '0', 'US', 'https', 0, 999, 3),
|
||||
($1, '2026-09-10', 'gre-en', 'US', 'https', 15169, 400, 2),
|
||||
($2, '2026-09-10', 'gre-jh', 'US', 'https', 15169, 500, 5),
|
||||
($2, '2026-09-10', 'ether1', 'US', 'https', 15169, 200, 2)
|
||||
`, [serverId, enId])
|
||||
|
||||
try {
|
||||
const all = await getStatistics({ from: "2026-09-01", to: "2026-09-30" })
|
||||
@@ -71,14 +95,41 @@ try {
|
||||
assert.ok(all.countries.some((r) => r.id === "US"))
|
||||
assert.ok(all.users.some((r) => r.id === "u-stats-1"))
|
||||
const unbound = all.users.find((r) => r.id === STATISTICS_UNBOUND_USER_ID)
|
||||
assert.ok(unbound)
|
||||
assert.equal(unbound.bytes, 70)
|
||||
assert.equal(unbound, undefined, "WAN не в Прочие / Без привязки")
|
||||
assert.ok(all.servers.some((r) => r.id === String(serverId)))
|
||||
assert.ok(!all.servers.some((r) => r.id === String(enId)), "EN-транзит не в сетевом KPI")
|
||||
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)))
|
||||
assert.ok(!all.interfaces.some((r) => r.label.includes(" · —") || r.label.endsWith("· —")))
|
||||
assert.ok(!all.interfaces.some((r) => r.label.includes("gre-en")))
|
||||
const wanRow = all.interfaces.find((r) => r.id === `${serverId}:wan1`)
|
||||
assert.ok(wanRow)
|
||||
assert.ok(wanRow.label.includes("WAN · интернет"))
|
||||
assert.equal(wanRow.bytes, 70)
|
||||
|
||||
const nodeSlice = await getStatistics({
|
||||
from: "2026-09-01",
|
||||
to: "2026-09-30",
|
||||
serverId,
|
||||
})
|
||||
assert.equal(nodeSlice.kpis.bytes, 1070)
|
||||
const nodeWan = nodeSlice.interfaces.find((r) => r.id === `${serverId}:wan1`)
|
||||
assert.ok(nodeWan)
|
||||
assert.ok(nodeWan.label.includes("WAN · интернет"))
|
||||
assert.ok(!nodeSlice.interfaces.some((r) => r.label.includes(" · —")))
|
||||
assert.ok(!nodeSlice.users.some((r) => r.id === STATISTICS_UNBOUND_USER_ID))
|
||||
|
||||
const enSlice = await getStatistics({
|
||||
from: "2026-09-01",
|
||||
to: "2026-09-30",
|
||||
serverId: enId,
|
||||
})
|
||||
assert.equal(enSlice.kpis.bytes, 200)
|
||||
assert.ok(enSlice.interfaces.some((r) => r.label.includes("WAN · интернет") && r.label.includes("ether1")))
|
||||
assert.ok(!enSlice.interfaces.some((r) => r.label.includes("gre-jh")))
|
||||
|
||||
const sliced = await getStatistics({
|
||||
from: "2026-09-01",
|
||||
@@ -127,9 +178,10 @@ try {
|
||||
} 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])
|
||||
await dbQuery(`DELETE FROM servers WHERE id = $1`, [serverId])
|
||||
invalidateFlowCatalogCache()
|
||||
await dbQuery(`DELETE FROM flow_daily_facts WHERE server_id IN ($1, $2)`, [serverId, enId])
|
||||
await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id IN ($1, $2)`, [serverId, enId])
|
||||
await dbQuery(`DELETE FROM servers WHERE id IN ($1, $2)`, [serverId, enId])
|
||||
}
|
||||
|
||||
console.log("statistics-aggregate.test.ts: ok")
|
||||
|
||||
@@ -15,8 +15,17 @@ import {
|
||||
displayFactIface,
|
||||
expandBindingIfaces,
|
||||
factIfaceAliases,
|
||||
listCachedIfaceNames,
|
||||
} from "./traffic-flow-ifindex.js"
|
||||
import { refreshServerIfaces } from "./traffic-flow-ifaces.js"
|
||||
import {
|
||||
isDashDisplayIface,
|
||||
isJunkFactIface,
|
||||
isOverlayGreIface,
|
||||
isWanFactIface,
|
||||
wanIfaceLabel,
|
||||
} from "./traffic-flow-facts-filter.js"
|
||||
import { getServerCatalog, loadFlowTopology, type FlowTopology } from "./traffic-flow-topology.js"
|
||||
|
||||
const TOP_N = 200
|
||||
const HOUR_WINDOW_MS = 48 * 3600_000
|
||||
@@ -91,6 +100,10 @@ interface FilterCtx {
|
||||
userIfaces: Array<{ serverId: number; iface: string }> | null
|
||||
unboundOnly: boolean
|
||||
boundIfaces: Array<{ serverId: number; iface: string }>
|
||||
overlayIfaces: Array<{ serverId: number; iface: string }>
|
||||
wanIfaces: Array<{ serverId: number; iface: string }>
|
||||
excludeServerIds: number[]
|
||||
topo: FlowTopology | null
|
||||
}
|
||||
|
||||
function ifaceFilterAliases(iface: string, serverId?: number): string[] {
|
||||
@@ -169,16 +182,31 @@ function factWhere(alias: string, grain: "hour" | "day", ctx: FilterCtx): { sql:
|
||||
}
|
||||
}
|
||||
if (ctx.unboundOnly) {
|
||||
if (ctx.boundIfaces.length === 0) {
|
||||
/* весь трафик без привязок */
|
||||
} else {
|
||||
const tuples = ctx.boundIfaces.map(() => "(?, ?)").join(", ")
|
||||
const skip = [...ctx.boundIfaces, ...ctx.wanIfaces]
|
||||
if (skip.length) {
|
||||
const tuples = skip.map(() => "(?, ?)").join(", ")
|
||||
parts.push(`(${alias}.server_id, ${alias}.iface) NOT IN (${tuples})`)
|
||||
for (const u of ctx.boundIfaces) {
|
||||
params.push(u.serverId, u.iface)
|
||||
}
|
||||
for (const u of skip) params.push(u.serverId, u.iface)
|
||||
}
|
||||
}
|
||||
parts.push(`${alias}.iface NOT IN ('0', '—', '__unknown__', 'wg-flow', '')`)
|
||||
const greLike = `(LOWER(${alias}.iface) LIKE 'gre%' OR LOWER(${alias}.iface) LIKE '%gre-tunnel%')`
|
||||
if (ctx.boundIfaces.length) {
|
||||
const tuples = ctx.boundIfaces.map(() => "(?, ?)").join(", ")
|
||||
parts.push(`(NOT ${greLike} OR (${alias}.server_id, ${alias}.iface) IN (${tuples}))`)
|
||||
for (const u of ctx.boundIfaces) params.push(u.serverId, u.iface)
|
||||
} else {
|
||||
parts.push(`NOT ${greLike}`)
|
||||
}
|
||||
if (ctx.overlayIfaces.length) {
|
||||
const tuples = ctx.overlayIfaces.map(() => "(?, ?)").join(", ")
|
||||
parts.push(`(${alias}.server_id, ${alias}.iface) NOT IN (${tuples})`)
|
||||
for (const u of ctx.overlayIfaces) params.push(u.serverId, u.iface)
|
||||
}
|
||||
if (ctx.excludeServerIds.length) {
|
||||
parts.push(`${alias}.server_id NOT IN (${ctx.excludeServerIds.map(() => "?").join(", ")})`)
|
||||
params.push(...ctx.excludeServerIds)
|
||||
}
|
||||
return { sql: parts.join(" AND "), params }
|
||||
}
|
||||
|
||||
@@ -274,12 +302,61 @@ function userBindJoinSql(tuples: UserBindTuple[]): { sql: string; params: unknow
|
||||
}
|
||||
}
|
||||
|
||||
function expandIfaceTuples(
|
||||
items: Array<{ serverId: number; iface: string }>,
|
||||
): Array<{ serverId: number; iface: string }> {
|
||||
const seen = new Set<string>()
|
||||
const out: Array<{ serverId: number; iface: string }> = []
|
||||
for (const t of items) {
|
||||
for (const iface of factIfaceAliases(t.iface, t.serverId)) {
|
||||
const k = `${t.serverId}\0${iface}`
|
||||
if (seen.has(k)) continue
|
||||
seen.add(k)
|
||||
out.push({ serverId: t.serverId, iface })
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
async function loadPayloadScope(serverId?: number): Promise<{
|
||||
overlayIfaces: Array<{ serverId: number; iface: string }>
|
||||
wanIfaces: Array<{ serverId: number; iface: string }>
|
||||
excludeServerIds: number[]
|
||||
topo: FlowTopology
|
||||
}> {
|
||||
const topo = await loadFlowTopology()
|
||||
const catalog = await getServerCatalog()
|
||||
await warmIfaceCache(catalog.list.map((s) => s.id))
|
||||
const overlayRaw: Array<{ serverId: number; iface: string }> = []
|
||||
const wanRaw: Array<{ serverId: number; iface: string }> = []
|
||||
for (const s of catalog.list) {
|
||||
if (serverId != null && s.id !== serverId) continue
|
||||
const wanSet = topo.wanIfaces.get(s.id)
|
||||
const wanNames = wanSet && wanSet.size > 0
|
||||
? [...wanSet]
|
||||
: s.type === "home-router" ? [] : ["ether1"]
|
||||
for (const name of wanNames) wanRaw.push({ serverId: s.id, iface: name })
|
||||
for (const name of listCachedIfaceNames(s.id)) {
|
||||
if (isOverlayGreIface(topo, s.id, name)) overlayRaw.push({ serverId: s.id, iface: name })
|
||||
}
|
||||
}
|
||||
return {
|
||||
overlayIfaces: expandIfaceTuples(overlayRaw),
|
||||
wanIfaces: expandIfaceTuples(wanRaw),
|
||||
excludeServerIds: serverId != null
|
||||
? []
|
||||
: catalog.list.filter((s) => s.type === "exit-node").map((s) => s.id),
|
||||
topo,
|
||||
}
|
||||
}
|
||||
|
||||
async function buildFilterCtx(query: StatisticsQuery, period: ParsedPeriod): Promise<FilterCtx | null> {
|
||||
const bindTuples = await loadBindUserTuples()
|
||||
const boundIfaces = uniqueBoundIfaces(bindTuples)
|
||||
const unboundOnly = query.userId === STATISTICS_UNBOUND_USER_ID
|
||||
const userIfaces = unboundOnly ? null : await resolveUserIfaces(query.userId)
|
||||
if (userIfaces && userIfaces.length === 0) return null
|
||||
const scope = await loadPayloadScope(query.serverId)
|
||||
return {
|
||||
...period,
|
||||
serverId: query.serverId,
|
||||
@@ -290,6 +367,10 @@ async function buildFilterCtx(query: StatisticsQuery, period: ParsedPeriod): Pro
|
||||
userIfaces,
|
||||
unboundOnly,
|
||||
boundIfaces,
|
||||
overlayIfaces: scope.overlayIfaces,
|
||||
wanIfaces: scope.wanIfaces,
|
||||
excludeServerIds: scope.excludeServerIds,
|
||||
topo: scope.topo,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -370,7 +451,11 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
|
||||
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 ifaceRows = collapseServerIfaceRows(ifaceRowsRaw).filter((r) => {
|
||||
if (isJunkFactIface(r.iface) || isDashDisplayIface(r.iface)) return false
|
||||
if (ctx.topo && isOverlayGreIface(ctx.topo, r.serverId, r.iface)) return false
|
||||
return true
|
||||
})
|
||||
const ifaceCount = ifaceRows.length
|
||||
|
||||
let userRows: Array<{ id: string; bytes: number; packets: number }> = []
|
||||
@@ -442,12 +527,16 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
|
||||
period.windowSec,
|
||||
)
|
||||
const interfaces = toBreakdown(
|
||||
ifaceRows.map((r) => ({
|
||||
id: `${r.serverId}:${r.iface}`,
|
||||
label: `${serverNames.get(r.serverId) || r.serverId} · ${r.iface}`,
|
||||
bytes: Number(r.bytes) || 0,
|
||||
packets: Number(r.packets) || 0,
|
||||
})),
|
||||
ifaceRows.map((r) => {
|
||||
const serverName = serverNames.get(r.serverId) || String(r.serverId)
|
||||
const wan = ctx.topo ? isWanFactIface(ctx.topo, r.serverId, r.iface) : false
|
||||
return {
|
||||
id: `${r.serverId}:${r.iface}`,
|
||||
label: wan ? wanIfaceLabel(serverName, r.iface) : `${serverName} · ${r.iface}`,
|
||||
bytes: Number(r.bytes) || 0,
|
||||
packets: Number(r.packets) || 0,
|
||||
}
|
||||
}),
|
||||
bytes,
|
||||
period.windowSec,
|
||||
)
|
||||
@@ -466,13 +555,14 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
|
||||
if (!ctx.unboundOnly && !ctx.userIfaces) {
|
||||
let unboundBytes = 0
|
||||
let unboundPackets = 0
|
||||
if (ctx.boundIfaces.length === 0) {
|
||||
const skipUnbound = [...ctx.boundIfaces, ...ctx.wanIfaces]
|
||||
if (skipUnbound.length === 0) {
|
||||
unboundBytes = bytes
|
||||
unboundPackets = packets
|
||||
} else {
|
||||
const tuples = ctx.boundIfaces.map(() => "(?, ?)").join(", ")
|
||||
const tuples = skipUnbound.map(() => "(?, ?)").join(", ")
|
||||
const unboundParams = [...where.params]
|
||||
for (const u of ctx.boundIfaces) unboundParams.push(u.serverId, u.iface)
|
||||
for (const u of skipUnbound) unboundParams.push(u.serverId, u.iface)
|
||||
const unboundRows = await dbAll<{ bytes: number; packets: number }>(`
|
||||
SELECT COALESCE(SUM(f.bytes), 0) AS bytes, COALESCE(SUM(f.packets), 0) AS packets
|
||||
FROM ${table} f
|
||||
@@ -486,7 +576,7 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
|
||||
const denom = bytes || 1
|
||||
users.push({
|
||||
id: STATISTICS_UNBOUND_USER_ID,
|
||||
label: "Без привязки",
|
||||
label: "Прочие",
|
||||
bytes: unboundBytes,
|
||||
packets: unboundPackets,
|
||||
bps: (unboundBytes * 8) / period.windowSec,
|
||||
@@ -706,6 +796,7 @@ async function loadPivotLabels(
|
||||
rowIds: string[],
|
||||
colIds: string[],
|
||||
): Promise<{ row: Map<string, string>; col: Map<string, string> }> {
|
||||
const topo = await loadFlowTopology()
|
||||
const serverNames = new Map<string, string>()
|
||||
const allServers = await db.select({ id: servers.id, name: servers.name, host: servers.host }).from(servers)
|
||||
for (const s of allServers) serverNames.set(String(s.id), s.name || s.host)
|
||||
@@ -733,7 +824,11 @@ async function loadPivotLabels(
|
||||
const iface = id.slice(colon + 1)
|
||||
const sidNum = Number(sid)
|
||||
const name = Number.isFinite(sidNum) ? displayFactIface(sidNum, iface) : iface
|
||||
return `${serverNames.get(sid) || sid} · ${name}`
|
||||
const serverName = serverNames.get(sid) || sid
|
||||
if (Number.isFinite(sidNum) && isWanFactIface(topo, sidNum, name)) {
|
||||
return wanIfaceLabel(serverName, name)
|
||||
}
|
||||
return `${serverName} · ${name}`
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
@@ -11,7 +11,14 @@ import { invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js"
|
||||
import { isIsoCountry } from "./traffic-flow-brands.js"
|
||||
import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js"
|
||||
import { canonicalFactIface } from "./traffic-flow-ifindex.js"
|
||||
import { shouldWriteFlowFact } from "./traffic-flow-facts-filter.js"
|
||||
import { pickInternetPeer } from "./traffic-flow-ip.js"
|
||||
import {
|
||||
getServerCatalog,
|
||||
loadFlowTopology,
|
||||
peekFlowTopology,
|
||||
peekServerCatalog,
|
||||
} from "./traffic-flow-topology.js"
|
||||
import {
|
||||
bumpFlowFact,
|
||||
factsPendingSize,
|
||||
@@ -338,6 +345,11 @@ export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): vo
|
||||
const bucketAt = minuteBucketIso()
|
||||
const hourAt = hourBucketIso()
|
||||
const ripeMisses: string[] = []
|
||||
const topo = peekFlowTopology()
|
||||
const catalog = peekServerCatalog()
|
||||
if (!topo) void loadFlowTopology().catch(() => {})
|
||||
if (!catalog) void getServerCatalog().catch(() => {})
|
||||
const serverType = catalog?.byId.get(serverId)?.type
|
||||
for (const raw of flows) {
|
||||
const flow = normalizeParsedFlow(raw)
|
||||
addToTick(serverId, flow, flow.bytes)
|
||||
@@ -358,16 +370,29 @@ export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): vo
|
||||
bumpDim(serverId, bucketAt, "service", classified.service, flow.bytes, flow.packets)
|
||||
if (country) bumpDim(serverId, bucketAt, "country", country, flow.bytes, flow.packets)
|
||||
bumpDim(serverId, bucketAt, "asn", asnKey, flow.bytes, flow.packets)
|
||||
bumpFlowFact({
|
||||
if (shouldWriteFlowFact({
|
||||
serverId,
|
||||
bucketAt: hourAt,
|
||||
iface: canonicalFactIface(serverId, flow.inIface),
|
||||
country: country || "XX",
|
||||
service: classified.service,
|
||||
asn: ripe?.ok && ripe.asn ? ripe.asn : 0,
|
||||
bytes: flow.bytes,
|
||||
packets: flow.packets,
|
||||
})
|
||||
serverType,
|
||||
inIface: flow.inIface,
|
||||
outIface: flow.outIface,
|
||||
proto: flow.proto,
|
||||
srcPort: flow.srcPort,
|
||||
dstPort: flow.dstPort,
|
||||
src: flow.src,
|
||||
dst: flow.dst,
|
||||
topo,
|
||||
})) {
|
||||
bumpFlowFact({
|
||||
serverId,
|
||||
bucketAt: hourAt,
|
||||
iface: canonicalFactIface(serverId, flow.inIface),
|
||||
country: country || "XX",
|
||||
service: classified.service,
|
||||
asn: ripe?.ok && ripe.asn ? ripe.asn : 0,
|
||||
bytes: flow.bytes,
|
||||
packets: flow.packets,
|
||||
})
|
||||
}
|
||||
|
||||
const key = pendingKey(serverId, bucketAt, flow)
|
||||
const prev = pending.get(key)
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
import assert from "node:assert/strict"
|
||||
import {
|
||||
isJunkFactIface,
|
||||
isOverlayGreIface,
|
||||
isWanFactIface,
|
||||
shouldWriteFlowFact,
|
||||
} from "./traffic-flow-facts-filter.js"
|
||||
import { seedFlowTopologyForTests, type FlowTopology } from "./traffic-flow-topology.js"
|
||||
import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js"
|
||||
|
||||
function topo(partial: Partial<FlowTopology> = {}): FlowTopology {
|
||||
const wanIfaces = partial.wanIfaces ?? new Map([[1, new Set(["ether1"])]])
|
||||
const clientIfaces = partial.clientIfaces ?? new Map([[1, new Set(["gre-client"])]])
|
||||
const clientByIface = partial.clientByIface ?? new Map()
|
||||
const enHosts = partial.enHosts ?? new Set(["198.51.100.1"])
|
||||
const jhHosts = partial.jhHosts ?? new Set(["203.0.113.10"])
|
||||
return {
|
||||
clientIfaces,
|
||||
clientByIface,
|
||||
enNodes: partial.enNodes ?? [{ id: 2, name: "en", hosts: ["198.51.100.1"] }],
|
||||
enHosts,
|
||||
jhHosts,
|
||||
wanIfaces,
|
||||
plane: partial.plane ?? {
|
||||
clientIfaceNames: new Set(["gre-client"]),
|
||||
enHosts,
|
||||
jhHosts,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
resetIfaceCacheForTests()
|
||||
rememberServerIfaces(1, [{ name: "ether1", ifindex: "2" }])
|
||||
seedFlowTopologyForTests(topo())
|
||||
|
||||
assert.equal(isJunkFactIface("0"), true)
|
||||
assert.equal(isJunkFactIface(""), true)
|
||||
assert.equal(isJunkFactIface("ether1"), false)
|
||||
assert.equal(isWanFactIface(topo(), 1, "ether1"), true)
|
||||
assert.equal(isOverlayGreIface(topo(), 1, "gre-en"), true)
|
||||
assert.equal(isOverlayGreIface(topo(), 1, "gre-client"), false)
|
||||
assert.equal(isOverlayGreIface(topo(), 1, "ether1"), false)
|
||||
|
||||
const overlayOuter = shouldWriteFlowFact({
|
||||
serverId: 1,
|
||||
serverType: "jump-host",
|
||||
inIface: "ether1",
|
||||
outIface: "gre-en",
|
||||
proto: 47,
|
||||
srcPort: 0,
|
||||
dstPort: 0,
|
||||
src: "203.0.113.10",
|
||||
dst: "198.51.100.1",
|
||||
topo: topo(),
|
||||
})
|
||||
assert.equal(overlayOuter, false, "overlay proto 47 на ether1 не в facts")
|
||||
|
||||
const payloadGre = shouldWriteFlowFact({
|
||||
serverId: 1,
|
||||
serverType: "jump-host",
|
||||
inIface: "gre-client",
|
||||
outIface: "gre-en",
|
||||
proto: 6,
|
||||
srcPort: 51234,
|
||||
dstPort: 443,
|
||||
src: "10.100.1.17",
|
||||
dst: "8.8.8.8",
|
||||
topo: topo(),
|
||||
})
|
||||
assert.equal(payloadGre, true, "payload на GRE — да")
|
||||
|
||||
const payloadWan = shouldWriteFlowFact({
|
||||
serverId: 1,
|
||||
serverType: "jump-host",
|
||||
inIface: "ether1",
|
||||
outIface: "gre-client",
|
||||
proto: 6,
|
||||
srcPort: 443,
|
||||
dstPort: 51234,
|
||||
src: "8.8.8.8",
|
||||
dst: "10.100.1.17",
|
||||
topo: topo(),
|
||||
})
|
||||
assert.equal(payloadWan, true, "payload на ether1 WAN — да")
|
||||
|
||||
const junkZero = shouldWriteFlowFact({
|
||||
serverId: 1,
|
||||
serverType: "jump-host",
|
||||
inIface: "0",
|
||||
proto: 6,
|
||||
srcPort: 443,
|
||||
dstPort: 80,
|
||||
src: "1.1.1.1",
|
||||
dst: "8.8.8.8",
|
||||
topo: topo(),
|
||||
})
|
||||
assert.equal(junkZero, false)
|
||||
|
||||
const enTopo = topo({
|
||||
clientIfaces: new Map([[2, new Set()]]),
|
||||
wanIfaces: new Map([[2, new Set(["ether1"])]]),
|
||||
})
|
||||
const enTransit = shouldWriteFlowFact({
|
||||
serverId: 2,
|
||||
serverType: "exit-node",
|
||||
inIface: "gre-jh",
|
||||
outIface: "ether1",
|
||||
proto: 6,
|
||||
srcPort: 51234,
|
||||
dstPort: 443,
|
||||
src: "10.100.1.17",
|
||||
dst: "8.8.8.8",
|
||||
topo: enTopo,
|
||||
})
|
||||
assert.equal(enTransit, false, "EN-транзит без клиента — нет")
|
||||
|
||||
const enWan = shouldWriteFlowFact({
|
||||
serverId: 2,
|
||||
serverType: "exit-node",
|
||||
inIface: "ether1",
|
||||
proto: 6,
|
||||
srcPort: 443,
|
||||
dstPort: 80,
|
||||
src: "8.8.8.8",
|
||||
dst: "198.51.100.1",
|
||||
topo: enTopo,
|
||||
})
|
||||
assert.equal(enWan, true, "WAN payload на EN — да")
|
||||
|
||||
seedFlowTopologyForTests(null)
|
||||
resetIfaceCacheForTests()
|
||||
console.log("traffic-flow-facts-filter.test.ts: ok")
|
||||
@@ -0,0 +1,85 @@
|
||||
import { mapRosInterfaceType } from "../modules/users/iface-type.js"
|
||||
import { canonicalFactIface } from "./traffic-flow-ifindex.js"
|
||||
import { classifyFlowPlane } from "./traffic-flow-planes.js"
|
||||
import { resolveClient, type FlowTopology } from "./traffic-flow-topology.js"
|
||||
|
||||
const JUNK_IFACE = new Set(["", "0", "—", "__unknown__", "wg-flow"])
|
||||
|
||||
export function isJunkFactIface(iface: string | null | undefined): boolean {
|
||||
const n = String(iface ?? "").trim()
|
||||
if (JUNK_IFACE.has(n)) return true
|
||||
return /^#?0$/.test(n)
|
||||
}
|
||||
|
||||
export function isDashDisplayIface(iface: string): boolean {
|
||||
return String(iface ?? "").trim() === "—"
|
||||
}
|
||||
|
||||
export function isGreIfaceName(name: string): boolean {
|
||||
return mapRosInterfaceType("", name) === "gre"
|
||||
}
|
||||
|
||||
/** WAN uplink: `wanIfaces` топологии, иначе ether1 у JH/EN без wan_uplinks. */
|
||||
export function isWanFactIface(
|
||||
topo: FlowTopology | null | undefined,
|
||||
serverId: number,
|
||||
iface: string,
|
||||
): boolean {
|
||||
const name = String(iface ?? "").trim()
|
||||
if (!name || isJunkFactIface(name) || isDashDisplayIface(name)) return false
|
||||
const wan = topo?.wanIfaces.get(serverId)
|
||||
if (wan && wan.size > 0) return wan.has(name)
|
||||
return /^ether1$/i.test(name)
|
||||
}
|
||||
|
||||
/** GRE между своими серверами / транзит EN: gre-имя, не клиент, не WAN. */
|
||||
export function isOverlayGreIface(
|
||||
topo: FlowTopology | null | undefined,
|
||||
serverId: number,
|
||||
iface: string,
|
||||
): boolean {
|
||||
const name = String(iface ?? "").trim()
|
||||
if (!name || isWanFactIface(topo, serverId, name)) return false
|
||||
if (topo?.clientIfaces.get(serverId)?.has(name)) return false
|
||||
if (name === "wg-flow") return false
|
||||
return isGreIfaceName(name)
|
||||
}
|
||||
|
||||
export function shouldWriteFlowFact(opts: {
|
||||
serverId: number
|
||||
serverType?: string
|
||||
inIface: string
|
||||
outIface?: string
|
||||
proto: number
|
||||
srcPort: number
|
||||
dstPort: number
|
||||
src: string
|
||||
dst: string
|
||||
topo?: FlowTopology | null
|
||||
}): boolean {
|
||||
const inName = canonicalFactIface(opts.serverId, opts.inIface) || String(opts.inIface ?? "").trim()
|
||||
if (isJunkFactIface(inName) || isJunkFactIface(opts.inIface)) return false
|
||||
const outRaw = String(opts.outIface ?? "").trim()
|
||||
const outName = outRaw ? (canonicalFactIface(opts.serverId, outRaw) || outRaw) : ""
|
||||
const plane = classifyFlowPlane({
|
||||
src: opts.src,
|
||||
dst: opts.dst,
|
||||
proto: opts.proto,
|
||||
srcPort: opts.srcPort,
|
||||
dstPort: opts.dstPort,
|
||||
inIface: inName,
|
||||
outIface: outName || undefined,
|
||||
}, opts.topo?.plane)
|
||||
if (plane !== "payload") return false
|
||||
if (opts.serverType === "exit-node" && opts.topo) {
|
||||
const client =
|
||||
resolveClient(opts.topo, opts.serverId, inName)
|
||||
?? (outName ? resolveClient(opts.topo, opts.serverId, outName) : null)
|
||||
if (!client && isOverlayGreIface(opts.topo, opts.serverId, inName)) return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
export function wanIfaceLabel(serverName: string, iface: string): string {
|
||||
return `${serverName} · ${iface} · WAN · интернет`
|
||||
}
|
||||
@@ -161,6 +161,12 @@ export function expandBindingIfaces(
|
||||
return out
|
||||
}
|
||||
|
||||
export function listCachedIfaceNames(serverId: number): string[] {
|
||||
const map = cache.get(serverId)
|
||||
if (!map) return []
|
||||
return [...new Set(map.values())]
|
||||
}
|
||||
|
||||
export function ifaceCacheHas(serverId: number): boolean {
|
||||
return cache.has(serverId)
|
||||
}
|
||||
|
||||
@@ -48,6 +48,15 @@ export function invalidateFlowCatalogCache(): void {
|
||||
serverCatalogCache = null
|
||||
}
|
||||
|
||||
export function peekFlowTopology(): FlowTopology | null {
|
||||
if (seeded) return seeded
|
||||
return topologyCache?.topo ?? null
|
||||
}
|
||||
|
||||
export function peekServerCatalog(): { list: ServerCatalogEntry[]; byId: Map<number, ServerCatalogEntry> } | null {
|
||||
return serverCatalogCache
|
||||
}
|
||||
|
||||
export async function getServerCatalog(): Promise<{ list: ServerCatalogEntry[]; byId: Map<number, ServerCatalogEntry> }> {
|
||||
const now = Date.now()
|
||||
if (serverCatalogCache && now - serverCatalogCache.at < CATALOG_TTL_MS) {
|
||||
|
||||
Reference in New Issue
Block a user