DenozordecandCursor
39c8ec4a02
Docker images / prepare-release (push) Successful in 11s
Docker images / backend-test (push) Successful in 2m9s
Docker images / frontend-image (push) Successful in 3m20s
Docker images / updater-image (push) Successful in 49s
Docker images / backend-image (push) Successful in 2m44s
Docker images / notify-webhook (push) Skipped
Docker images / publish-release (push) Successful in 10s
Коллектор снова отдаёт назначения в живом потоке. Ошибки записи видны в статусе, лишние перезаписи минутных агрегатов убраны. Co-authored-by: Cursor <[email protected]>
499 lines
14 KiB
TypeScript
499 lines
14 KiB
TypeScript
import assert from "node:assert/strict"
|
|
import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js"
|
|
import {
|
|
ingestParsedFlowsForServerForTests,
|
|
resetFlowRingsForTests,
|
|
} from "./traffic-flow-ingest.js"
|
|
import { buildFlowAnalytics, formatLiveSseFromBuilder, getFlowMonthly, listFlowClients, listFlowExporters } from "./traffic-flow-analytics.js"
|
|
import { withPgOrSkip } from "../test/pg.js"
|
|
import { dbQuery } from "../db/index.js"
|
|
import { seedFlowTopologyForTests, type FlowTopology } from "./traffic-flow-topology.js"
|
|
import { disableCatalogFetchForTests, resetFlowCatalogForTests, seedFlowCatalogForTests } from "./traffic-flow-classify.js"
|
|
import {
|
|
disableRipeEnqueueForTests,
|
|
disableRipePersistForTests,
|
|
resetRipeCacheForTests,
|
|
seedRipeCacheForTests,
|
|
} from "./traffic-flow-ripe.js"
|
|
|
|
{
|
|
const liveErr = await formatLiveSseFromBuilder(() => {
|
|
throw new Error("SQLITE_BUSY")
|
|
})
|
|
assert.equal(liveErr.event, "error")
|
|
assert.equal((liveErr.data as { error: string }).error, "SQLITE_BUSY")
|
|
const liveOk = await formatLiveSseFromBuilder(() => ({ ok: true }))
|
|
assert.equal(liveOk.event, "sample")
|
|
const liveAsync = await formatLiveSseFromBuilder(async () => ({
|
|
uniqueSrc: 3,
|
|
destinations: [{ id: "8.8.8.8", label: "8.8.8.8", bytes: 1, packets: 1, bps: 1, percent: 100 }],
|
|
}))
|
|
assert.equal(liveAsync.event, "sample")
|
|
assert.notEqual(JSON.stringify(liveAsync.data), "{}")
|
|
assert.equal((liveAsync.data as { uniqueSrc: number }).uniqueSrc, 3)
|
|
assert.ok(Array.isArray((liveAsync.data as { destinations: unknown[] }).destinations))
|
|
const liveReject = await formatLiveSseFromBuilder(async () => {
|
|
throw new Error("pg down")
|
|
})
|
|
assert.equal(liveReject.event, "error")
|
|
assert.equal((liveReject.data as { error: string }).error, "pg down")
|
|
}
|
|
|
|
if (!(await withPgOrSkip())) {
|
|
console.log("traffic-flow-analytics.test.ts: skip")
|
|
process.exit(0)
|
|
}
|
|
|
|
disableCatalogFetchForTests()
|
|
resetFlowCatalogForTests()
|
|
disableRipePersistForTests()
|
|
resetRipeCacheForTests()
|
|
disableRipeEnqueueForTests()
|
|
resetIfaceCacheForTests()
|
|
resetFlowRingsForTests()
|
|
seedFlowTopologyForTests({
|
|
clientIfaces: new Map(),
|
|
clientByIface: new Map(),
|
|
enNodes: [],
|
|
enHosts: new Set(),
|
|
jhHosts: new Set(),
|
|
wanIfaces: new Map(),
|
|
plane: { clientIfaceNames: new Set(), enHosts: new Set(), jhHosts: new Set() },
|
|
})
|
|
rememberServerIfaces(7, [
|
|
{ ".id": "*2", name: "ether1" },
|
|
{ ".id": "*B", name: "ether2" },
|
|
{ ".id": "*A", name: "wg-flow" },
|
|
])
|
|
|
|
ingestParsedFlowsForServerForTests(7, [
|
|
{
|
|
src: "10.1.1.8",
|
|
dst: "8.8.8.8",
|
|
proto: 6,
|
|
srcPort: 51234,
|
|
dstPort: 443,
|
|
bytes: 12_000,
|
|
packets: 10,
|
|
inIface: "2",
|
|
outIface: "11",
|
|
},
|
|
{
|
|
src: "10.1.1.8",
|
|
dst: "1.1.1.1",
|
|
proto: 17,
|
|
srcPort: 53000,
|
|
dstPort: 53,
|
|
bytes: 800,
|
|
packets: 4,
|
|
inIface: "2",
|
|
outIface: "",
|
|
},
|
|
])
|
|
|
|
try {
|
|
const all = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
|
|
assert.equal(all.applications[0]?.label, "HTTPS")
|
|
assert.ok(all.protocols.some((p) => p.label === "TCP"))
|
|
assert.equal(all.ifaces[0]?.name, "ether1")
|
|
assert.notEqual(all.ifaces[0]?.name, "2")
|
|
const conv = all.conversationsList[0]
|
|
assert.ok(conv)
|
|
assert.equal(conv.inIface, "ether1")
|
|
assert.equal(conv.inIfaceIndex, "2")
|
|
assert.equal(conv.application, "HTTPS")
|
|
assert.ok(!/^\d+$/.test(conv.inIface))
|
|
|
|
const filtered = await buildFlowAnalytics({ minutes: 5, serverId: 7, iface: "ether1" })
|
|
assert.ok(filtered.bytes >= 12_000)
|
|
assert.equal(filtered.ifaces[0]?.name, "ether1")
|
|
|
|
const miss = await buildFlowAnalytics({ minutes: 5, serverId: 7, iface: "wg-flow" })
|
|
assert.equal(miss.conversations, 0)
|
|
|
|
const other = await buildFlowAnalytics({ minutes: 5, serverId: 99 })
|
|
assert.equal(other.conversations, 0)
|
|
} finally {
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
}
|
|
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
rememberServerIfaces(7, [
|
|
{ ".id": "*2", name: "ether1" },
|
|
{ ".id": "*B", name: "ether2" },
|
|
{ ".id": "*A", name: "wg-flow" },
|
|
])
|
|
ingestParsedFlowsForServerForTests(7, [
|
|
{
|
|
src: "10.1.1.8",
|
|
dst: "8.8.8.8",
|
|
proto: 6,
|
|
srcPort: 51234,
|
|
dstPort: 443,
|
|
bytes: 12_000,
|
|
packets: 10,
|
|
inIface: "2",
|
|
outIface: "11",
|
|
},
|
|
{
|
|
src: "10.1.1.8",
|
|
dst: "8.8.8.8",
|
|
proto: 6,
|
|
srcPort: 51234,
|
|
dstPort: 443,
|
|
bytes: 9_000,
|
|
packets: 9,
|
|
inIface: "11",
|
|
outIface: "",
|
|
},
|
|
])
|
|
try {
|
|
const summed = await buildFlowAnalytics({ minutes: 5, serverId: 7, dedup: false })
|
|
assert.equal(summed.bytes, 21_000)
|
|
assert.equal(summed.conversations, 2)
|
|
const deduped = await buildFlowAnalytics({ minutes: 5, serverId: 7, dedup: true })
|
|
assert.equal(deduped.bytes, 12_000)
|
|
assert.equal(deduped.conversations, 1)
|
|
assert.equal(deduped.dedupApplied, true)
|
|
assert.equal(deduped.interfaces.length, 2)
|
|
} finally {
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
}
|
|
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
disableRipeEnqueueForTests()
|
|
seedRipeCacheForTests({
|
|
prefix: "8.8.8.0/24",
|
|
asn: 15169,
|
|
country: "US",
|
|
lat: 37.4,
|
|
lng: -122.1,
|
|
holder: "GOOGLE",
|
|
ok: true,
|
|
fetchedAt: Date.now(),
|
|
})
|
|
seedFlowCatalogForTests({
|
|
cidrs: [{ cidr: "8.8.8.0/24", purpose: "steam-gaming" }],
|
|
})
|
|
rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
|
|
ingestParsedFlowsForServerForTests(7, [
|
|
{
|
|
src: "10.1.1.8",
|
|
dst: "8.8.8.8",
|
|
proto: 6,
|
|
srcPort: 51234,
|
|
dstPort: 443,
|
|
bytes: 12_000,
|
|
packets: 10,
|
|
inIface: "2",
|
|
outIface: "",
|
|
},
|
|
])
|
|
try {
|
|
const geo = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
|
|
assert.equal(geo.categories?.[0]?.label, "Игры")
|
|
assert.ok(geo.asns?.some((r) => r.label.includes("AS15169")))
|
|
assert.equal(geo.countries?.[0]?.id, "US")
|
|
assert.equal(geo.mapEdges?.[0]?.toCountry, "US")
|
|
assert.ok(geo.mapEdges?.every((e) => e.toCountry !== "?"))
|
|
assert.equal(geo.conversationsList[0]?.dstCountry, "US")
|
|
assert.equal(geo.asns?.[0]?.id, "15169")
|
|
} finally {
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
resetRipeCacheForTests()
|
|
resetFlowCatalogForTests()
|
|
}
|
|
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
disableRipeEnqueueForTests()
|
|
resetRipeCacheForTests()
|
|
resetFlowCatalogForTests()
|
|
seedRipeCacheForTests({
|
|
prefix: "1.1.1.0/24",
|
|
asn: 13335,
|
|
country: "?",
|
|
lat: null,
|
|
lng: null,
|
|
holder: "CLOUDFLARENET, US",
|
|
ok: true,
|
|
fetchedAt: Date.now(),
|
|
})
|
|
rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
|
|
ingestParsedFlowsForServerForTests(7, [
|
|
{
|
|
src: "10.1.1.8",
|
|
dst: "1.1.1.1",
|
|
proto: 6,
|
|
srcPort: 51234,
|
|
dstPort: 443,
|
|
bytes: 5000,
|
|
packets: 5,
|
|
inIface: "2",
|
|
outIface: "",
|
|
},
|
|
])
|
|
try {
|
|
const cf = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
|
|
assert.equal(cf.countries?.[0]?.id, "US")
|
|
assert.ok(cf.mapEdges?.every((e) => e.toCountry !== "?"))
|
|
assert.equal(cf.services?.[0]?.label, "Cloudflare")
|
|
assert.equal(cf.categories?.[0]?.label, "CDN")
|
|
assert.equal(cf.conversationsList[0]?.dstCountry, "US")
|
|
assert.equal(cf.asns?.[0]?.id, "13335")
|
|
} finally {
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
resetRipeCacheForTests()
|
|
resetFlowCatalogForTests()
|
|
}
|
|
|
|
{
|
|
ingestParsedFlowsForServerForTests(7, [
|
|
{
|
|
src: "10.1.1.8",
|
|
dst: "8.8.8.8",
|
|
proto: 6,
|
|
srcPort: 51234,
|
|
dstPort: 443,
|
|
bytes: 12_000,
|
|
packets: 10,
|
|
inIface: "2",
|
|
outIface: "",
|
|
},
|
|
])
|
|
const degraded = await buildFlowAnalytics({ minutes: 5, serverId: 7, skipHeavy: true })
|
|
assert.equal(degraded.degraded, true)
|
|
assert.equal(degraded.conversationsList.length, 0)
|
|
assert.ok((degraded.bytes ?? 0) >= 12_000)
|
|
const exporters = await listFlowExporters(5)
|
|
const clients = await listFlowClients(5)
|
|
assert.ok(Array.isArray(exporters.exporters))
|
|
assert.ok(Array.isArray(clients.clients))
|
|
resetFlowRingsForTests()
|
|
}
|
|
|
|
{
|
|
if (await withPgOrSkip()) {
|
|
const { pool } = await import("../db/index.js")
|
|
const { ensurePartitionFor } = await import("../db/partitions.js")
|
|
await ensurePartitionFor(pool, "flow_daily_dims", "month", new Date("2026-09-01T00:00:00Z"))
|
|
await dbQuery(`DELETE FROM flow_daily_dims WHERE server_id = 7 AND day >= '2026-09-01' AND day < '2026-10-01'`)
|
|
await dbQuery(`
|
|
INSERT INTO flow_daily_dims (server_id, day, dim, key, bytes, packets)
|
|
VALUES
|
|
(7, '2026-09-01', 'country', 'US', 1000, 10),
|
|
(7, '2026-09-02', 'country', 'US', 500, 5),
|
|
(7, '2026-09-01', 'service', 'steam', 800, 8),
|
|
(7, '2026-09-01', 'asn', '15169', 900, 9),
|
|
(7, '2026-09-01', 'asn', 'other', 100, 1)
|
|
`)
|
|
const monthly = await getFlowMonthly("2026-09", 7)
|
|
assert.equal(monthly.bytes, 1500)
|
|
assert.equal(monthly.countries[0]?.id, "US")
|
|
assert.equal(monthly.countries[0]?.bytes, 1500)
|
|
assert.ok(monthly.asns.some((row) => row.id === "other"))
|
|
await dbQuery(`DELETE FROM flow_daily_dims WHERE server_id = 7 AND day >= '2026-09-01' AND day < '2026-10-01'`)
|
|
}
|
|
}
|
|
|
|
{
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
resetRipeCacheForTests()
|
|
resetFlowCatalogForTests()
|
|
disableRipeEnqueueForTests()
|
|
rememberServerIfaces(7, [
|
|
{ ".id": "*2", name: "gre-client" },
|
|
{ ".id": "*3", name: "NSK-SERVHOST-RTK" },
|
|
{ ".id": "*4", name: "gre-en-nsk" },
|
|
])
|
|
const topo: FlowTopology = {
|
|
clientIfaces: new Map([[7, new Set(["gre-client"])]]),
|
|
clientByIface: new Map([["7|gre-client", {
|
|
userId: "u1",
|
|
login: "alice",
|
|
name: "Alice",
|
|
serverId: 7,
|
|
interfaceName: "gre-client",
|
|
}]]),
|
|
enNodes: [{ id: 9, name: "NSK-SERVHOST-RTK", hosts: ["198.51.100.1"] }],
|
|
enHosts: new Set(["198.51.100.1"]),
|
|
jhHosts: new Set(["203.0.113.10"]),
|
|
wanIfaces: new Map(),
|
|
plane: {
|
|
clientIfaceNames: new Set(["gre-client"]),
|
|
enHosts: new Set(["198.51.100.1"]),
|
|
jhHosts: new Set(["203.0.113.10"]),
|
|
},
|
|
}
|
|
seedFlowTopologyForTests(topo)
|
|
seedRipeCacheForTests({
|
|
prefix: "173.194.0.0/16",
|
|
asn: 15169,
|
|
country: "US",
|
|
lat: null,
|
|
lng: null,
|
|
holder: "GOOGLE",
|
|
ok: true,
|
|
fetchedAt: Date.now(),
|
|
})
|
|
ingestParsedFlowsForServerForTests(7, [
|
|
{
|
|
src: "10.100.1.17",
|
|
dst: "173.194.160.163",
|
|
proto: 6,
|
|
srcPort: 51234,
|
|
dstPort: 443,
|
|
bytes: 12_000,
|
|
packets: 10,
|
|
inIface: "2",
|
|
outIface: "3",
|
|
nextHop: "198.51.100.1",
|
|
},
|
|
{
|
|
src: "203.0.113.10",
|
|
dst: "198.51.100.1",
|
|
proto: 47,
|
|
srcPort: 0,
|
|
dstPort: 0,
|
|
bytes: 5_000_000,
|
|
packets: 4000,
|
|
inIface: "4",
|
|
outIface: "4",
|
|
},
|
|
{
|
|
src: "10.100.1.17",
|
|
dst: "10.100.1.18",
|
|
proto: 6,
|
|
srcPort: 50000,
|
|
dstPort: 443,
|
|
bytes: 8000,
|
|
packets: 8,
|
|
inIface: "2",
|
|
outIface: "2",
|
|
},
|
|
])
|
|
try {
|
|
const def = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
|
|
assert.equal(def.bytes, 12_000)
|
|
assert.equal(def.bytesPayload, 12_000)
|
|
assert.equal(def.bytesOverlay, 5_000_000)
|
|
assert.equal(def.bytesMesh, 8000)
|
|
assert.equal(def.excludeOverlayApplied, true)
|
|
assert.equal(def.excludeMeshApplied, true)
|
|
assert.ok(!def.conversationsList.some((r) => r.proto === 47))
|
|
assert.equal(def.conversationsList[0]?.service, "Google")
|
|
assert.equal(def.conversationsList[0]?.category, "Веб")
|
|
assert.equal(def.conversationsList[0]?.clientName, "Alice")
|
|
assert.equal(def.conversationsList[0]?.enName, "NSK-SERVHOST-RTK")
|
|
assert.equal(def.conversationsList[0]?.plane, "payload")
|
|
const path = def.paths?.[0]
|
|
assert.ok(path)
|
|
assert.equal(path.clientName, "Alice")
|
|
assert.equal(path.enName, "NSK-SERVHOST-RTK")
|
|
assert.equal(path.dst, "173.194.160.163")
|
|
const withAll = await buildFlowAnalytics({ minutes: 5, serverId: 7, excludeOverlay: false, excludeMesh: false })
|
|
assert.equal(withAll.bytes, 12_000 + 5_000_000 + 8000)
|
|
assert.ok(withAll.conversationsList.some((r) => r.plane === "overlay"))
|
|
assert.ok(withAll.conversationsList.some((r) => r.plane === "client_mesh"))
|
|
} finally {
|
|
seedFlowTopologyForTests(null)
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
resetRipeCacheForTests()
|
|
resetFlowCatalogForTests()
|
|
}
|
|
}
|
|
|
|
{
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
resetRipeCacheForTests()
|
|
disableRipeEnqueueForTests()
|
|
rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
|
|
ingestParsedFlowsForServerForTests(7, [
|
|
{
|
|
src: "173.194.151.65",
|
|
dst: "10.200.100.53",
|
|
proto: 6,
|
|
srcPort: 443,
|
|
dstPort: 57182,
|
|
bytes: 12_000,
|
|
packets: 10,
|
|
inIface: "2",
|
|
outIface: "2",
|
|
},
|
|
{
|
|
src: "104.18.35.51",
|
|
dst: "10.200.100.53",
|
|
proto: 6,
|
|
srcPort: 443,
|
|
dstPort: 53880,
|
|
bytes: 3_000,
|
|
packets: 4,
|
|
inIface: "2",
|
|
outIface: "2",
|
|
},
|
|
])
|
|
try {
|
|
const rev = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
|
|
const google = rev.conversationsList.find((r) => r.src === "173.194.151.65")
|
|
const cf = rev.conversationsList.find((r) => r.src === "104.18.35.51")
|
|
assert.equal(google?.service, "Google")
|
|
assert.equal(google?.category, "Веб")
|
|
assert.equal(cf?.service, "Cloudflare")
|
|
assert.equal(cf?.category, "CDN")
|
|
} finally {
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
resetRipeCacheForTests()
|
|
}
|
|
}
|
|
|
|
{
|
|
if (await withPgOrSkip()) {
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
const sidRow = await dbQuery<{ id: number }>(`SELECT id FROM servers LIMIT 1`)
|
|
const sid = sidRow.rows[0]?.id
|
|
if (sid) {
|
|
const { pool } = await import("../db/index.js")
|
|
const { ensurePartitionFor } = await import("../db/partitions.js")
|
|
await ensurePartitionFor(pool, "traffic_samples", "week", new Date())
|
|
rememberServerIfaces(sid, [{ ".id": "*4", name: "gre-en-nsk" }])
|
|
ingestParsedFlowsForServerForTests(sid, [{
|
|
src: "10.100.1.17",
|
|
dst: "8.8.8.8",
|
|
proto: 6,
|
|
srcPort: 1,
|
|
dstPort: 443,
|
|
bytes: 100,
|
|
packets: 1,
|
|
inIface: "4",
|
|
outIface: "4",
|
|
}])
|
|
await dbQuery(`
|
|
INSERT INTO traffic_samples (server_id, interface_name, sampled_at, rx_bytes, tx_bytes, rx_bps, tx_bps)
|
|
VALUES ($1, 'gre-en-nsk', now(), 9000000, 1000000, 40000000, 2000000)
|
|
`, [sid])
|
|
try {
|
|
const wire = await buildFlowAnalytics({ minutes: 5, serverId: sid })
|
|
assert.ok((wire.bpsWire ?? 0) >= 40_000_000)
|
|
assert.notEqual(wire.bpsWire, (wire.bytes * 8) / 300)
|
|
} finally {
|
|
await dbQuery(`DELETE FROM traffic_samples WHERE server_id = $1 AND interface_name = 'gre-en-nsk'`, [sid])
|
|
}
|
|
}
|
|
resetFlowRingsForTests()
|
|
resetIfaceCacheForTests()
|
|
}
|
|
}
|
|
|
|
console.log("traffic-flow-analytics.test.ts: ok")
|