diff --git a/backend/package.json b/backend/package.json index febecb1..f422cb5 100644 --- a/backend/package.json +++ b/backend/package.json @@ -14,7 +14,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-dedup.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-analytics.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-dedup.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", "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts" }, "dependencies": { diff --git a/backend/src/services/traffic-flow-analytics.test.ts b/backend/src/services/traffic-flow-analytics.test.ts index 634d420..0aaf060 100644 --- a/backend/src/services/traffic-flow-analytics.test.ts +++ b/backend/src/services/traffic-flow-analytics.test.ts @@ -157,7 +157,53 @@ try { 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 = 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() diff --git a/backend/src/services/traffic-flow-analytics.ts b/backend/src/services/traffic-flow-analytics.ts index 08b790c..6f8f17f 100644 --- a/backend/src/services/traffic-flow-analytics.ts +++ b/backend/src/services/traffic-flow-analytics.ts @@ -14,7 +14,7 @@ import { protoName } from "./traffic-flow-parse.js" import { getFlowListenerState, getRingMbps, - listStoredFlowRows, + listFlowRowsForWindow, type PendingFlowRow, } from "./traffic-flow-ingest.js" import { resolveIfaceName } from "./traffic-flow-ifaces.js" @@ -23,6 +23,7 @@ import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js" import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js" import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js" import { classifyFlowDst, refreshFlowCatalogInBackground } from "./traffic-flow-classify.js" +import { isIsoCountry } from "./traffic-flow-brands.js" export interface FlowAnalyticsQuery { minutes: number @@ -37,14 +38,18 @@ function bpsToMbps(bps: number): number { return bps / 1_000_000 } -function topN(map: Map, windowSec: number, n: number): FlowBreakdownRow[] { +function topN( + map: Map, + windowSec: number, + n: number, +): FlowBreakdownRow[] { const total = [...map.values()].reduce((a, v) => a + v.bytes, 0) || 1 return [...map.entries()] .sort((a, b) => b[1].bytes - a[1].bytes) .slice(0, n) .map(([id, v]) => ({ id, - label: id, + label: v.label || id, bytes: v.bytes, packets: v.packets, bps: (v.bytes * 8) / windowSec, @@ -52,10 +57,17 @@ function topN(map: Map, windowSec: n })) } -function bump(map: Map, id: string, bytes: number, packets: number) { - const prev = map.get(id) ?? { bytes: 0, packets: 0 } +function bump( + map: Map, + id: string, + bytes: number, + packets: number, + label?: string, +) { + const prev = map.get(id) ?? { bytes: 0, packets: 0, label } prev.bytes += bytes prev.packets += packets + if (label) prev.label = label map.set(id, prev) } @@ -95,13 +107,13 @@ function snapshotStatus(serverId: number): FlowEntityCard["status"] { return "online" } -function topLabel(map: Map, fallback = "—"): string { +function topLabel(map: Map, fallback = "—"): string { let best = fallback let bestBytes = 0 - for (const [label, v] of map) { + for (const [id, v] of map) { if (v.bytes > bestBytes) { bestBytes = v.bytes - best = label + best = v.label || id } } return best @@ -111,8 +123,7 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto { const settings = getTrafficFlowSettingsRow() const top = Math.min(50, Math.max(10, settings.topN)) const windowSec = Math.max(60, q.minutes * 60) - const sinceIso = new Date(Date.now() - q.minutes * 60_000).toISOString() - const raw = listStoredFlowRows(sinceIso) + const raw = listFlowRowsForWindow(q.minutes) const allow = q.userId ? userIfaceAllow(q.userId) : null const serverRows = db.select().from(servers).all() const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host])) @@ -122,15 +133,15 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto { refreshFlowCatalogInBackground() - const applications = new Map() - const protocols = new Map() - const sources = new Map() - const destinations = new Map() + const applications = new Map() + const protocols = new Map() + const sources = new Map() + const destinations = new Map() const ifacesMap = new Map() - const asns = new Map() - const countries = new Map() - const categories = new Map() - const services = new Map() + const asns = new Map() + const countries = new Map() + const categories = new Map() + const services = new Map() const conv = new Map() const edgeAcc = new Map }>() const srcs = new Set() @@ -170,11 +181,13 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto { bump(categories, classified.category, r.bytes, r.packets) bump(services, classified.service, r.bytes, r.packets) if (ripe?.ok && ripe.asn) { + const asnId = String(ripe.asn) const asnLabel = ripe.holder ? `AS${ripe.asn} ${ripe.holder}` : `AS${ripe.asn}` - bump(asns, asnLabel, r.bytes, r.packets) + bump(asns, asnId, r.bytes, r.packets, asnLabel) } - if (ripe?.ok && ripe.country && ripe.country !== "—") { - bump(countries, ripe.country, r.bytes, r.packets) + const dstCountry = ripe?.ok && isIsoCountry(ripe.country) ? ripe.country : "" + if (dstCountry) { + bump(countries, dstCountry, r.bytes, r.packets) } const ckey = wantDedup @@ -203,13 +216,13 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto { application: app, category: classified.category, service: classified.service, - dstCountry: ripe?.country && ripe.country !== "—" ? ripe.country : undefined, + dstCountry: dstCountry || undefined, dstAsn: ripe?.asn || undefined, rawBytes: r.bytes, }) } - const toCountry = ripe?.ok && ripe.country && ripe.country !== "—" ? ripe.country : "" + const toCountry = dstCountry if (toCountry) { const fromCountry = countryById.get(r.serverId) || "UN" const ekey = `${r.serverId}|${toCountry}` @@ -348,8 +361,7 @@ function cardFromServer( export function listFlowExporters(minutes: number): FlowExportersDto { const settings = getTrafficFlowSettingsRow() - const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString() - const rows = listStoredFlowRows(sinceIso) + const rows = listFlowRowsForWindow(minutes) const ids = new Set() for (const r of rows) ids.add(r.serverId) for (const p of listHostPeers()) ids.add(p.serverId) diff --git a/backend/src/services/traffic-flow-brands.test.ts b/backend/src/services/traffic-flow-brands.test.ts new file mode 100644 index 0000000..54d141c --- /dev/null +++ b/backend/src/services/traffic-flow-brands.test.ts @@ -0,0 +1,26 @@ +import assert from "node:assert/strict" +import { + brandByAsn, + countryFromHolder, + lookupBrand, + OTHER_SERVICE, + resolveRipeCountry, +} from "./traffic-flow-brands.js" + +assert.equal(resolveRipeCountry("?", 13335, "CLOUDFLARENET, US"), "US") +assert.equal(resolveRipeCountry("EU", 13335, ""), "US") +assert.equal(resolveRipeCountry("?", 0, "CLOUDFLARENET, US"), "US") +assert.equal(countryFromHolder("CLOUDFLARENET, US"), "US") +assert.equal(resolveRipeCountry("NL", 0, ""), "NL") +assert.equal(resolveRipeCountry("?", 0, ""), "") + +assert.equal(brandByAsn(13335)?.service, "Cloudflare") +assert.equal(brandByAsn(13335)?.category, "CDN") +assert.equal(brandByAsn(32590)?.service, "Steam") +assert.equal(brandByAsn(32590)?.category, "Игры") +assert.equal(brandByAsn(401115)?.service, "ChatGPT") +assert.equal(lookupBrand("1.1.1.1", 13335)?.service, "Cloudflare") +assert.equal(lookupBrand("203.0.113.9", 64500), null) +assert.equal(OTHER_SERVICE, "Прочее") + +console.log("traffic-flow-brands.test.ts: ok") diff --git a/backend/src/services/traffic-flow-brands.ts b/backend/src/services/traffic-flow-brands.ts new file mode 100644 index 0000000..72d28d8 --- /dev/null +++ b/backend/src/services/traffic-flow-brands.ts @@ -0,0 +1,105 @@ +import { ipInCidrV4, parseCidrV4 } from "./traffic-flow-ip.js" + +export const OTHER_SERVICE = "Прочее" + +export interface BrandHit { + service: string + category: string +} + +const ASN_BRANDS = new Map([ + [13335, { service: "Cloudflare", category: "CDN" }], + [209242, { service: "Cloudflare", category: "CDN" }], + [54113, { service: "Fastly", category: "CDN" }], + [20940, { service: "Akamai", category: "CDN" }], + [16509, { service: "Amazon", category: "CDN" }], + [14618, { service: "Amazon", category: "CDN" }], + [8075, { service: "Microsoft", category: "CDN" }], + [13238, { service: "Yandex", category: "CDN" }], + [32590, { service: "Steam", category: "Игры" }], + [2906, { service: "Netflix", category: "Видео / стриминг" }], + [40027, { service: "Netflix", category: "Видео / стриминг" }], + [15169, { service: "Google", category: "Видео / стриминг" }], + [36040, { service: "YouTube", category: "Видео / стриминг" }], + [46489, { service: "Twitch", category: "Видео / стриминг" }], + [401115, { service: "ChatGPT", category: "ИИ" }], + [49544, { service: "Discord", category: "Голос" }], + [62041, { service: "Telegram", category: "Голос" }], + [59930, { service: "Telegram", category: "Голос" }], + [211157, { service: "Telegram", category: "Голос" }], + [32934, { service: "Meta", category: "CDN" }], + [396986, { service: "TikTok", category: "Видео / стриминг" }], +]) + +const ASN_HQ_COUNTRY = new Map([ + [13335, "US"], + [209242, "US"], + [54113, "US"], + [20940, "US"], + [16509, "US"], + [14618, "US"], + [8075, "US"], + [15169, "US"], + [32590, "US"], + [2906, "US"], + [40027, "US"], + [36040, "US"], + [46489, "US"], + [401115, "US"], + [49544, "US"], + [32934, "US"], + [13238, "RU"], + [62041, "NL"], + [59930, "NL"], + [211157, "NL"], +]) + +const CIDR_BRANDS: Array<{ cidr: string; prefixLen: number; hit: BrandHit }> = [ + { cidr: "104.16.0.0/13", prefixLen: 13, hit: { service: "Cloudflare", category: "CDN" } }, + { cidr: "104.24.0.0/14", prefixLen: 14, hit: { service: "Cloudflare", category: "CDN" } }, + { cidr: "172.64.0.0/13", prefixLen: 13, hit: { service: "Cloudflare", category: "CDN" } }, + { cidr: "162.158.0.0/15", prefixLen: 15, hit: { service: "Cloudflare", category: "CDN" } }, +].sort((a, b) => b.prefixLen - a.prefixLen) + +const NON_ISO = new Set(["EU", "AP", "ZZ", "XX", "A1", "A2", "O1"]) + +export function isIsoCountry(code: string): boolean { + const c = String(code ?? "").trim().toUpperCase() + return /^[A-Z]{2}$/.test(c) && !NON_ISO.has(c) +} + +export function normalizeIsoCountry(code: string): string { + const c = String(code ?? "").trim().toUpperCase() + return isIsoCountry(c) ? c : "" +} + +/** `CLOUDFLARENET, US` → `US`. */ +export function countryFromHolder(holder: string): string { + const m = String(holder ?? "").trim().match(/,\s*([A-Za-z]{2})\s*$/) + return m?.[1] ? normalizeIsoCountry(m[1]) : "" +} + +export function countryForAsn(asn: number): string { + if (!asn) return "" + return ASN_HQ_COUNTRY.get(asn) ?? "" +} + +export function resolveRipeCountry(country: string, asn: number, holder: string): string { + return normalizeIsoCountry(country) || countryFromHolder(holder) || countryForAsn(asn) +} + +export function brandByAsn(asn: number): BrandHit | null { + if (!asn) return null + return ASN_BRANDS.get(asn) ?? null +} + +export function brandByCidr(ip: string): BrandHit | null { + for (const row of CIDR_BRANDS) { + if (parseCidrV4(row.cidr) && ipInCidrV4(ip, row.cidr)) return row.hit + } + return null +} + +export function lookupBrand(ip: string, asn: number): BrandHit | null { + return brandByCidr(ip) || brandByAsn(asn) +} diff --git a/backend/src/services/traffic-flow-classify.test.ts b/backend/src/services/traffic-flow-classify.test.ts index 495c453..b6bb797 100644 --- a/backend/src/services/traffic-flow-classify.test.ts +++ b/backend/src/services/traffic-flow-classify.test.ts @@ -16,5 +16,10 @@ assert.equal(miss.category, "DNS") const cdn = classifyFlowDst("203.0.113.9", 6, 443, 1, { prefix: "203.0.113.0/24", asn: 13335, country: "US", lat: null, lng: null, holder: "CLOUDFLARENET", ok: true, fetchedAt: Date.now() }) assert.equal(cdn.category, "CDN") +assert.equal(cdn.service, "Cloudflare") + +const amazonHolder = classifyFlowDst("203.0.113.50", 6, 443, 1, { prefix: "203.0.113.0/24", asn: 64500, country: "RU", lat: null, lng: null, holder: "AMAZON-AES - Amazon.com, Inc.", ok: true, fetchedAt: Date.now() }) +assert.equal(amazonHolder.service, "Прочее") +assert.notEqual(amazonHolder.service, "AMAZON-AES - Amazon.com, Inc.") console.log("traffic-flow-classify.test.ts: ok") diff --git a/backend/src/services/traffic-flow-classify.ts b/backend/src/services/traffic-flow-classify.ts index 30e7ac7..1093f67 100644 --- a/backend/src/services/traffic-flow-classify.ts +++ b/backend/src/services/traffic-flow-classify.ts @@ -1,3 +1,4 @@ +import { lookupBrand, OTHER_SERVICE } from "./traffic-flow-brands.js" import { db } from "../db/index.js" import { evobgpSettings } from "../db/schema.js" import { ipInCidrV4, parseCidrV4 } from "./traffic-flow-ip.js" @@ -50,9 +51,10 @@ export function categoryFromPurpose(purpose: string, proto: number, dstPort: num if (/streaming|youtube|netflix|twitch|video/.test(p)) return "Видео / стриминг" if (/cdn|cloudflare|akamai|fastly/.test(p)) return "CDN" if (/voip|discord|zoom/.test(p)) return "Голос" + if (/openai|chatgpt|\bai\b/.test(p)) return "ИИ" const app = applicationName(proto, dstPort, srcPort) if (app === "DNS" || app === "SSH" || app === "BGP") return app - return "Проче" + return OTHER_SERVICE } function matchCidr(ip: string): CatalogCidr | null { @@ -70,13 +72,13 @@ export function classifyFlowDst( ripe: FlowIpMeta | null, ): FlowClassification { const hit = matchCidr(dst) + const brand = lookupBrand(dst, ripe?.asn ?? 0) const asnName = ripe?.asn ? asnPurpose.get(ripe.asn) : undefined - const purpose = hit?.purpose || asnName || ripe?.holder || "" - const service = (hit?.purpose || asnName || ripe?.holder || "Проче").trim() || "Проче" - return { - service, - category: categoryFromPurpose(purpose, proto, dstPort, srcPort), - } + const service = (hit?.purpose || brand?.service || asnName || OTHER_SERVICE).trim() || OTHER_SERVICE + const category = hit + ? categoryFromPurpose(hit.purpose, proto, dstPort, srcPort) + : (brand?.category || categoryFromPurpose(asnName || "", proto, dstPort, srcPort)) + return { service, category } } async function fetchCatalog(): Promise { diff --git a/backend/src/services/traffic-flow-ifaces.test.ts b/backend/src/services/traffic-flow-ifaces.test.ts index 3d04e97..4eec62d 100644 --- a/backend/src/services/traffic-flow-ifaces.test.ts +++ b/backend/src/services/traffic-flow-ifaces.test.ts @@ -4,6 +4,8 @@ import { resetIfaceCacheForTests, resolveIfaceName, rosIdToIfIndex, + shouldRefreshIfaces, + markIfaceRefreshAttempt, } from "./traffic-flow-ifindex.js" import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js" @@ -35,6 +37,14 @@ assert.equal(rosIdToIfIndex("*12"), 18) assert.equal(resolveIfaceName(8, "10").name, "gre1") assert.equal(resolveIfaceName(8, "18").name, "gre1") +resetIfaceCacheForTests() +assert.equal(shouldRefreshIfaces(9), true) +rememberServerIfaces(9, [{ ".id": "*2", name: "ether1" }]) +assert.equal(shouldRefreshIfaces(9), false) +resetIfaceCacheForTests() +markIfaceRefreshAttempt(9) +assert.equal(shouldRefreshIfaces(9), false) + assert.equal(applicationName(6, 443), "HTTPS") assert.equal(applicationName(17, 53), "DNS") assert.equal(applicationName(6, 22), "SSH") diff --git a/backend/src/services/traffic-flow-ifaces.ts b/backend/src/services/traffic-flow-ifaces.ts index f2544ba..898e4d9 100644 --- a/backend/src/services/traffic-flow-ifaces.ts +++ b/backend/src/services/traffic-flow-ifaces.ts @@ -3,8 +3,9 @@ import { db } from "../db/index.js" import { servers } from "../db/schema.js" import { MikrotikClient } from "./mikrotik.js" import { - ifaceCacheFresh, rememberServerIfaces, + shouldRefreshIfaces, + markIfaceRefreshAttempt, type RosIfaceIndexRow, } from "./traffic-flow-ifindex.js" @@ -15,13 +16,15 @@ export { resetIfaceCacheForTests, resolveIfaceName, rosIdToIfIndex, + shouldRefreshIfaces, + markIfaceRefreshAttempt, } from "./traffic-flow-ifindex.js" const inflight = new Set() export async function refreshServerIfaces(serverId: number, force = false): Promise { if (inflight.has(serverId)) return - if (!force && ifaceCacheFresh(serverId)) return + if (!force && !shouldRefreshIfaces(serverId)) return inflight.add(serverId) try { const row = db.select().from(servers).where(eq(servers.id, serverId)).limit(1).all()[0] @@ -32,6 +35,7 @@ export async function refreshServerIfaces(serverId: number, force = false): Prom } catch { /* keep previous cache */ } finally { + markIfaceRefreshAttempt(serverId) inflight.delete(serverId) } } diff --git a/backend/src/services/traffic-flow-ifindex.ts b/backend/src/services/traffic-flow-ifindex.ts index 65df4c3..e8dac65 100644 --- a/backend/src/services/traffic-flow-ifindex.ts +++ b/backend/src/services/traffic-flow-ifindex.ts @@ -6,6 +6,7 @@ export interface RosIfaceIndexRow { const cache = new Map>() const fetchedAt = new Map() +const lastAttempt = new Map() export const IFACE_CACHE_TTL_MS = 60_000 @@ -52,7 +53,19 @@ export function ifaceCacheFresh(serverId: number, ttlMs = IFACE_CACHE_TTL_MS): b return Boolean(prev && Date.now() - prev < ttlMs && cache.has(serverId)) } +/** Не ходить в REST, пока кэш жив или с момента последней попытки не прошёл TTL. */ +export function shouldRefreshIfaces(serverId: number, ttlMs = IFACE_CACHE_TTL_MS): boolean { + if (ifaceCacheFresh(serverId, ttlMs)) return false + const attempted = lastAttempt.get(serverId) ?? 0 + return !(attempted && Date.now() - attempted < ttlMs) +} + +export function markIfaceRefreshAttempt(serverId: number, at = Date.now()): void { + lastAttempt.set(serverId, at) +} + export function resetIfaceCacheForTests(): void { cache.clear() fetchedAt.clear() + lastAttempt.clear() } diff --git a/backend/src/services/traffic-flow-ingest.test.ts b/backend/src/services/traffic-flow-ingest.test.ts new file mode 100644 index 0000000..ec15d77 --- /dev/null +++ b/backend/src/services/traffic-flow-ingest.test.ts @@ -0,0 +1,44 @@ +import assert from "node:assert/strict" +import { + markIfaceRefreshAttempt, + rememberServerIfaces, + resetIfaceCacheForTests, + shouldRefreshIfaces, +} from "./traffic-flow-ifindex.js" +import { + lastFlushUsedTransactionForTests, + maybeRefreshIfaces, + resetFlowRingsForTests, + setRefreshIfacesForTests, +} from "./traffic-flow-ingest.js" + +resetIfaceCacheForTests() +resetFlowRingsForTests() + +let refreshCalls = 0 +setRefreshIfacesForTests(async () => { + refreshCalls += 1 +}) + +rememberServerIfaces(1, [{ ".id": "*A", name: "wg-flow" }]) +assert.equal(shouldRefreshIfaces(1), false) +assert.equal(maybeRefreshIfaces(1), false) +assert.equal(refreshCalls, 0) + +resetIfaceCacheForTests() +assert.equal(shouldRefreshIfaces(2), true) +assert.equal(maybeRefreshIfaces(2), true) +assert.equal(refreshCalls, 1) + +markIfaceRefreshAttempt(2) +assert.equal(shouldRefreshIfaces(2), false) +assert.equal(maybeRefreshIfaces(2), false) +assert.equal(refreshCalls, 1) + +assert.equal(lastFlushUsedTransactionForTests(), false) + +resetFlowRingsForTests() +resetIfaceCacheForTests() +setRefreshIfacesForTests(null) + +console.log("traffic-flow-ingest.test.ts: ok") diff --git a/backend/src/services/traffic-flow-ingest.ts b/backend/src/services/traffic-flow-ingest.ts index 02a71aa..97f3a3d 100644 --- a/backend/src/services/traffic-flow-ingest.ts +++ b/backend/src/services/traffic-flow-ingest.ts @@ -1,6 +1,6 @@ import { createSocket, type Socket } from "node:dgram" import { desc, eq, gte, sql } from "drizzle-orm" -import { db } from "../db/index.js" +import { db, sqliteDatabase } from "../db/index.js" import { flowBuckets, servers } from "../db/schema.js" import type { FlowStatsDto, FlowTalkerDto } from "@mmapp/contracts/traffic-flow" import { parseFlowPacket, protoName, type ParsedFlow } from "./traffic-flow-parse.js" @@ -11,7 +11,7 @@ import { recordFlowListenerError, recordFlowPacket, } from "./traffic-flow-settings.js" -import { ifaceCacheFresh, refreshServerIfaces, resolveIfaceName } from "./traffic-flow-ifaces.js" +import { refreshServerIfaces, resolveIfaceName, shouldRefreshIfaces } from "./traffic-flow-ifaces.js" import { applicationName } from "./traffic-flow-apps.js" export interface FlowListenerState { @@ -35,6 +35,8 @@ export interface PendingFlowRow { const TICK_MS = 2_000 const RING_LEN = 60 +const LIVE_WINDOW_MS = 15 * 60_000 +const PRUNE_MS = 5 * 60_000 let socket: Socket | null = null let state: FlowListenerState = { bound: false, address: null } @@ -45,7 +47,38 @@ const pending = new Map() +const recent = new Map() let flushTimer: ReturnType | null = null +let lastPruneAt = 0 +let refreshIfacesImpl: (serverId: number, force?: boolean) => Promise = refreshServerIfaces +let lastFlushUsedTransaction = false + +const upsertFlowStmt = sqliteDatabase.prepare(` + INSERT INTO flow_buckets ( + server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface + ) VALUES ( + @serverId, @bucketAt, @src, @dst, @proto, @srcPort, @dstPort, @bytes, @packets, @inIface + ) + ON CONFLICT(server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface) + DO UPDATE SET + bytes = bytes + excluded.bytes, + packets = packets + excluded.packets +`) + +const upsertFlowTx = sqliteDatabase.transaction((rows: Array<{ + serverId: number + bucketAt: string + src: string + dst: string + proto: number + srcPort: number + dstPort: number + bytes: number + packets: number + inIface: string +}>) => { + for (const row of rows) upsertFlowStmt.run(row) +}) const tickAccum = new Map() const rings = new Map() @@ -142,12 +175,55 @@ function resolveServerId(exporterIp: string): number | null { }) } +export function setRefreshIfacesForTests(fn: typeof refreshServerIfaces | null): void { + refreshIfacesImpl = fn ?? refreshServerIfaces +} + +/** REST /interface только при протухшем TTL, не из-за #N в пакете. */ +export function maybeRefreshIfaces(serverId: number): boolean { + if (!shouldRefreshIfaces(serverId)) return false + void refreshIfacesImpl(serverId) + return true +} + +export function lastFlushUsedTransactionForTests(): boolean { + return lastFlushUsedTransaction +} + +function pendingKey(serverId: number, bucketAt: string, flow: ParsedFlow): string { + return `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}\0${flow.inIface}` +} + +function rowKey(row: PendingFlowRow): string { + return `${row.serverId}|${row.bucketAt}|${row.src}|${row.dst}|${row.proto}|${row.srcPort}|${row.dstPort}|${row.inIface}` +} + +function mergeInto(map: Map, row: PendingFlowRow): void { + const key = rowKey(row) + const prev = map.get(key) + if (prev) { + prev.bytes += row.bytes + prev.packets += row.packets + return + } + map.set(key, { ...row }) +} + +function rememberRecent(rows: PendingFlowRow[]): void { + for (const row of rows) mergeInto(recent, row) +} + +function pruneRecent(sinceMs = Date.now() - LIVE_WINDOW_MS): void { + const cutoff = new Date(sinceMs).toISOString() + for (const [key, row] of recent) { + if (row.bucketAt < cutoff) recent.delete(key) + } +} + function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean { const serverId = resolveServerId(exporterIp) if (serverId == null) return false - const needsRefresh = !ifaceCacheFresh(serverId) - || flows.some((f) => resolveIfaceName(serverId, f.inIface).name.startsWith("#")) - if (needsRefresh) void refreshServerIfaces(serverId, true) + maybeRefreshIfaces(serverId) const bucketAt = minuteBucketIso() for (const flow of flows) { addToTick(serverId, flow.inIface, flow.outIface, flow.bytes) @@ -170,7 +246,17 @@ function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean { } export function peekPendingFlows(): PendingFlowRow[] { - return [...pending.values()].map((row) => ({ + return [...pending.values()].map(toPendingRow) +} + +function toPendingRow(row: { + serverId: number + bucketAt: string + flow: ParsedFlow + bytes: number + packets: number +}): PendingFlowRow { + return { serverId: row.serverId, bucketAt: row.bucketAt, src: row.flow.src || "0.0.0.0", @@ -182,53 +268,17 @@ export function peekPendingFlows(): PendingFlowRow[] { packets: row.packets, inIface: row.flow.inIface, outIface: row.flow.outIface, - })) + } } -function flushPending() { - if (pending.size === 0) return +function pruneStoredBuckets(): void { + const now = Date.now() + if (now - lastPruneAt < PRUNE_MS) return + lastPruneAt = now const settings = getTrafficFlowSettingsRow() const topN = Math.max(20, settings.topN) - const cutoff = new Date(Date.now() - settings.retentionHours * 3600_000).toISOString() - const rows = [...pending.values()] - pending.clear() - - for (const row of rows) { - try { - db.insert(flowBuckets).values({ - serverId: row.serverId, - bucketAt: row.bucketAt, - src: row.flow.src || "0.0.0.0", - dst: row.flow.dst || "0.0.0.0", - proto: row.flow.proto, - srcPort: row.flow.srcPort, - dstPort: row.flow.dstPort, - bytes: row.bytes, - packets: row.packets, - inIface: row.flow.inIface, - }).onConflictDoUpdate({ - target: [ - flowBuckets.serverId, - flowBuckets.bucketAt, - flowBuckets.src, - flowBuckets.dst, - flowBuckets.proto, - flowBuckets.srcPort, - flowBuckets.dstPort, - flowBuckets.inIface, - ], - set: { - bytes: sql`${flowBuckets.bytes} + excluded.bytes`, - packets: sql`${flowBuckets.packets} + excluded.packets`, - }, - }).run() - } catch { - // ignore single-row failures - } - } - + const cutoff = new Date(now - settings.retentionHours * 3600_000).toISOString() db.delete(flowBuckets).where(sql`${flowBuckets.bucketAt} < ${cutoff}`).run() - const latest = db.select({ bucketAt: flowBuckets.bucketAt }).from(flowBuckets) .orderBy(desc(flowBuckets.bucketAt)).limit(1).all()[0]?.bucketAt if (!latest) return @@ -248,6 +298,58 @@ function flushPending() { } } +function flushPending() { + pruneRecent() + if (pending.size === 0) { + pruneStoredBuckets() + lastFlushUsedTransaction = false + return + } + const rows = [...pending.values()].map(toPendingRow) + pending.clear() + rememberRecent(rows) + lastFlushUsedTransaction = false + try { + upsertFlowTx(rows.map((r) => ({ + serverId: r.serverId, + bucketAt: r.bucketAt, + src: r.src, + dst: r.dst, + proto: r.proto, + srcPort: r.srcPort, + dstPort: r.dstPort, + bytes: r.bytes, + packets: r.packets, + inIface: r.inIface, + }))) + lastFlushUsedTransaction = true + } catch { + for (const r of rows) { + try { + upsertFlowStmt.run({ + serverId: r.serverId, + bucketAt: r.bucketAt, + src: r.src, + dst: r.dst, + proto: r.proto, + srcPort: r.srcPort, + dstPort: r.dstPort, + bytes: r.bytes, + packets: r.packets, + inIface: r.inIface, + }) + } catch { + /* ignore single-row failures */ + } + } + } + pruneStoredBuckets() +} + +export function flushPendingForTests(): void { + flushPending() +} + function onTick() { rollFlowRings() flushPending() @@ -306,12 +408,24 @@ export function startTrafficFlowListener() { flushTimer = setInterval(onTick, TICK_MS) } +export function listLiveFlowRows(sinceIso: string): PendingFlowRow[] { + const merged = new Map() + for (const row of recent.values()) { + if (row.bucketAt < sinceIso) continue + mergeInto(merged, row) + } + for (const row of peekPendingFlows()) { + if (row.bucketAt < sinceIso) continue + mergeInto(merged, row) + } + return [...merged.values()] +} + export function listStoredFlowRows(sinceIso: string): PendingFlowRow[] { const stored = db.select().from(flowBuckets).where(gte(flowBuckets.bucketAt, sinceIso)).all() const merged = new Map() for (const r of stored) { - const key = `${r.serverId}|${r.bucketAt}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}|${r.inIface}` - merged.set(key, { + mergeInto(merged, { serverId: r.serverId, bucketAt: r.bucketAt, src: r.src, @@ -327,22 +441,21 @@ export function listStoredFlowRows(sinceIso: string): PendingFlowRow[] { } for (const p of peekPendingFlows()) { if (p.bucketAt < sinceIso) continue - const key = `${p.serverId}|${p.bucketAt}|${p.src}|${p.dst}|${p.proto}|${p.srcPort}|${p.dstPort}|${p.inIface}` - const prev = merged.get(key) - if (prev) { - prev.bytes += p.bytes - prev.packets += p.packets - } else { - merged.set(key, { ...p }) - } + mergeInto(merged, p) } return [...merged.values()] } +/** SSE / короткое окно — память; длинные окна — SQLite. */ +export function listFlowRowsForWindow(minutes: number): PendingFlowRow[] { + const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString() + if (minutes <= 15) return listLiveFlowRows(sinceIso) + return listStoredFlowRows(sinceIso) +} + export function listFlowTalkers(minutes = 5): FlowStatsDto { const settings = getTrafficFlowSettingsRow() - const rangeStart = new Date(Date.now() - minutes * 60_000).toISOString() - const rows = listStoredFlowRows(rangeStart) + const rows = listFlowRowsForWindow(minutes) const serverRows = db.select().from(servers).all() const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host])) const agg = new Map() @@ -426,7 +539,7 @@ export function ingestParsedFlowsForServerForTests(serverId: number, flows: Pars const bucketAt = minuteBucketIso() for (const flow of flows) { addToTick(serverId, flow.inIface, flow.outIface, flow.bytes) - const key = `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}\0${flow.inIface}` + const key = pendingKey(serverId, bucketAt, flow) const prev = pending.get(key) if (prev) { prev.bytes += flow.bytes @@ -448,4 +561,8 @@ export function resetFlowRingsForTests() { tickAccum.clear() rings.clear() pending.clear() + recent.clear() + lastPruneAt = 0 + lastFlushUsedTransaction = false + refreshIfacesImpl = refreshServerIfaces } diff --git a/backend/src/services/traffic-flow-ripe.test.ts b/backend/src/services/traffic-flow-ripe.test.ts index b91f2eb..a6bd826 100644 --- a/backend/src/services/traffic-flow-ripe.test.ts +++ b/backend/src/services/traffic-flow-ripe.test.ts @@ -69,4 +69,35 @@ enqueueRipeMisses(["203.0.113.50"]) await flushRipeQueueForTests() assert.equal(ripeFetchCountForTests(), afterNeg) +resetRipeCacheForTests() +disableRipePersistForTests() +seedRipeCacheForTests({ + prefix: "1.1.1.0/24", + asn: 13335, + country: "?", + lat: null, + lng: null, + holder: "CLOUDFLARENET, US", + ok: true, + fetchedAt: Date.now(), +}) +assert.equal(lookupRipeCached("1.1.1.1")?.country, "US") +assert.ok(lookupRipeCached("1.1.1.1")?.country !== "?") + +resetRipeCacheForTests() +disableRipePersistForTests() +setRipeFetchForTests(async (input) => { + const url = String(input) + const body = url.includes("network-info") + ? { data: { prefix: "1.0.0.0/24", asns: ["13335"] } } + : url.includes("maxmind-geo-lite") + ? { data: { located_resources: [{ locations: [{ country: "?" }] }] } } + : { data: { holder: "CLOUDFLARENET, US" } } + return new Response(JSON.stringify(body), { status: 200, headers: { "Content-Type": "application/json" } }) +}) +enqueueRipeMisses(["1.0.0.1"]) +await flushRipeQueueForTests() +assert.equal(lookupRipeCached("1.0.0.1")?.country, "US") +assert.equal(lookupRipeCached("1.0.0.1")?.asn, 13335) + console.log("traffic-flow-ripe.test.ts: ok") diff --git a/backend/src/services/traffic-flow-ripe.ts b/backend/src/services/traffic-flow-ripe.ts index 53ab431..7159223 100644 --- a/backend/src/services/traffic-flow-ripe.ts +++ b/backend/src/services/traffic-flow-ripe.ts @@ -1,5 +1,6 @@ import { sqliteDatabase } from "../db/index.js" import { ipInCidrV4, ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js" +import { resolveRipeCountry } from "./traffic-flow-brands.js" export interface FlowIpMeta { prefix: string @@ -15,6 +16,7 @@ export interface FlowIpMeta { const HIT_TTL_MS = 24 * 60 * 60_000 const NEG_TTL_MS = 6 * 60 * 60_000 const MAX_NEW_PREFIX_PER_MIN = 30 +const MAX_QUEUE = 90 const CONCURRENCY = 3 const RIPE_BASE = "https://stat.ripe.net/data" const UA = "MikrotikManager-flow/1.0" @@ -107,13 +109,15 @@ function loadSqlite(): void { }> for (const r of rows) { const fetchedAt = Date.parse(r.fetched_at) + const asn = Number(r.asn ?? 0) || 0 + const holder = r.holder || "" mem.set(r.prefix, { prefix: r.prefix, - asn: Number(r.asn ?? 0) || 0, - country: r.country || "—", + asn, + country: resolveRipeCountry(r.country || "", asn, holder) || "—", lat: r.lat == null ? null : Number(r.lat), lng: r.lng == null ? null : Number(r.lng), - holder: r.holder || "", + holder, ok: r.ok !== 0, fetchedAt: Number.isFinite(fetchedAt) ? fetchedAt : 0, }) @@ -206,6 +210,8 @@ export function lookupRipeCached(ip: string): FlowIpMeta | null { } } return best + ? { ...best, country: resolveRipeCountry(best.country, best.asn, best.holder) || "—" } + : null } async function ripeJson(path: string, resource: string): Promise { @@ -247,7 +253,7 @@ function pickGeo(data: unknown): { country: string; lat: number | null; lng: num } } const loc = d?.data?.located_resources?.[0]?.locations?.[0] - const country = String(loc?.country ?? "").trim().toUpperCase() + const country = resolveRipeCountry(String(loc?.country ?? ""), 0, "") const lat = loc?.latitude == null ? null : Number(loc.latitude) const lng = loc?.longitude == null ? null : Number(loc.longitude) return { @@ -299,14 +305,15 @@ async function resolveIp(ip: string): Promise { /* best-effort */ } } + const country = resolveRipeCountry(geo.country, asn, holder) const entry: FlowIpMeta = { prefix, asn, - country: geo.country, + country: country || "—", lat: geo.lat, lng: geo.lng, holder, - ok: Boolean(asn || (geo.country && geo.country !== "—")), + ok: Boolean(asn || country), fetchedAt: Date.now(), } mem.set(prefix, entry) @@ -360,6 +367,7 @@ export function enqueueRipeMisses(ips: Iterable): void { if (!enqueueEnabled) return loadSqlite() for (const raw of ips) { + if (queue.length >= MAX_QUEUE) break const ip = String(raw ?? "").trim() if (!ip || isNonPublicIp(ip)) continue if (lookupRipeCached(ip)) continue diff --git a/components/flag.tsx b/components/flag.tsx index b0acceb..ba63bd9 100644 --- a/components/flag.tsx +++ b/components/flag.tsx @@ -46,6 +46,7 @@ function nearestCdnSize(px: number): number { export function Flag({ code, size = 20, className }: FlagProps) { if (!code) return null const lower = code.toLowerCase() + if (!/^[a-z]{2}$/.test(lower)) return null const name = countryName(code.toUpperCase()) const cdnSrc = nearestCdnSize(size) const cdnSrc2x = nearestCdnSize(size * 2) diff --git a/components/traffic/flow-analytics-panel.tsx b/components/traffic/flow-analytics-panel.tsx index 0df90d0..36e4711 100644 --- a/components/traffic/flow-analytics-panel.tsx +++ b/components/traffic/flow-analytics-panel.tsx @@ -2,7 +2,7 @@ import { useMemo, useState } from "react" import { type ColumnDef, getCoreRowModel, useReactTable } from "@tanstack/react-table" -import type { FlowAnalyticsDto, FlowBreakdownRow, FlowEntityCard } from "@mmapp/contracts/traffic-flow" +import type { FlowAnalyticsDto, FlowBreakdownRow, FlowEntityCard, FlowTalkerDto } from "@mmapp/contracts/traffic-flow" import { ArrowDownIcon, ArrowUpIcon, GitBranchIcon, GlobeIcon, LayersIcon, UsersIcon } from "lucide-react" import { Badge } from "@/components/reui/badge" import { KpiStatGrid } from "@/components/reui-kit/kpi-stat-grid" @@ -108,14 +108,36 @@ export function FlowEntityCardView({ ) } +type SessionFilter = { + kind: "application" | "category" | "service" | "asn" | "country" | "protocol" | "source" | "destination" | "iface" + value: string + label: string +} + +function talkerMatchesFilter(row: FlowTalkerDto, filter: SessionFilter): boolean { + switch (filter.kind) { + case "application": return row.application === filter.value + case "category": return row.category === filter.value + case "service": return row.service === filter.value + case "asn": return String(row.dstAsn ?? "") === filter.value + case "country": return row.dstCountry === filter.value + case "protocol": return row.protoName === filter.value + case "source": return row.src === filter.value + case "destination": return row.dst === filter.value + case "iface": return row.inIface === filter.value + } +} + function FlowBreakdownGrid({ rows, empty, country, + onPick, }: { rows: FlowBreakdownRow[] empty?: string country?: boolean + onPick?: (row: FlowBreakdownRow) => void }) { const columns = useMemo[]>( () => [ @@ -178,7 +200,12 @@ function FlowBreakdownGrid({ }) return ( - + ) } @@ -206,13 +233,18 @@ export function FlowAnalyticsDetail({ emptyHint?: string }) { const [slice, setSlice] = useState("applications") - const [mapCountry, setMapCountry] = useState(null) + const [sessionFilter, setSessionFilter] = useState(null) const rxNow = analytics ? analytics.bpsNow / 1_000_000 : (card?.rxNow ?? 0) const bytes = analytics?.bytes ?? card?.bytes ?? 0 const sessionRows = (analytics?.conversationsList ?? []).filter((row) => - mapCountry ? row.dstCountry === mapCountry : true, + sessionFilter ? talkerMatchesFilter(row, sessionFilter) : true, ) + function pickBreakdown(kind: SessionFilter["kind"], row: FlowBreakdownRow) { + setSessionFilter({ kind, value: row.id, label: row.label }) + setSlice("sessions") + } + if (!card) { return (

@@ -375,47 +407,47 @@ export function FlowAnalyticsDetail({ Интерфейсы - + pickBreakdown("application", row)} /> - + pickBreakdown("category", row)} /> - + pickBreakdown("service", row)} /> - + pickBreakdown("asn", row)} /> - + pickBreakdown("country", row)} /> { - setMapCountry(iso) + setSessionFilter({ kind: "country", value: iso, label: iso }) setSlice("sessions") }} /> - + pickBreakdown("protocol", row)} /> - + pickBreakdown("source", row)} /> - + pickBreakdown("destination", row)} /> - {mapCountry ? ( + {sessionFilter ? (

- фильтр страны {mapCountry} + Фильтр: {sessionFilter.label} @@ -423,13 +455,14 @@ export function FlowAnalyticsDetail({ ) : null} pickBreakdown("iface", row)} /> diff --git a/components/traffic/flow-traffic-map.tsx b/components/traffic/flow-traffic-map.tsx index f7e9193..aefc578 100644 --- a/components/traffic/flow-traffic-map.tsx +++ b/components/traffic/flow-traffic-map.tsx @@ -72,7 +72,9 @@ function FlowTrafficMap({ header: () => Назначение, cell: ({ row }) => ( - + {/^[a-z]{2}$/i.test(row.original.toCountry) + ? + : null} {row.original.toCountry} {row.original.toAsn ? AS{row.original.toAsn} : null} @@ -176,7 +178,12 @@ function FlowTrafficMap({ ) : null} - + onSelectCountry?.(row.toCountry)} + />
) }