diff --git a/app/(main)/traffic/page.tsx b/app/(main)/traffic/page.tsx index 6d5d37a..c51e16e 100644 --- a/app/(main)/traffic/page.tsx +++ b/app/(main)/traffic/page.tsx @@ -17,13 +17,13 @@ import { cn } from "@/lib/utils" import { fmtGB, fmtRate } from "@/lib/fmt-rate" import { useDataSource } from "@/lib/data-source" import { useTrafficLive } from "@/hooks/use-traffic-live" +import { useFlowLive } from "@/hooks/use-flow-live" import { requestJson } from "@/shared/api/http-client" -import { getTrafficFlows } from "@/shared/api/traffic-flow" +import { getFlowAnalytics, getFlowClients, getFlowExporters, getTrafficFlows } from "@/shared/api/traffic-flow" import { listServers } from "@/shared/api/servers" -import { TrafficFlowsDataGrid } from "@/components/data-grids/traffic-flows-data-grid" import { FlowOverlaySheet } from "@/components/traffic/flow-overlay-sheet" -import { DataPageCard } from "@/components/data-page-card" -import type { FlowStatsDto } from "@mmapp/contracts/traffic-flow" +import { FlowAnalyticsDetail, FlowEntityCardView } from "@/components/traffic/flow-analytics-panel" +import type { FlowAnalyticsDto, FlowEntityCard, FlowStatsDto } from "@mmapp/contracts/traffic-flow" import type { ServerRead } from "@mmapp/contracts/servers" import { Badge } from "@/components/reui/badge" import { @@ -309,6 +309,7 @@ const TRAFFIC_RANGE_LABELS: Record = { } type GroupMode = "servers" | "users" | "ifaces" | "flows" +type FlowScope = "servers" | "users" type SortField = "rx" | "tx" | "name" | "sessions" type SortDir = "asc" | "desc" @@ -740,7 +741,7 @@ const SORT_FIELDS: Array<{ field: SortField; label: string; modesOnly?: GroupMod { field: "rx", label: "RX" }, { field: "tx", label: "TX" }, { field: "name", label: "Имя" }, - { field: "sessions", label: "Сессий", modesOnly: ["servers", "users"] }, + { field: "sessions", label: "Сессий", modesOnly: ["servers", "users", "flows"] }, ] export default function TrafficPage() { @@ -765,6 +766,11 @@ export default function TrafficPage() { const [liveUsers, setLiveUsers] = useState([]) const [liveBoundIfaces, setLiveBoundIfaces] = useState([]) const [flowStats, setFlowStats] = useState(null) + const [flowScope, setFlowScope] = useState("servers") + const [flowExporters, setFlowExporters] = useState([]) + const [flowClients, setFlowClients] = useState([]) + const [flowAnalytics, setFlowAnalytics] = useState(null) + const [flowIface, setFlowIface] = useState("__all__") const [overlayOpen, setOverlayOpen] = useState(false) const [catalogServers, setCatalogServers] = useState([]) const effectiveMode: GroupMode = groupMode @@ -774,6 +780,15 @@ export default function TrafficPage() { serverId: selectedId, iface: selectedIface, }) + const flowLiveEnabled = isLive && effectiveMode === "flows" && Boolean(selectedId) + const { sample: flowLiveSample, error: flowLiveError } = useFlowLive({ + enabled: flowLiveEnabled, + backendUrl, + range, + serverId: flowScope === "servers" ? selectedId : undefined, + userId: flowScope === "users" ? selectedId : undefined, + iface: flowIface, + }) const toLiveServer = (s: LiveTrafficServer): ServerTraffic => { return { @@ -862,20 +877,50 @@ export default function TrafficPage() { setLiveBusy(true) setLiveError(null) try { - const stats = await getTrafficFlows(backendUrl, range) + const [stats, exporters, clients] = await Promise.all([ + getTrafficFlows(backendUrl, range), + getFlowExporters(backendUrl, range), + getFlowClients(backendUrl, range), + ]) setFlowStats(stats) + setFlowExporters(exporters.exporters) + setFlowClients(clients.clients) + setSelectedId((prev) => { + const list = flowScope === "users" ? clients.clients : exporters.exporters + if (list.some((x) => x.id === prev)) return prev + return list[0]?.id ?? "" + }) } catch (e) { setLiveError(e instanceof Error ? e.message : "Не удалось загрузить потоки") } finally { setLiveBusy(false) } - }, [isLive, backendUrl, range]) + }, [isLive, backendUrl, range, flowScope]) useEffect(() => { if (!isLive || effectiveMode !== "flows") return void loadFlows() + const t = window.setInterval(() => { void loadFlows() }, 5000) + return () => window.clearInterval(t) }, [isLive, effectiveMode, loadFlows]) + useEffect(() => { + if (!isLive || effectiveMode !== "flows" || !selectedId) { + setFlowAnalytics(null) + return + } + void getFlowAnalytics(backendUrl, { + range, + serverId: flowScope === "servers" ? selectedId : undefined, + userId: flowScope === "users" ? selectedId : undefined, + iface: flowIface, + }).then(setFlowAnalytics).catch(() => setFlowAnalytics(null)) + }, [isLive, effectiveMode, selectedId, range, flowScope, flowIface, backendUrl]) + + useEffect(() => { + setFlowIface("__all__") + }, [selectedId, flowScope]) + useEffect(() => { if (!isLive) return void listServers(backendUrl).then(setCatalogServers).catch(() => setCatalogServers([])) @@ -924,6 +969,11 @@ export default function TrafficPage() { if (next === "servers") setSelectedId(activeServerTraffic[0]?.id ?? "srv1") else if (next === "users") setSelectedId((isLive ? liveUsers : userTraffic)[0]?.id ?? "u1") else if (next === "ifaces") setSelectedId((isLive ? liveBoundIfaces : boundIfaces)[0]?.id ?? "") + else if (next === "flows") { + setFlowScope("servers") + setFlowIface("__all__") + setSelectedId(flowExporters[0]?.id ?? "") + } setSortField("rx") setSortDir("desc") setSearch("") @@ -978,6 +1028,22 @@ export default function TrafficPage() { }) }, [sortField, sortDir, q, activeBoundIfaces]) + const flowCards = flowScope === "users" ? flowClients : flowExporters + const sortedFlowCards = useMemo(() => { + return [...flowCards] + .filter((c) => !q || c.name.toLowerCase().includes(q) || c.subtitle.toLowerCase().includes(q) || c.site.toLowerCase().includes(q)) + .sort((a, b) => { + let v = 0 + if (sortField === "rx") v = a.rxNow - b.rxNow + else if (sortField === "tx") v = a.txNow - b.txNow + else if (sortField === "name") v = a.name.localeCompare(b.name) + else if (sortField === "sessions") v = a.sessions - b.sessions + return sortDir === "desc" ? -v : v + }) + }, [flowCards, q, sortField, sortDir]) + const selFlowCard = sortedFlowCards.find((c) => c.id === selectedId) ?? sortedFlowCards[0] ?? null + const displayedFlow = flowLiveSample ?? flowAnalytics + const selServer = useMemo(() => activeServerTraffic.find(s => s.id === selectedId) ?? activeServerTraffic[0], [selectedId, activeServerTraffic]) const detailServer = liveDetailServer?.id === selectedId ? liveDetailServer : selServer const selUser = useMemo(() => activeUserTraffic.find(u => u.id === selectedId) ?? activeUserTraffic[0], [selectedId, activeUserTraffic]) @@ -995,33 +1061,35 @@ export default function TrafficPage() { const visibleSortFields = SORT_FIELDS.filter(s => !s.modesOnly || s.modesOnly.includes(effectiveMode)) const ingestLine = flowIngestLine(flowStats) + const flowError = liveError || flowLiveError const flowKpiItems = [ { id: "exporters", label: "Экспортёры", - value: String(flowStats?.exportersOnline ?? 0), + value: String(flowExporters.length), icon: , iconClassName: "text-info", }, { - id: "bytes", - label: "Байт/мин", - value: flowStats ? fmtRate((flowStats.bytesPerMin * 8) / 1_000_000) : "—", + id: "bps", + label: "Скорость", + value: displayedFlow ? fmtRate(displayedFlow.bpsNow / 1_000_000) : (flowStats ? fmtRate((flowStats.bytesPerMin * 8) / 1_000_000) : "—"), + hint: displayedFlow?.live ? "live" : undefined, icon: , iconClassName: "text-success", }, { id: "src", label: "Уник. src", - value: String(flowStats?.uniqueSrc ?? 0), + value: String(displayedFlow?.uniqueSrc ?? flowStats?.uniqueSrc ?? 0), icon: , iconClassName: "text-muted-foreground", }, { id: "proto", label: "Топ протокол", - value: flowStats?.topProto ?? "—", + value: displayedFlow?.topProto ?? flowStats?.topProto ?? "—", icon: , iconClassName: "text-warning", }, @@ -1107,51 +1175,106 @@ export default function TrafficPage() { {effectiveMode === "flows" ? (
-
-

- IPFIX top-разговоры. Счётчики интерфейсов — в режимах Серверы / Клиенты / Интерфейсы. -

- {ingestLine ? ( -

- {ingestLine} -

- ) : null} +

+ {ingestLine ?? "IPFIX коллектор"} +

+ +
+ {flowError && ( +
+ {flowError}
-
+ )} +
+
- {TRAFFIC_RANGE_KEYS.map((key) => ( + + +
+
+ + setSearch(e.target.value)} + placeholder="Поиск…" + className="w-full pl-8 pr-3 h-8 text-xs bg-muted/50 border border-border rounded-md outline-none focus:border-primary transition-colors" + /> +
+
+ Сортировка: + {visibleSortFields.map(({ field, label }) => ( ))}
- +
+ {sortedFlowCards.map((card) => ( + setSelectedId(card.id)} + /> + ))} + {sortedFlowCards.length === 0 ? ( +

+ {flowEmptyHint(flowStats) ?? "Нет экспортёров IPFIX. Подключите jump-host."} +

+ ) : null} +
+ + + setRange(r as Range)} + selectedIface={flowIface} + onIface={setFlowIface} + liveHint={displayedFlow?.live ? "live" : undefined} + emptyHint={flowEmptyHint(flowStats)} + /> + +
- {liveError && ( -
- {liveError} -
- )} - - - + let hasIfaceUnique = false + for (const idx of flowIndexes) { + if (!idx.name || !idx.unique) continue + const info = sqlite.prepare(`PRAGMA index_info(${JSON.stringify(idx.name)})`).all() as Array<{ name?: string }> + const names = info.map((c) => c.name) + if (names.includes("in_iface") && names.includes("src") && names.includes("dst")) { + hasIfaceUnique = true + } + } + if (!hasIfaceUnique) { + sqlite.exec(`DROP INDEX IF EXISTS idx_flow_buckets_unique`) + sqlite.exec(` + CREATE UNIQUE INDEX IF NOT EXISTS idx_flow_buckets_unique + ON flow_buckets(server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface) + `) + } +} + const certIssueJobCols = sqlite.prepare(`PRAGMA table_info('certificate_issue_jobs')`).all() as Array<{ name?: string }> if (!certIssueJobCols.some((c) => c.name === "source")) { sqlite.exec(`ALTER TABLE certificate_issue_jobs ADD COLUMN source TEXT NOT NULL DEFAULT 'manual'`) diff --git a/backend/src/db/schema.ts b/backend/src/db/schema.ts index 4b5854e..e614bfd 100644 --- a/backend/src/db/schema.ts +++ b/backend/src/db/schema.ts @@ -198,7 +198,7 @@ export const flowBuckets = sqliteTable("flow_buckets", { inIface: text("in_iface").notNull().default(""), }, (t) => [ uniqueIndex("idx_flow_buckets_unique").on( - t.serverId, t.bucketAt, t.src, t.dst, t.proto, t.srcPort, t.dstPort, + t.serverId, t.bucketAt, t.src, t.dst, t.proto, t.srcPort, t.dstPort, t.inIface, ), ]) diff --git a/backend/src/routes/traffic-flow.ts b/backend/src/routes/traffic-flow.ts index c514ec6..661f705 100644 --- a/backend/src/routes/traffic-flow.ts +++ b/backend/src/routes/traffic-flow.ts @@ -1,5 +1,6 @@ import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod" import type { FastifyReply, FastifyRequest } from "fastify" +import { env } from "../config.js" import { trafficFlowOverlayRequestSchema, trafficFlowSettingsPatchSchema, @@ -12,12 +13,19 @@ import { } from "../services/traffic-flow-settings.js" import { getFlowListenerState, - listFlowTalkers, startTrafficFlowListener, + listFlowTalkers, } from "../services/traffic-flow-ingest.js" +import { + buildFlowAnalytics, + listFlowClients, + listFlowExporters, +} from "../services/traffic-flow-analytics.js" import { applyFlowOverlay } from "../services/traffic-flow-overlay.js" import { listTrafficFlowHostFiles } from "../services/traffic-flow-host-files.js" +const LIVE_TICK_MS = 2000 + function rangeToMinutes(range: string | undefined): number { switch ((range ?? "5m").toLowerCase()) { case "5m": return 5 @@ -29,6 +37,22 @@ function rangeToMinutes(range: string | undefined): number { } } +function parseId(raw: unknown): number | undefined { + if (raw == null || raw === "") return undefined + const n = Number.parseInt(String(raw), 10) + return Number.isFinite(n) ? n : undefined +} + +function analyticsQuery(req: FastifyRequest) { + const q = req.query as { range?: string; serverId?: string; userId?: string; iface?: string } + return { + minutes: rangeToMinutes(q.range), + serverId: parseId(q.serverId), + userId: q.userId?.trim() || undefined, + iface: q.iface?.trim() || undefined, + } +} + async function sendFlowTalkers(req: FastifyRequest, reply: FastifyReply) { const q = req.query as { range?: string } return reply.send(listFlowTalkers(rangeToMinutes(q.range))) @@ -58,6 +82,28 @@ async function applyOverlayHandler(req: FastifyRequest, reply: FastifyReply) { } } +function writeSse(raw: NodeJS.WritableStream, event: string, data: unknown) { + raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`) +} + +function sleep(ms: number, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal.aborted) { + reject(new Error("aborted")) + return + } + const timer = setTimeout(() => { + signal.removeEventListener("abort", onAbort) + resolve() + }, ms) + const onAbort = () => { + clearTimeout(timer) + reject(new Error("aborted")) + } + signal.addEventListener("abort", onAbort, { once: true }) + }) +} + const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => { app.get("/traffic/flow/settings", async (_req, reply) => { return reply.send(toTrafficFlowSettingsDto(getFlowListenerState())) @@ -94,6 +140,63 @@ const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => { app.get("/traffic/flow", sendFlowTalkers) app.get("/traffic/flows", sendFlowTalkers) + + app.get("/traffic/flow/exporters", async (req, reply) => { + const q = req.query as { range?: string } + return reply.send(listFlowExporters(rangeToMinutes(q.range))) + }) + + app.get("/traffic/flow/clients", async (req, reply) => { + const q = req.query as { range?: string } + return reply.send(listFlowClients(rangeToMinutes(q.range))) + }) + + app.get("/traffic/flow/analytics", async (req, reply) => { + return reply.send(buildFlowAnalytics(analyticsQuery(req))) + }) + + app.get("/traffic/flow/live", async (req, reply) => { + const query = analyticsQuery(req) + const abort = new AbortController() + const onClose = () => abort.abort() + req.raw.on("close", onClose) + + reply.hijack() + req.raw.setTimeout(0) + reply.raw.setTimeout(0) + const origin = typeof req.headers.origin === "string" ? req.headers.origin : "" + const allowed = env.CORS_ORIGIN + const sseHeaders: Record = { + "Content-Type": "text/event-stream; charset=utf-8", + "Cache-Control": "no-cache, no-transform", + Connection: "keep-alive", + "X-Accel-Buffering": "no", + } + if (origin && (allowed === "*" || allowed === origin)) { + sseHeaders["Access-Control-Allow-Origin"] = origin + sseHeaders["Access-Control-Allow-Credentials"] = "true" + sseHeaders["Access-Control-Allow-Headers"] = "Authorization, Accept" + sseHeaders.Vary = "Origin" + } + reply.raw.writeHead(200, sseHeaders) + reply.raw.write(":\n\n") + + try { + while (!abort.signal.aborted) { + writeSse(reply.raw, "sample", buildFlowAnalytics(query)) + await sleep(LIVE_TICK_MS, abort.signal) + } + } catch { + /* abort / disconnect */ + } finally { + req.raw.off("close", onClose) + try { + reply.raw.end() + } catch { + /* already closed */ + } + } + }) } export default trafficFlowRoutes diff --git a/backend/src/services/traffic-collector.ts b/backend/src/services/traffic-collector.ts index 76f68ce..8a688c2 100644 --- a/backend/src/services/traffic-collector.ts +++ b/backend/src/services/traffic-collector.ts @@ -5,9 +5,12 @@ import type { TrafficRunSnapshot } from "../types/scheduler-run-snapshot.js" import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js" import { MikrotikClient } from "./mikrotik.js" import { bpsToMbps, rateBpsFromDelta, shouldIncludeIface } from "./traffic-rate.js" +import { rememberServerIfaces } from "./traffic-flow-ifindex.js" interface RosIfaceTraffic { + ".id"?: string name?: string + ifindex?: string running?: string disabled?: string "rx-byte"?: string @@ -139,6 +142,7 @@ export async function collectTrafficOnce(): Promise { try { const client = MikrotikClient.fromServer(srv) const ifaces = await client.get("/interface") + rememberServerIfaces(srv.id, ifaces) const prevWave = readPreviousWave(srv.id) const nowMs = Date.parse(now) let sumRxMbps = 0 diff --git a/backend/src/services/traffic-flow-analytics.test.ts b/backend/src/services/traffic-flow-analytics.test.ts new file mode 100644 index 0000000..c2e2804 --- /dev/null +++ b/backend/src/services/traffic-flow-analytics.test.ts @@ -0,0 +1,68 @@ +import assert from "node:assert/strict" +import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js" +import { + ingestParsedFlowsForServerForTests, + resetFlowRingsForTests, +} from "./traffic-flow-ingest.js" +import { buildFlowAnalytics } from "./traffic-flow-analytics.js" + +resetIfaceCacheForTests() +resetFlowRingsForTests() +rememberServerIfaces(7, [ + { ".id": "*2", name: "ether1" }, + { ".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: "10", + }, + { + 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 = 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 = buildFlowAnalytics({ minutes: 5, serverId: 7, iface: "ether1" }) + assert.ok(filtered.bytes >= 12_000) + assert.equal(filtered.ifaces[0]?.name, "ether1") + + const miss = buildFlowAnalytics({ minutes: 5, serverId: 7, iface: "wg-flow" }) + assert.equal(miss.conversations, 0) + + const other = buildFlowAnalytics({ minutes: 5, serverId: 99 }) + assert.equal(other.conversations, 0) +} finally { + resetFlowRingsForTests() + resetIfaceCacheForTests() +} + +console.log("traffic-flow-analytics.test.ts: ok") diff --git a/backend/src/services/traffic-flow-analytics.ts b/backend/src/services/traffic-flow-analytics.ts new file mode 100644 index 0000000..1bc72d6 --- /dev/null +++ b/backend/src/services/traffic-flow-analytics.ts @@ -0,0 +1,303 @@ +import { eq } from "drizzle-orm" +import { db } from "../db/index.js" +import { appUsers, servers, userInterfaceBindings } from "../db/schema.js" +import type { + FlowAnalyticsDto, + FlowBreakdownRow, + FlowClientsDto, + FlowEntityCard, + FlowExportersDto, + FlowTalkerDto, +} from "@mmapp/contracts/traffic-flow" +import { protoName } from "./traffic-flow-parse.js" +import { + getFlowListenerState, + getRingMbps, + listStoredFlowRows, + type PendingFlowRow, +} from "./traffic-flow-ingest.js" +import { resolveIfaceName } from "./traffic-flow-ifaces.js" +import { getTrafficFlowSettingsRow, listHostPeers } from "./traffic-flow-settings.js" +import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js" + +export interface FlowAnalyticsQuery { + minutes: number + serverId?: number + userId?: string + iface?: string +} + +function bpsToMbps(bps: number): number { + return bps / 1_000_000 +} + +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, + bytes: v.bytes, + packets: v.packets, + bps: (v.bytes * 8) / windowSec, + percent: (v.bytes / total) * 100, + })) +} + +function bump(map: Map, id: string, bytes: number, packets: number) { + const prev = map.get(id) ?? { bytes: 0, packets: 0 } + prev.bytes += bytes + prev.packets += packets + map.set(id, prev) +} + +function userIfaceAllow(userId: string): Map> | null { + if (!userId) return null + const binds = db.select().from(userInterfaceBindings).where(eq(userInterfaceBindings.userId, userId)).all() + const allow = new Map>() + for (const b of binds) { + const set = allow.get(b.serverId) ?? new Set() + set.add(b.interfaceName) + allow.set(b.serverId, set) + } + return allow +} + +function seriesFromRows(rows: PendingFlowRow[], minutes: number): { rx: number[]; tx: number[] } { + const slots = Math.min(60, Math.max(5, minutes)) + const slotMs = (minutes * 60_000) / slots + const start = Date.now() - minutes * 60_000 + const rx = Array(slots).fill(0) as number[] + const tx = Array(slots).fill(0) as number[] + for (const r of rows) { + const t = Date.parse(r.bucketAt) + if (!Number.isFinite(t)) continue + const idx = Math.min(slots - 1, Math.max(0, Math.floor((t - start) / slotMs))) + rx[idx] += r.bytes + } + const slotSec = Math.max(1, slotMs / 1000) + return { + rx: rx.map((b) => bpsToMbps((b * 8) / slotSec)), + tx, + } +} + +function snapshotStatus(serverId: number): FlowEntityCard["status"] { + void serverId + return "online" +} + +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 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])) + + const applications = new Map() + const protocols = new Map() + const sources = new Map() + const destinations = new Map() + const ifacesMap = new Map() + const conv = new Map() + const srcs = new Set() + const dsts = new Set() + let totalBytes = 0 + let totalPackets = 0 + const matched: PendingFlowRow[] = [] + + for (const r of raw) { + const resolved = resolveIfaceName(r.serverId, r.inIface) + if (!flowRowMatchesFilter(r, resolved.name, q, allow)) continue + matched.push(r) + totalBytes += r.bytes + totalPackets += r.packets + srcs.add(r.src) + dsts.add(r.dst) + const app = applicationName(r.proto, r.dstPort, r.srcPort) + bump(applications, app, r.bytes, r.packets) + bump(protocols, protoName(r.proto), r.bytes, r.packets) + bump(sources, r.src, r.bytes, r.packets) + bump(destinations, r.dst, r.bytes, r.packets) + const ifaceKey = resolved.name + const prevIf = ifacesMap.get(ifaceKey) ?? { bytes: 0, packets: 0, index: resolved.index } + prevIf.bytes += r.bytes + prevIf.packets += r.packets + ifacesMap.set(ifaceKey, prevIf) + + const ckey = `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}|${r.inIface}` + const prev = conv.get(ckey) + if (prev) { + prev.rawBytes += r.bytes + prev.bytes += r.bytes + prev.packets += r.packets + } else { + conv.set(ckey, { + serverId: String(r.serverId), + serverName: nameById.get(r.serverId) ?? String(r.serverId), + src: r.src, + dst: r.dst, + proto: r.proto, + protoName: protoName(r.proto), + srcPort: r.srcPort, + dstPort: r.dstPort, + bytes: r.bytes, + packets: r.packets, + bps: 0, + inIface: resolved.name, + inIfaceIndex: resolved.index, + application: app, + rawBytes: r.bytes, + }) + } + } + + const conversationsList = [...conv.values()] + .map((t) => ({ ...t, bps: (t.rawBytes * 8) / windowSec })) + .sort((a, b) => b.bytes - a.bytes) + .slice(0, top) + .map(({ rawBytes: _raw, ...rest }) => rest) + + let topProto = "—" + let topProtoBytes = 0 + for (const [label, v] of protocols) { + if (v.bytes > topProtoBytes) { + topProtoBytes = v.bytes + topProto = label + } + } + + const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : "__all__" + const ringServer = q.serverId ?? (matched[0]?.serverId ?? 0) + const ring = ringServer + ? getRingMbps(ringServer, ifaceFilter === "__all__" ? "__all__" : (ifacesMap.get(ifaceFilter)?.index || ifaceFilter)) + : { rx: Array(60).fill(0) as number[], tx: Array(60).fill(0) as number[], rxNow: 0, txNow: 0 } + + const fromBuckets = seriesFromRows(matched, q.minutes) + const rxSeries = q.minutes <= 15 ? ring.rx : fromBuckets.rx + const txSeries = q.minutes <= 15 ? ring.tx : fromBuckets.tx + + const ifaceRows = [...ifacesMap.entries()] + .sort((a, b) => b[1].bytes - a[1].bytes) + .map(([name, v]) => ({ + name, + index: v.index, + bps: (v.bytes * 8) / windowSec, + })) + + const protoBreakdown = topN(protocols, windowSec, top) + const listener = getFlowListenerState() + + return { + bpsNow: (ring.rxNow + ring.txNow) * 1_000_000 || (totalBytes * 8) / windowSec, + bytes: totalBytes, + packets: totalPackets, + conversations: conv.size, + uniqueSrc: srcs.size, + uniqueDst: dsts.size, + topProto, + rxSeries, + txSeries, + applications: topN(applications, windowSec, top), + protocols: protoBreakdown, + sources: topN(sources, windowSec, top), + destinations: topN(destinations, windowSec, top), + interfaces: [...ifacesMap.entries()].map(([label, v]) => ({ + id: label, + label, + bytes: v.bytes, + packets: v.packets, + bps: (v.bytes * 8) / windowSec, + percent: totalBytes > 0 ? (v.bytes / totalBytes) * 100 : 0, + })).sort((a, b) => b.bytes - a.bytes), + conversationsList, + ifaces: ifaceRows, + live: listener.bound, + } +} + +function cardFromServer( + s: typeof servers.$inferSelect, + minutes: number, +): FlowEntityCard { + const analytics = buildFlowAnalytics({ minutes, serverId: s.id }) + const ring = getRingMbps(s.id, "__all__") + return { + id: String(s.id), + name: s.name || s.host, + subtitle: s.host, + site: s.site || "—", + country: s.country || "UN", + status: snapshotStatus(s.id), + rxNow: ring.rxNow || bpsToMbps(analytics.bpsNow), + txNow: ring.txNow, + sessions: analytics.conversations, + rxSeries: ring.rx.some((v) => v > 0) ? ring.rx : analytics.rxSeries, + txSeries: ring.tx, + bytes: analytics.bytes, + } +} + +export function listFlowExporters(minutes: number): FlowExportersDto { + const settings = getTrafficFlowSettingsRow() + const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString() + const rows = listStoredFlowRows(sinceIso) + const ids = new Set() + for (const r of rows) ids.add(r.serverId) + for (const p of listHostPeers()) ids.add(p.serverId) + const serverRows = db.select().from(servers).all() + const exporters = serverRows + .filter((s) => ids.has(s.id)) + .map((s) => cardFromServer(s, minutes)) + .sort((a, b) => b.rxNow - a.rxNow) + const listener = getFlowListenerState() + return { + exporters, + lastExporterIp: settings.lastExporterIp ?? null, + lastError: settings.lastError || null, + packetsReceived: settings.packetsReceived, + lastDatagramAt: settings.lastDatagramAt ?? null, + listenerBound: listener.bound, + listenerAddress: listener.address, + } +} + +export function listFlowClients(minutes: number): FlowClientsDto { + const users = db.select().from(appUsers).all() + const binds = db.select().from(userInterfaceBindings).all() + const byUser = new Map() + for (const b of binds) { + const list = byUser.get(b.userId) ?? [] + list.push(b) + byUser.set(b.userId, list) + } + const clients: FlowEntityCard[] = [] + for (const u of users) { + const userBinds = byUser.get(u.id) ?? [] + if (userBinds.length === 0) continue + const analytics = buildFlowAnalytics({ minutes, userId: u.id }) + const firstServer = userBinds[0]?.serverId + const ring = firstServer ? getRingMbps(firstServer, "__all__") : { rx: Array(60).fill(0) as number[], tx: Array(60).fill(0) as number[], rxNow: 0, txNow: 0 } + clients.push({ + id: u.id, + name: u.login, + subtitle: u.name || u.login, + site: `${userBinds.length} ifaces`, + country: "UN", + status: u.active ? "online" : "offline", + rxNow: bpsToMbps(analytics.bpsNow) || ring.rxNow, + txNow: ring.txNow, + sessions: analytics.conversations, + rxSeries: analytics.rxSeries, + txSeries: analytics.txSeries, + bytes: analytics.bytes, + }) + } + clients.sort((a, b) => b.rxNow - a.rxNow) + return { clients } +} diff --git a/backend/src/services/traffic-flow-apps.ts b/backend/src/services/traffic-flow-apps.ts new file mode 100644 index 0000000..5694fe9 --- /dev/null +++ b/backend/src/services/traffic-flow-apps.ts @@ -0,0 +1,79 @@ +import { protoName } from "./traffic-flow-parse.js" + +const WELL_KNOWN: Record = { + "6:80": "HTTP", + "6:443": "HTTPS", + "6:8080": "HTTP-alt", + "6:8443": "HTTPS-alt", + "6:22": "SSH", + "6:21": "FTP", + "6:25": "SMTP", + "6:110": "POP3", + "6:143": "IMAP", + "6:993": "IMAPS", + "6:995": "POP3S", + "6:587": "SMTP", + "6:465": "SMTPS", + "6:3306": "MySQL", + "6:5432": "PostgreSQL", + "6:6379": "Redis", + "6:3389": "RDP", + "6:445": "SMB", + "6:139": "NetBIOS", + "6:179": "BGP", + "6:8291": "WinBox", + "6:8728": "ROS-API", + "6:8729": "ROS-API-SSL", + "17:53": "DNS", + "6:53": "DNS", + "17:123": "NTP", + "17:161": "SNMP", + "17:162": "SNMP-trap", + "17:500": "IKE", + "17:4500": "NAT-T", + "17:1194": "OpenVPN", + "17:51820": "WireGuard", + "17:4789": "VXLAN", + "17:4739": "IPFIX", + "17:2055": "NetFlow", + "17:67": "DHCP", + "17:68": "DHCP", + "17:69": "TFTP", + "17:1812": "RADIUS", + "1:0": "ICMP", + "47:0": "GRE", + "50:0": "ESP", + "89:0": "OSPF", +} + +export function applicationName(proto: number, dstPort: number, srcPort = 0): string { + if (proto === 1) return "ICMP" + if (proto === 47) return "GRE" + if (proto === 50) return "ESP" + if (proto === 89) return "OSPF" + const dstKey = `${proto}:${dstPort}` + const srcKey = `${proto}:${srcPort}` + return WELL_KNOWN[dstKey] ?? WELL_KNOWN[srcKey] ?? `${protoName(proto)}/${dstPort || srcPort || "—"}` +} + +export interface FlowMatchQuery { + serverId?: number + userId?: string + iface?: string +} + +export function flowRowMatchesFilter( + row: { serverId: number; inIface: string }, + resolvedName: string, + q: FlowMatchQuery, + allow: Map> | null, +): boolean { + if (q.serverId != null && row.serverId !== q.serverId) return false + if (allow) { + const names = allow.get(row.serverId) + if (!names || !names.has(resolvedName)) return false + } + const iface = q.iface && q.iface !== "__all__" ? q.iface : "" + if (iface && resolvedName !== iface && row.inIface !== iface) return false + return true +} diff --git a/backend/src/services/traffic-flow-ifaces.test.ts b/backend/src/services/traffic-flow-ifaces.test.ts new file mode 100644 index 0000000..80d9cff --- /dev/null +++ b/backend/src/services/traffic-flow-ifaces.test.ts @@ -0,0 +1,43 @@ +import assert from "node:assert/strict" +import { + rememberServerIfaces, + resetIfaceCacheForTests, + resolveIfaceName, + rosIdToIfIndex, +} from "./traffic-flow-ifindex.js" +import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js" + +assert.equal(rosIdToIfIndex("*A"), 10) +assert.equal(rosIdToIfIndex("*D"), 13) +assert.equal(rosIdToIfIndex("*2"), 2) +assert.equal(rosIdToIfIndex("*9"), 9) +assert.equal(rosIdToIfIndex("0"), 0) +assert.equal(rosIdToIfIndex(""), null) + +resetIfaceCacheForTests() +rememberServerIfaces(7, [ + { ".id": "*2", name: "ether1" }, + { ".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, "13").name, "bridge") +assert.equal(resolveIfaceName(7, "0").name, "—") +assert.equal(resolveIfaceName(7, "ether1").name, "ether1") +assert.equal(resolveIfaceName(7, "99").name, "#99") + +assert.equal(applicationName(6, 443), "HTTPS") +assert.equal(applicationName(17, 53), "DNS") +assert.equal(applicationName(6, 22), "SSH") +assert.equal(applicationName(17, 51820), "WireGuard") +assert.equal(applicationName(6, 179), "BGP") + +const allow = new Map>([[7, new Set(["ether1", "wg-flow"])]]) +assert.equal(flowRowMatchesFilter({ serverId: 7, inIface: "2" }, "ether1", {}, allow), true) +assert.equal(flowRowMatchesFilter({ serverId: 7, inIface: "2" }, "bridge", {}, allow), false) +assert.equal(flowRowMatchesFilter({ serverId: 7, inIface: "2" }, "ether1", { iface: "ether1" }, allow), true) +assert.equal(flowRowMatchesFilter({ serverId: 7, inIface: "2" }, "ether1", { iface: "wg-flow" }, allow), false) +assert.equal(flowRowMatchesFilter({ serverId: 8, inIface: "2" }, "ether1", { serverId: 7 }, null), false) + +console.log("traffic-flow-ifaces.test.ts: ok") diff --git a/backend/src/services/traffic-flow-ifaces.ts b/backend/src/services/traffic-flow-ifaces.ts new file mode 100644 index 0000000..44c401b --- /dev/null +++ b/backend/src/services/traffic-flow-ifaces.ts @@ -0,0 +1,36 @@ +import { eq } from "drizzle-orm" +import { db } from "../db/index.js" +import { servers } from "../db/schema.js" +import { MikrotikClient } from "./mikrotik.js" +import { + ifaceCacheFresh, + rememberServerIfaces, + type RosIfaceIndexRow, +} from "./traffic-flow-ifindex.js" + +export { + ifaceCacheHas, + rememberServerIfaces, + resetIfaceCacheForTests, + resolveIfaceName, + rosIdToIfIndex, +} 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 + inflight.add(serverId) + try { + const row = db.select().from(servers).where(eq(servers.id, serverId)).limit(1).all()[0] + if (!row) return + const client = MikrotikClient.fromServer(row) + const ifaces = await client.get("/interface") + rememberServerIfaces(serverId, Array.isArray(ifaces) ? ifaces : []) + } catch { + /* keep previous cache */ + } finally { + inflight.delete(serverId) + } +} diff --git a/backend/src/services/traffic-flow-ifindex.ts b/backend/src/services/traffic-flow-ifindex.ts new file mode 100644 index 0000000..0c00bc4 --- /dev/null +++ b/backend/src/services/traffic-flow-ifindex.ts @@ -0,0 +1,59 @@ +export interface RosIfaceIndexRow { + ".id"?: string + name?: string + ifindex?: string +} + +const cache = new Map>() +const fetchedAt = new Map() + +export const IFACE_CACHE_TTL_MS = 60_000 + +/** RouterOS `.id` (`*A`) → SNMP ifIndex (10). */ +export function rosIdToIfIndex(id: string | undefined | null): number | null { + if (!id) return null + const raw = String(id).trim() + const hex = raw.startsWith("*") ? raw.slice(1) : raw + if (!hex || !/^[0-9a-fA-F]+$/.test(hex)) return null + const n = parseInt(hex, 16) + return Number.isFinite(n) ? n : null +} + +export function rememberServerIfaces(serverId: number, rows: RosIfaceIndexRow[]): void { + const map = new Map() + for (const row of rows) { + const name = String(row.name ?? "").trim() + if (!name) continue + const fromProp = Number.parseInt(String(row.ifindex ?? ""), 10) + const idx = Number.isFinite(fromProp) && fromProp > 0 + ? fromProp + : rosIdToIfIndex(row[".id"]) + if (idx != null && idx > 0) map.set(idx, name) + } + cache.set(serverId, map) + fetchedAt.set(serverId, Date.now()) +} + +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 } +} + +export function ifaceCacheHas(serverId: number): boolean { + return cache.has(serverId) +} + +export function ifaceCacheFresh(serverId: number, ttlMs = IFACE_CACHE_TTL_MS): boolean { + const prev = fetchedAt.get(serverId) ?? 0 + return Boolean(prev && Date.now() - prev < ttlMs && cache.has(serverId)) +} + +export function resetIfaceCacheForTests(): void { + cache.clear() + fetchedAt.clear() +} diff --git a/backend/src/services/traffic-flow-ingest.ts b/backend/src/services/traffic-flow-ingest.ts index 40f8e62..3648709 100644 --- a/backend/src/services/traffic-flow-ingest.ts +++ b/backend/src/services/traffic-flow-ingest.ts @@ -11,12 +11,31 @@ import { recordFlowListenerError, recordFlowPacket, } from "./traffic-flow-settings.js" +import { ifaceCacheHas, refreshServerIfaces, resolveIfaceName } from "./traffic-flow-ifaces.js" +import { applicationName } from "./traffic-flow-apps.js" export interface FlowListenerState { bound: boolean address: string | null } +export interface PendingFlowRow { + serverId: number + bucketAt: string + src: string + dst: string + proto: number + srcPort: number + dstPort: number + bytes: number + packets: number + inIface: string + outIface: string +} + +const TICK_MS = 2_000 +const RING_LEN = 60 + let socket: Socket | null = null let state: FlowListenerState = { bound: false, address: null } const pending = new Map() let flushTimer: ReturnType | null = null +const tickAccum = new Map() +const rings = new Map() + export function getFlowListenerState(): FlowListenerState { return state } @@ -38,6 +60,66 @@ function minuteBucketIso(at = Date.now()): string { return d.toISOString() } +function ringKey(serverId: number, iface: string): string { + return `${serverId}\0${iface || "__all__"}` +} + +function bumpTick(key: string, inBytes: number, outBytes: number): void { + const prev = tickAccum.get(key) ?? { inBytes: 0, outBytes: 0 } + prev.inBytes += inBytes + prev.outBytes += outBytes + tickAccum.set(key, prev) +} + +function addToTick(serverId: number, inIface: string, outIface: string, bytes: number): void { + bumpTick(ringKey(serverId, "__all__"), bytes, 0) + if (inIface) bumpTick(ringKey(serverId, inIface), bytes, 0) + if (outIface && outIface !== inIface) bumpTick(ringKey(serverId, outIface), 0, bytes) +} + +function emptyRing(): { inBps: number[]; outBps: number[] } { + return { inBps: Array(RING_LEN).fill(0), outBps: Array(RING_LEN).fill(0) } +} + +export function rollFlowRings(): void { + const keys = new Set([...tickAccum.keys(), ...rings.keys()]) + const sec = TICK_MS / 1000 + for (const key of keys) { + const acc = tickAccum.get(key) ?? { inBytes: 0, outBytes: 0 } + tickAccum.delete(key) + const inBps = (acc.inBytes * 8) / sec + const outBps = (acc.outBytes * 8) / sec + let ring = rings.get(key) + if (!ring) { + ring = emptyRing() + rings.set(key, ring) + } + ring.inBps.push(inBps) + ring.inBps.shift() + ring.outBps.push(outBps) + ring.outBps.shift() + } +} + +export function getRingMbps(serverId: number, iface = "__all__"): { + rx: number[] + tx: number[] + rxNow: number + txNow: number +} { + const ring = rings.get(ringKey(serverId, iface)) + const scale = 1_000_000 + if (!ring) { + return { rx: Array(RING_LEN).fill(0), tx: Array(RING_LEN).fill(0), rxNow: 0, txNow: 0 } + } + return { + rx: ring.inBps.map((b) => b / scale), + tx: ring.outBps.map((b) => b / scale), + rxNow: (ring.inBps[RING_LEN - 1] ?? 0) / scale, + txNow: (ring.outBps[RING_LEN - 1] ?? 0) / scale, + } +} + function resolveServerId(exporterIp: string): number | null { const settings = getTrafficFlowSettingsRow() const rows = db.select({ @@ -63,9 +145,11 @@ function resolveServerId(exporterIp: string): number | null { function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean { const serverId = resolveServerId(exporterIp) if (serverId == null) return false + if (!ifaceCacheHas(serverId)) void refreshServerIfaces(serverId) const bucketAt = minuteBucketIso() for (const flow of flows) { - const key = `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}` + 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 prev = pending.get(key) if (prev) { prev.bytes += flow.bytes @@ -83,6 +167,22 @@ function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean { return true } +export function peekPendingFlows(): PendingFlowRow[] { + return [...pending.values()].map((row) => ({ + 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, + outIface: row.flow.outIface, + })) +} + function flushPending() { if (pending.size === 0) return const settings = getTrafficFlowSettingsRow() @@ -113,6 +213,7 @@ function flushPending() { flowBuckets.proto, flowBuckets.srcPort, flowBuckets.dstPort, + flowBuckets.inIface, ], set: { bytes: sql`${flowBuckets.bytes} + excluded.bytes`, @@ -145,6 +246,11 @@ function flushPending() { } } +function onTick() { + rollFlowRings() + flushPending() +} + function onMessage(msg: Buffer, rinfo: { address: string }) { try { const flows = parseFlowPacket(msg, rinfo.address) @@ -195,13 +301,46 @@ export function startTrafficFlowListener() { recordFlowListenerError("") }) socket = sock - flushTimer = setInterval(flushPending, 15_000) + flushTimer = setInterval(onTick, TICK_MS) +} + +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, { + 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, + outIface: "", + }) + } + 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 }) + } + } + return [...merged.values()] } export function listFlowTalkers(minutes = 5): FlowStatsDto { const settings = getTrafficFlowSettingsRow() const rangeStart = new Date(Date.now() - minutes * 60_000).toISOString() - const rows = db.select().from(flowBuckets).where(gte(flowBuckets.bucketAt, rangeStart)).all() + const rows = listStoredFlowRows(rangeStart) const serverRows = db.select().from(servers).all() const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host])) const agg = new Map() @@ -211,7 +350,8 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto { const exporters = new Set() let totalBytes = 0 for (const r of rows) { - const key = `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}` + const resolved = resolveIfaceName(r.serverId, r.inIface) + const key = `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}|${r.inIface}` const prev = agg.get(key) const bytes = r.bytes totalBytes += bytes @@ -236,7 +376,9 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto { bytes, packets: r.packets, bps: 0, - inIface: r.inIface, + inIface: resolved.name, + inIfaceIndex: resolved.index, + application: applicationName(r.proto, r.dstPort, r.srcPort), rawBytes: bytes, }) } @@ -273,5 +415,35 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto { export function ingestParsedFlowsForTests(exporterIp: string, flows: ParsedFlow[]) { queueFlows(exporterIp, flows) + rollFlowRings() flushPending() } + +/** Кладёт потоки в pending без flush в SQLite — для юнит-тестов аналитики. */ +export function ingestParsedFlowsForServerForTests(serverId: number, flows: ParsedFlow[]) { + 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 prev = pending.get(key) + if (prev) { + prev.bytes += flow.bytes + prev.packets += flow.packets + } else { + pending.set(key, { + serverId, + bucketAt, + flow: { ...flow }, + bytes: flow.bytes, + packets: flow.packets, + }) + } + } + rollFlowRings() +} + +export function resetFlowRingsForTests() { + tickAccum.clear() + rings.clear() + pending.clear() +} diff --git a/backend/src/services/traffic-flow-parse.test.ts b/backend/src/services/traffic-flow-parse.test.ts index 1cfc30d..7578b14 100644 --- a/backend/src/services/traffic-flow-parse.test.ts +++ b/backend/src/services/traffic-flow-parse.test.ts @@ -24,6 +24,7 @@ assert.equal(flows[0]?.src, "10.1.1.8") assert.equal(flows[0]?.dst, "8.8.8.8") assert.equal(flows[0]?.proto, 6) assert.equal(flows[0]?.bytes, 1500) +assert.equal(flows[0]?.inIface, "1") assert.equal(protoName(6), "TCP") assert.equal(parseFlowPacket(Buffer.from([0, 1]), "1.1.1.1").length, 0) @@ -66,4 +67,37 @@ resetFlowTemplatesForTests() assert.equal(fromData[0]?.dst, "8.8.8.8") } +resetFlowTemplatesForTests() +{ + const tpl = Buffer.alloc(16 + 24) + tpl.writeUInt16BE(10, 0) + tpl.writeUInt16BE(tpl.length, 2) + tpl.writeUInt16BE(2, 16) + tpl.writeUInt16BE(24, 18) + tpl.writeUInt16BE(256, 20) + tpl.writeUInt16BE(4, 22) + tpl.writeUInt16BE(8, 24) + tpl.writeUInt16BE(4, 26) + tpl.writeUInt16BE(12, 28) + tpl.writeUInt16BE(4, 30) + tpl.writeUInt16BE(10, 32) + tpl.writeUInt16BE(4, 34) + tpl.writeUInt16BE(82, 36) + tpl.writeUInt16BE(6, 38) + const data = Buffer.alloc(16 + 22) + data.writeUInt16BE(10, 0) + data.writeUInt16BE(data.length, 2) + data.writeUInt16BE(256, 16) + data.writeUInt16BE(22, 18) + data[20] = 10; data[21] = 1; data[22] = 1; data[23] = 8 + data[24] = 8; data[25] = 8; data[26] = 8; data[27] = 8 + data.writeUInt32BE(13, 28) + data.write("ether1", 32) + parseFlowPacket(tpl, "10.255.254.3") + const named = parseFlowPacket(data, "10.255.254.3") + assert.equal(named.length, 1) + assert.equal(named[0]?.inIface, "13") + assert.equal(named[0]?.src, "10.1.1.8") +} + console.log("traffic-flow-parse.test.ts: ok") diff --git a/backend/src/services/traffic-flow-parse.ts b/backend/src/services/traffic-flow-parse.ts index 96ac7c2..98c3822 100644 --- a/backend/src/services/traffic-flow-parse.ts +++ b/backend/src/services/traffic-flow-parse.ts @@ -7,6 +7,7 @@ export interface ParsedFlow { bytes: number packets: number inIface: string + outIface: string } interface FieldSpec { @@ -95,6 +96,7 @@ function parseNetflowV5(buf: Buffer): ParsedFlow[] { dstPort: buf.readUInt16BE(off + 34), proto: buf.readUInt8(off + 38), inIface: String(buf.readUInt16BE(off + 12)), + outIface: String(buf.readUInt16BE(off + 14)), }) off += 48 } @@ -144,6 +146,8 @@ function recordFromFields( let bytes = 0 let packets = 0 let inIface = "" + let outIface = "" + let ifaceName = "" for (const f of fields) { const field = consumeField(buf, off, f.length, limit) if (!field) return null @@ -191,12 +195,19 @@ function recordFromFields( case 10: inIface = String(readUint(data, 0, data.length)) break + case 14: + outIface = String(readUint(data, 0, data.length)) + break + case 82: + ifaceName = data.toString("utf8").replace(/\0/g, "").trim() + break default: break } off = field.next } - return { flow: { src, dst, proto, srcPort, dstPort, bytes, packets, inIface }, next: off } + if (!inIface && ifaceName) inIface = ifaceName + return { flow: { src, dst, proto, srcPort, dstPort, bytes, packets, inIface, outIface }, next: off } } function parseDataRecords( diff --git a/components/data-grids/traffic-flows-data-grid.tsx b/components/data-grids/traffic-flows-data-grid.tsx index e3c85f8..60b3327 100644 --- a/components/data-grids/traffic-flows-data-grid.tsx +++ b/components/data-grids/traffic-flows-data-grid.tsx @@ -59,6 +59,15 @@ function TrafficFlowsDataGrid({ ), meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD }, }, + { + id: "app", + accessorFn: (r) => r.application ?? r.protoName, + header: () => App, + cell: ({ row }) => ( + {row.original.application ?? row.original.protoName} + ), + meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD }, + }, { id: "proto", accessorKey: "protoName", diff --git a/components/traffic/flow-analytics-panel.tsx b/components/traffic/flow-analytics-panel.tsx new file mode 100644 index 0000000..7d459d0 --- /dev/null +++ b/components/traffic/flow-analytics-panel.tsx @@ -0,0 +1,369 @@ +"use client" + +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 { ArrowDownIcon, ArrowUpIcon, GitBranchIcon } from "lucide-react" +import { Badge } from "@/components/reui/badge" +import { KpiStatGrid } from "@/components/reui-kit/kpi-stat-grid" +import { TrafficRxTxChart } from "@/components/reui-kit/traffic-rx-tx-chart" +import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs" +import { DataGridShell } from "@/components/data-grids/shared/data-grid-shell" +import { + DATA_GRID_CELL_PAD, + DATA_GRID_CELL_PAD_FIRST, + DATA_GRID_CELL_PAD_LAST, +} from "@/components/data-grids/shared/data-grid-layout" +import { TrafficFlowsDataGrid } from "@/components/data-grids/traffic-flows-data-grid" +import { Sparkline } from "@/components/sparkline" +import { StatusDot } from "@/components/status-dot" +import { Flag } from "@/components/flag" +import { fmtRate } from "@/lib/fmt-rate" +import { cn } from "@/lib/utils" + +function formatBytes(n: number): string { + if (n >= 1_000_000_000) return `${(n / 1_000_000_000).toFixed(2)} ГБ` + if (n >= 1_000_000) return `${(n / 1_000_000).toFixed(1)} МБ` + if (n >= 1000) return `${(n / 1000).toFixed(1)} КБ` + return `${n} Б` +} + +const RANGE_KEYS = ["5m", "15m", "1h", "4h", "24h"] as const +const RANGE_LABELS: Record = { + "5m": "5м", + "15m": "15м", + "1h": "1ч", + "4h": "4ч", + "24h": "24ч", +} + +function MiniAreaChart({ rx, tx, height = 44 }: { rx: number[]; tx: number[]; height?: number }) { + const W = 300 + const H = height + const maxVal = Math.max(...rx, ...tx, 1) * 1.1 + const xAt = (i: number) => (rx.length <= 1 ? 0 : (i / (rx.length - 1)) * W) + const yAt = (v: number) => H - (v / maxVal) * H + const area = (arr: number[]) => { + const pts = arr.map((v, i) => `${xAt(i).toFixed(1)},${yAt(v).toFixed(1)}`).join(" L ") + return `M 0,${H} L ${pts} L ${W},${H} Z` + } + const line = (arr: number[]) => arr.map((v, i) => `${xAt(i).toFixed(1)},${yAt(v).toFixed(1)}`).join(" ") + return ( + + + + {tx.some((v) => v > 0) ? ( + + ) : null} + + ) +} + +export function FlowEntityCardView({ + card, + selected, + onClick, +}: { + card: FlowEntityCard + selected: boolean + onClick: () => void +}) { + return ( + + ) +} + +function FlowBreakdownGrid({ rows, empty }: { rows: FlowBreakdownRow[]; empty?: string }) { + const columns = useMemo[]>( + () => [ + { + id: "label", + accessorKey: "label", + header: () => Имя, + cell: ({ row }) => ( +
+ {row.original.label} +
+
+
+
+ ), + meta: { headerClassName: DATA_GRID_CELL_PAD_FIRST, cellClassName: DATA_GRID_CELL_PAD_FIRST }, + }, + { + id: "share", + accessorKey: "percent", + header: () => Доля, + cell: ({ row }) => ( + {row.original.percent.toFixed(1)}% + ), + meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD }, + }, + { + id: "rate", + accessorFn: (r) => r.bps, + header: () => Скорость, + cell: ({ row }) => ( + {fmtRate(row.original.bps / 1_000_000)} + ), + meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD }, + }, + { + id: "bytes", + accessorKey: "bytes", + header: () => Байты, + cell: ({ row }) => ( + {formatBytes(row.original.bytes)} + ), + meta: { headerClassName: DATA_GRID_CELL_PAD_LAST, cellClassName: DATA_GRID_CELL_PAD_LAST }, + }, + ], + [], + ) + + const table = useReactTable({ + data: rows, + columns, + getCoreRowModel: getCoreRowModel(), + getRowId: (row) => row.id, + }) + + return ( + + ) +} + +export function FlowAnalyticsDetail({ + card, + analytics, + range, + onRange, + selectedIface, + onIface, + liveHint, + emptyHint, +}: { + card: FlowEntityCard | null + analytics: FlowAnalyticsDto | null + range: string + onRange: (r: string) => void + selectedIface: string + onIface: (name: string) => void + liveHint?: string + emptyHint?: string +}) { + const [slice, setSlice] = useState("applications") + const rxNow = analytics ? analytics.bpsNow / 1_000_000 : (card?.rxNow ?? 0) + const bytes = analytics?.bytes ?? card?.bytes ?? 0 + + if (!card) { + return ( +

+ Выберите сервер или клиента слева +

+ ) + } + + return ( + <> +
+
+ +

{card.name}

+ + {card.country !== "UN" ? : null} + {card.site} + +
+
+ {RANGE_KEYS.map((r) => ( + + ))} +
+
+ + {analytics?.ifaces && analytics.ifaces.length > 0 ? ( +
+
+ Интерфейс: + + {analytics.ifaces.map((iface) => { + const active = selectedIface === iface.name + return ( + + ) + })} +
+
+ ) : null} + + + +
+ , + iconClassName: "text-success", + }, + { + id: "bytes", + label: "Байт за период", + value: formatBytes(bytes), + icon: , + iconClassName: "text-info", + }, + { + id: "flows", + label: "Разговоры", + value: String(analytics?.conversations ?? card.sessions), + icon: , + iconClassName: "text-warning", + }, + ]} + /> +
+ +
+
+

Аналитика потребления

+ {analytics?.live ? live : null} +
+ setSlice(String(v))} className="gap-3"> + + Приложения + Протоколы + Источники + Назначения + Разговоры + Интерфейсы + + + + + + + + + + + + + + + + + + + + +
+ +
+
+

Объём за период

+

{formatBytes(bytes)}

+ +
+
+

Уник. адреса

+

+ {analytics?.uniqueSrc ?? 0} + src · + {analytics?.uniqueDst ?? 0} + dst +

+
+
+ + ) +} diff --git a/hooks/use-flow-live.ts b/hooks/use-flow-live.ts new file mode 100644 index 0000000..4a7bdae --- /dev/null +++ b/hooks/use-flow-live.ts @@ -0,0 +1,90 @@ +"use client" + +import { useEffect, useState } from "react" +import type { FlowAnalyticsDto } from "@mmapp/contracts/traffic-flow" +import { resolveApiUrl, withAuthHeaders } from "@/shared/api/http-client" +import { flowQuery } from "@/shared/api/traffic-flow" + +function parseSseBlock(block: string): { event: string; data: string } { + let event = "message" + const dataLines: string[] = [] + for (const line of block.split("\n")) { + if (line.startsWith("event:")) event = line.slice(6).trim() + else if (line.startsWith("data:")) dataLines.push(line.slice(5).trim()) + } + return { event, data: dataLines.join("\n") } +} + +export function useFlowLive(opts: { + enabled: boolean + backendUrl: string + range: string + serverId?: string + userId?: string + iface?: string +}): { sample: FlowAnalyticsDto | null; error: string | null } { + const [sample, setSample] = useState(null) + const [error, setError] = useState(null) + + useEffect(() => { + if (!opts.enabled) { + setSample(null) + setError(null) + return + } + + const ac = new AbortController() + setSample(null) + setError(null) + const path = `/api/traffic/flow/live${flowQuery({ + range: opts.range, + serverId: opts.serverId, + userId: opts.userId, + iface: opts.iface, + })}` + const url = resolveApiUrl(opts.backendUrl, path) + + let buf = "" + + void (async () => { + try { + const res = await fetch(url, { + headers: withAuthHeaders({ Accept: "text/event-stream" }), + signal: ac.signal, + credentials: "include", + }) + if (!res.ok || !res.body) { + setError(`live HTTP ${res.status}`) + return + } + const reader = res.body.getReader() + const decoder = new TextDecoder() + while (!ac.signal.aborted) { + const { done, value } = await reader.read() + if (done) break + buf += decoder.decode(value, { stream: true }) + const parts = buf.split("\n\n") + buf = parts.pop() ?? "" + for (const raw of parts) { + if (!raw.trim() || raw.trim().startsWith(":")) continue + const ev = parseSseBlock(raw) + if (ev.event === "sample" && ev.data) { + setSample(JSON.parse(ev.data) as FlowAnalyticsDto) + setError(null) + } else if (ev.event === "error" && ev.data) { + const parsed = JSON.parse(ev.data) as { error?: string } + setError(parsed.error ?? "live error") + } + } + } + } catch (e) { + if (ac.signal.aborted) return + setError(e instanceof Error ? e.message : "live error") + } + })() + + return () => ac.abort() + }, [opts.enabled, opts.backendUrl, opts.range, opts.serverId, opts.userId, opts.iface]) + + return { sample, error } +} diff --git a/packages/contracts/src/traffic-flow.ts b/packages/contracts/src/traffic-flow.ts index 36f24e4..e956903 100644 --- a/packages/contracts/src/traffic-flow.ts +++ b/packages/contracts/src/traffic-flow.ts @@ -79,6 +79,8 @@ export const flowTalkerDtoSchema = z.object({ packets: z.number().nonnegative(), bps: z.number().nonnegative(), inIface: z.string(), + inIfaceIndex: z.string().optional(), + application: z.string().optional(), }) export const flowStatsDtoSchema = z.object({ @@ -101,5 +103,75 @@ export type TrafficFlowSettingsDto = z.infer export type TrafficFlowOverlayResult = z.infer export type TrafficFlowHostFile = z.infer +export const flowBreakdownRowSchema = z.object({ + id: z.string(), + label: z.string(), + bytes: z.number().nonnegative(), + packets: z.number().nonnegative(), + bps: z.number().nonnegative(), + percent: z.number().nonnegative(), +}) + +export const flowIfaceChipSchema = z.object({ + name: z.string(), + index: z.string(), + bps: z.number().nonnegative(), +}) + +export const flowEntityCardSchema = z.object({ + id: z.string(), + name: z.string(), + subtitle: z.string(), + site: z.string(), + country: z.string(), + status: z.enum(["online", "offline", "degraded"]), + rxNow: z.number(), + txNow: z.number(), + sessions: z.number().int().nonnegative(), + rxSeries: z.array(z.number()), + txSeries: z.array(z.number()), + bytes: z.number().nonnegative(), +}) + +export const flowAnalyticsDtoSchema = z.object({ + bpsNow: z.number().nonnegative(), + bytes: z.number().nonnegative(), + packets: z.number().nonnegative(), + conversations: z.number().int().nonnegative(), + uniqueSrc: z.number().int().nonnegative(), + uniqueDst: z.number().int().nonnegative(), + topProto: z.string(), + rxSeries: z.array(z.number()), + txSeries: z.array(z.number()), + applications: z.array(flowBreakdownRowSchema), + protocols: z.array(flowBreakdownRowSchema), + sources: z.array(flowBreakdownRowSchema), + destinations: z.array(flowBreakdownRowSchema), + interfaces: z.array(flowBreakdownRowSchema), + conversationsList: z.array(flowTalkerDtoSchema), + ifaces: z.array(flowIfaceChipSchema), + live: z.boolean(), +}) + +export const flowExportersDtoSchema = z.object({ + exporters: z.array(flowEntityCardSchema), + lastExporterIp: z.string().nullable().optional(), + lastError: z.string().nullable().optional(), + packetsReceived: z.number().int().nonnegative().optional(), + lastDatagramAt: z.string().nullable().optional(), + listenerBound: z.boolean().optional(), + listenerAddress: z.string().nullable().optional(), +}) + +export const flowClientsDtoSchema = z.object({ + clients: z.array(flowEntityCardSchema), +}) + export type FlowTalkerDto = z.infer export type FlowStatsDto = z.infer +export type FlowBreakdownRow = z.infer +export type FlowIfaceChip = z.infer +export type FlowEntityCard = z.infer +export type FlowAnalyticsDto = z.infer +export type FlowExportersDto = z.infer +export type FlowClientsDto = z.infer diff --git a/shared/api/traffic-flow.ts b/shared/api/traffic-flow.ts index c516ded..c110ac1 100644 --- a/shared/api/traffic-flow.ts +++ b/shared/api/traffic-flow.ts @@ -1,4 +1,7 @@ import type { + FlowAnalyticsDto, + FlowClientsDto, + FlowExportersDto, FlowStatsDto, TrafficFlowHostFile, TrafficFlowOverlayResult, @@ -50,3 +53,35 @@ export async function applyTrafficFlowOverlay( export async function getTrafficFlows(baseUrl: string, range = "5m"): Promise { return requestJson(baseUrl, `/api/traffic/flows?range=${encodeURIComponent(range)}`) } + +function flowQuery(params: { + range?: string + serverId?: string + userId?: string + iface?: string +}): string { + const q = new URLSearchParams() + if (params.range) q.set("range", params.range) + if (params.serverId) q.set("serverId", params.serverId) + if (params.userId) q.set("userId", params.userId) + if (params.iface && params.iface !== "__all__") q.set("iface", params.iface) + const s = q.toString() + return s ? `?${s}` : "" +} + +export async function getFlowExporters(baseUrl: string, range = "5m"): Promise { + return requestJson(baseUrl, `/api/traffic/flow/exporters?range=${encodeURIComponent(range)}`) +} + +export async function getFlowClients(baseUrl: string, range = "5m"): Promise { + return requestJson(baseUrl, `/api/traffic/flow/clients?range=${encodeURIComponent(range)}`) +} + +export async function getFlowAnalytics( + baseUrl: string, + params: { range?: string; serverId?: string; userId?: string; iface?: string }, +): Promise { + return requestJson(baseUrl, `/api/traffic/flow/analytics${flowQuery(params)}`) +} + +export { flowQuery }