feat(traffic): показать поток NetFlow на карте сети

Скорость между узлами считается как в Трафике (5 мин, без overlay/mesh), отдельно от ёмкости WAN и BT.

Co-authored-by: Cursor <[email protected]>

backend/src/routes/traffic-flow.ts

@@ -24,6 +24,7 @@ import {
   listFlowExporters,
   safeBuildLiveFlowSample,
 } from "../services/traffic-flow-analytics.js"
+import { buildFlowMapHops } from "../services/traffic-flow-map-hops.js"
 import { applyFlowOverlay } from "../services/traffic-flow-overlay.js"
 import { listTrafficFlowHostFiles } from "../services/traffic-flow-host-files.js"
 import { appendEvent } from "../modules/events/service/events-service.js"
@@ -222,6 +223,10 @@ const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => {
     return reply.send(buildFlowAnalytics(analyticsQuery(req)))
   })
 
+  app.get("/traffic/flow/map-hops", async (req, reply) => {
+    return reply.send(buildFlowMapHops(analyticsQuery(req)))
+  })
+
   app.get("/traffic/flow/monthly", async (req, reply) => {
     const q = req.query as { month?: string; serverId?: string }
     const now = new Date()

backend/src/services/traffic-flow-map-hops.test.ts

@@ -0,0 +1,157 @@
+import assert from "node:assert/strict"
+import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js"
+import {
+  ingestParsedFlowsForServerForTests,
+  resetFlowRingsForTests,
+} from "./traffic-flow-ingest.js"
+import { buildFlowMapHops } from "./traffic-flow-map-hops.js"
+import { seedFlowTopologyForTests, type FlowTopology } from "./traffic-flow-topology.js"
+import { disableCatalogFetchForTests, resetFlowCatalogForTests } from "./traffic-flow-classify.js"
+import {
+  disableRipeEnqueueForTests,
+  disableRipePersistForTests,
+  resetRipeCacheForTests,
+} from "./traffic-flow-ripe.js"
+
+disableCatalogFetchForTests()
+resetFlowCatalogForTests()
+disableRipePersistForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+
+const topo: FlowTopology = {
+  clientIfaces: new Map([[7, new Set(["gre-client"])]]),
+  clientByIface: new Map([["7|gre-client", {
+    userId: "u1",
+    login: "alice",
+    name: "Alice",
+    serverId: 7,
+    interfaceName: "gre-client",
+  }]]),
+  enNodes: [{ id: 9, name: "NSK-EN", hosts: ["198.51.100.1"] }],
+  enHosts: new Set(["198.51.100.1"]),
+  jhHosts: new Set(["203.0.113.10"]),
+  wanIfaces: new Map([[3, new Set(["ether1-rt"])]]),
+  plane: {
+    clientIfaceNames: new Set(["gre-client"]),
+    enHosts: new Set(["198.51.100.1"]),
+    jhHosts: new Set(["203.0.113.10"]),
+  },
+}
+
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+seedFlowTopologyForTests(topo)
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "gre-client" },
+  { ".id": "*3", name: "gre-jh-en" },
+  { ".id": "*A", name: "wg-flow" },
+])
+rememberServerIfaces(3, [
+  { ".id": "*1", name: "ether1-rt" },
+])
+
+ingestParsedFlowsForServerForTests(7, [
+  {
+    src: "10.100.1.17",
+    dst: "8.8.8.8",
+    proto: 6,
+    srcPort: 51234,
+    dstPort: 443,
+    bytes: 12_000,
+    packets: 10,
+    inIface: "2",
+    outIface: "3",
+    nextHop: "198.51.100.1",
+  },
+  {
+    src: "203.0.113.10",
+    dst: "198.51.100.1",
+    proto: 47,
+    srcPort: 0,
+    dstPort: 0,
+    bytes: 5_000_000,
+    packets: 4000,
+    inIface: "3",
+    outIface: "3",
+  },
+  {
+    src: "10.100.1.17",
+    dst: "10.100.1.18",
+    proto: 6,
+    srcPort: 50000,
+    dstPort: 443,
+    bytes: 8000,
+    packets: 8,
+    inIface: "2",
+    outIface: "2",
+  },
+  {
+    src: "10.255.254.1",
+    dst: "10.255.254.2",
+    proto: 17,
+    srcPort: 4739,
+    dstPort: 2055,
+    bytes: 400,
+    packets: 2,
+    inIface: "10",
+    outIface: "",
+  },
+])
+ingestParsedFlowsForServerForTests(3, [
+  {
+    src: "192.168.1.10",
+    dst: "8.8.4.4",
+    proto: 6,
+    srcPort: 40000,
+    dstPort: 443,
+    bytes: 3000,
+    packets: 4,
+    inIface: "1",
+    outIface: "1",
+  },
+])
+
+try {
+  const def = buildFlowMapHops({ minutes: 5 })
+  assert.equal(def.excludeOverlayApplied, true)
+  assert.equal(def.excludeMeshApplied, true)
+  assert.equal(def.dedupApplied, true)
+  assert.equal(def.windowSec, 300)
+
+  const payloadGre = def.hops.find((h) => h.kind === "gre" && h.fromId === "7" && h.toId === "9")
+  assert.ok(payloadGre, "payload JH→EN hop")
+  assert.equal(payloadGre.bytes, 12_000)
+  assert.equal(payloadGre.bps, (12_000 * 8) / 300)
+  assert.equal(payloadGre.bpsFwd, (12_000 * 8) / 300)
+  assert.equal(payloadGre.iface, "gre-jh-en")
+
+  const greIface = def.hops.find((h) => h.kind === "iface" && h.iface === "gre-jh-en" && h.fromId === "7")
+  assert.ok(greIface)
+  assert.equal(greIface.bytes, 12_000)
+  assert.equal(greIface.bpsFwd, (12_000 * 8) / 300)
+
+  assert.ok(!def.hops.some((h) => h.bytes >= 5_000_000), "overlay GRE proto 47 excluded")
+  assert.ok(!def.hops.some((h) => h.iface === "wg-flow"), "mgmt wg-flow excluded")
+  const clientIngress = def.hops.find((h) => h.iface === "gre-client" && h.fromId === "7" && h.kind === "iface")
+  assert.ok(clientIngress, "payload ingress on client iface")
+  assert.equal(clientIngress.bytes, 12_000)
+
+  const wan = def.hops.find((h) => h.kind === "wan" && h.fromId === "3" && h.iface === "ether1-rt")
+  assert.ok(wan, "WAN hop from home-router")
+  assert.equal(wan.bytes, 3000)
+
+  const withAll = buildFlowMapHops({ minutes: 5, excludeOverlay: false, excludeMesh: false })
+  const overlayIface = withAll.hops.find((h) => h.iface === "gre-jh-en" && h.fromId === "7")
+  assert.ok(overlayIface && overlayIface.bytes >= 5_000_000)
+  const meshIface = withAll.hops.find((h) => h.iface === "gre-client" && h.fromId === "7" && h.kind === "iface")
+  assert.ok(meshIface && meshIface.bytes >= 20_000)
+} finally {
+  seedFlowTopologyForTests(null)
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
+console.log("traffic-flow-map-hops.test.ts: ok")

backend/src/services/traffic-flow-map-hops.ts

@@ -0,0 +1,216 @@
+import { eq } from "drizzle-orm"
+import type { FlowMapHop, FlowMapHopsDto } from "@mmapp/contracts/traffic-flow"
+import { db } from "../db/index.js"
+import { servers, userInterfaceBindings } from "../db/schema.js"
+import { flowRowMatchesFilter } from "./traffic-flow-apps.js"
+import { dedupFlowRowsMaxBytes } from "./traffic-flow-dedup.js"
+import { getFlowListenerState, listFlowRowsForWindow } from "./traffic-flow-ingest.js"
+import { resolveIfaceName } from "./traffic-flow-ifaces.js"
+import { classifyFlowPlane, shouldKeepPlane } from "./traffic-flow-planes.js"
+import { loadFlowTopology, resolveEn } from "./traffic-flow-topology.js"
+
+export interface FlowMapHopsQuery {
+  minutes: number
+  serverId?: number
+  userId?: string
+  iface?: string
+  dedup?: boolean
+  excludeMesh?: boolean
+  excludeOverlay?: boolean
+}
+
+interface HopAcc {
+  fromId: string
+  fromLabel: string
+  toId: string
+  toLabel: string
+  kind: FlowMapHop["kind"]
+  iface?: string
+  bytes: number
+  bytesFwd: number
+  bytesRev: number
+}
+
+function userIfaceAllow(userId: string): Map<number, Set<string>> | null {
+  if (!userId) return null
+  const binds = db.select().from(userInterfaceBindings).where(eq(userInterfaceBindings.userId, userId)).all()
+  const allow = new Map<number, Set<string>>()
+  for (const b of binds) {
+    const set = allow.get(b.serverId) ?? new Set<string>()
+    set.add(b.interfaceName)
+    allow.set(b.serverId, set)
+  }
+  return allow
+}
+
+function ifaceUsable(name: string): boolean {
+  return Boolean(name) && name !== "—"
+}
+
+function bump(acc: Map<string, HopAcc>, key: string, seed: Omit<HopAcc, "bytes" | "bytesFwd" | "bytesRev">, bytes: number, dir: "fwd" | "rev" | "both"): void {
+  const prev = acc.get(key)
+  const addFwd = dir === "fwd" || dir === "both" ? bytes : 0
+  const addRev = dir === "rev" || dir === "both" ? bytes : 0
+  if (prev) {
+    prev.bytes += bytes
+    prev.bytesFwd += addFwd
+    prev.bytesRev += addRev
+    if (seed.iface && !prev.iface) prev.iface = seed.iface
+    return
+  }
+  acc.set(key, {
+    ...seed,
+    bytes,
+    bytesFwd: addFwd,
+    bytesRev: addRev,
+  })
+}
+
+function toHop(a: HopAcc, windowSec: number): FlowMapHop {
+  return {
+    fromId: a.fromId,
+    fromLabel: a.fromLabel,
+    toId: a.toId,
+    toLabel: a.toLabel,
+    kind: a.kind,
+    ...(a.iface ? { iface: a.iface } : {}),
+    bytes: a.bytes,
+    bps: (a.bytes * 8) / windowSec,
+    bpsFwd: (a.bytesFwd * 8) / windowSec,
+    bpsRev: (a.bytesRev * 8) / windowSec,
+  }
+}
+
+/** Hop-rates для карты сети: те же фильтры, что у общего NetFlow (dedup / mesh / overlay). */
+export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
+  const windowSec = Math.max(60, q.minutes * 60)
+  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]))
+  const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : ""
+  const wantDedup = q.dedup !== false && !ifaceFilter
+  const excludeMesh = q.excludeMesh !== false
+  const excludeOverlay = q.excludeOverlay !== false
+  const topo = loadFlowTopology()
+
+  const matched = []
+  for (const r of raw) {
+    const resolved = resolveIfaceName(r.serverId, r.inIface)
+    const outResolved = resolveIfaceName(r.serverId, r.outIface)
+    if (!flowRowMatchesFilter(r, resolved.name, q, allow)) continue
+    const plane = classifyFlowPlane({
+      src: r.src,
+      dst: r.dst,
+      proto: r.proto,
+      srcPort: r.srcPort,
+      dstPort: r.dstPort,
+      inIface: resolved.name,
+      outIface: outResolved.name,
+    }, topo.plane)
+    if (!shouldKeepPlane(plane, { excludeMesh, excludeOverlay })) continue
+    matched.push(r)
+  }
+
+  const working = wantDedup ? dedupFlowRowsMaxBytes(matched) : matched
+  const hops = new Map<string, HopAcc>()
+
+  for (const r of working) {
+    const inRes = resolveIfaceName(r.serverId, r.inIface)
+    const outRes = resolveIfaceName(r.serverId, r.outIface)
+    const inName = inRes.name
+    const outName = outRes.name
+    const fromId = String(r.serverId)
+    const fromLabel = nameById.get(r.serverId) ?? fromId
+    const wanSet = topo.wanIfaces.get(r.serverId)
+
+    const inOk = ifaceUsable(inName)
+    const outOk = ifaceUsable(outName)
+    const sameIface = inOk && outOk && inName.toLowerCase() === outName.toLowerCase()
+    if (sameIface) {
+      bump(hops, `iface|${fromId}|${inName.toLowerCase()}`, {
+        fromId,
+        fromLabel,
+        toId: "",
+        toLabel: "",
+        kind: "iface",
+        iface: inName,
+      }, r.bytes, "fwd")
+    } else {
+      if (inOk) {
+        bump(hops, `iface|${fromId}|${inName.toLowerCase()}`, {
+          fromId,
+          fromLabel,
+          toId: "",
+          toLabel: "",
+          kind: "iface",
+          iface: inName,
+        }, r.bytes, "rev")
+      }
+      if (outOk) {
+        bump(hops, `iface|${fromId}|${outName.toLowerCase()}`, {
+          fromId,
+          fromLabel,
+          toId: "",
+          toLabel: "",
+          kind: "iface",
+          iface: outName,
+        }, r.bytes, "fwd")
+      }
+    }
+
+    const enOut = ifaceUsable(outName) ? resolveEn(topo, r.nextHop, outName) : null
+    const enIn = ifaceUsable(inName) ? resolveEn(topo, "", inName) : null
+    const en = (enOut && enOut.id !== r.serverId ? enOut : null)
+      ?? (enIn && enIn.id !== r.serverId ? enIn : null)
+    if (en) {
+      const toId = String(en.id)
+      const dir: "fwd" | "rev" = enOut && enOut.id === en.id ? "fwd" : "rev"
+      const greIface = dir === "fwd" && ifaceUsable(outName) ? outName : (ifaceUsable(inName) ? inName : undefined)
+      bump(hops, `gre|${fromId}|${toId}`, {
+        fromId,
+        fromLabel,
+        toId,
+        toLabel: en.name,
+        kind: "gre",
+        iface: greIface,
+      }, r.bytes, dir)
+    }
+
+    if (wanSet?.size) {
+      if (ifaceUsable(inName) && wanSet.has(inName)) {
+        bump(hops, `wan|${fromId}|${inName.toLowerCase()}`, {
+          fromId,
+          fromLabel,
+          toId: "",
+          toLabel: "",
+          kind: "wan",
+          iface: inName,
+        }, r.bytes, "rev")
+      }
+      if (ifaceUsable(outName) && wanSet.has(outName) && outName.toLowerCase() !== inName.toLowerCase()) {
+        bump(hops, `wan|${fromId}|${outName.toLowerCase()}`, {
+          fromId,
+          fromLabel,
+          toId: "",
+          toLabel: "",
+          kind: "wan",
+          iface: outName,
+        }, r.bytes, "fwd")
+      }
+    }
+  }
+
+  const listener = getFlowListenerState()
+  return {
+    hops: [...hops.values()]
+      .map((a) => toHop(a, windowSec))
+      .sort((a, b) => b.bytes - a.bytes),
+    live: listener.bound,
+    rangeMinutes: q.minutes,
+    windowSec,
+    dedupApplied: wantDedup,
+    excludeMeshApplied: excludeMesh,
+    excludeOverlayApplied: excludeOverlay,
+  }
+}