Compare commits

..
1 Commits
Author SHA1 Message Date
DenozordecandCursor 60a0970d73 feat(traffic-flow): enhance traffic flow analytics and overlay configuration
Docker images / prepare-release (push) Successful in 12s
Docker images / backend-test (push) Successful in 2m42s
Docker images / frontend-image (push) Successful in 3m22s
Docker images / updater-image (push) Successful in 46s
Docker images / backend-image (push) Successful in 2m55s
Docker images / notify-webhook (push) Skipped
Docker images / publish-release (push) Successful in 12s
- Updated the traffic flow analytics to include new state variables for named bytes, total bytes, and window seconds, improving data granularity.
- Modified the NetworkMapPage to display classified traffic data, including a new section for showing classified and total bytes.
- Enhanced the applyFlowOverlay function to support a new option for disabling GRE fast path, allowing for more flexible traffic flow configurations.
- Added a toggle in the FlowOverlaySheet component to enable or disable the GRE fast path setting, improving user control over traffic processing.
- Updated tests to validate the new functionalities and ensure accurate traffic flow analytics.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-12 00:03:13 +07:00
18 changed files with 365 additions and 69 deletions
+29 -1
View File
@@ -761,7 +761,7 @@ function ServiceNode({
onMouseDown={(e) => { e.stopPropagation(); onMouseDown(e) }}
onClick={(e) => { e.stopPropagation(); onClick() }}
>
<title>{`${label} · ${serviceSharePct(share)} трафика окна`}</title>
<title>{`${label} · ${serviceSharePct(share)} payload окна`}</title>
{isSel && (
<rect
x={-bw / 2 - 6}
@@ -1100,6 +1100,9 @@ export default function NetworkMapPage() {
const [mapServiceEdges, setMapServiceEdges] = useState<FlowMapServiceEdge[]>([])
const [mapServicePaths, setMapServicePaths] = useState<FlowMapServicePath[]>([])
const [mapSharePct, setMapSharePct] = useState(5)
const [mapNamedBytes, setMapNamedBytes] = useState(0)
const [mapTotalBytes, setMapTotalBytes] = useState(0)
const [mapWindowSec, setMapWindowSec] = useState(300)
/** FQDN из GRE outer → IPv4 (ответ POST /api/network/resolve-hosts), для матчинга с WAN. */
const [greResolvedIpv4ByHost, setGreResolvedIpv4ByHost] = useState<Record<string, string>>({})
const [dataError, setDataError] = useState<string | null>(null)
@@ -1196,6 +1199,8 @@ export default function NetworkMapPage() {
setMapServiceEdges(MOCK_MAP_SERVICE_EDGES)
setMapServicePaths(MOCK_MAP_SERVICE_PATHS)
setMapSharePct(5)
setMapNamedBytes(0)
setMapTotalBytes(0)
setDataError(null)
})
return
@@ -1303,6 +1308,9 @@ export default function NetworkMapPage() {
setMapServiceEdges(res.serviceEdges ?? [])
setMapServicePaths(res.servicePaths ?? [])
if (res.mapServiceMinSharePct != null) setMapSharePct(res.mapServiceMinSharePct)
setMapNamedBytes(res.namedBytes ?? 0)
setMapTotalBytes(res.totalBytes ?? 0)
if (res.windowSec) setMapWindowSec(res.windowSec)
})
.catch((err: unknown) => {
if (cancelled) return
@@ -2739,6 +2747,26 @@ export default function NetworkMapPage() {
<span className="text-xs text-muted-foreground">Доля окна</span>
<span className="text-xs font-mono font-medium text-cyan-400">{serviceSharePct(liveSelectedService.share)}</span>
</div>
{mapTotalBytes > 0 && (
<div className="flex items-center justify-between py-2 border-b border-border/50">
<span className="text-xs text-muted-foreground">Классифицировано</span>
<span className="text-xs font-mono font-medium text-cyan-400">
{formatNetflowRate({
bytes: mapNamedBytes,
bps: (mapNamedBytes * 8) / Math.max(1, mapWindowSec),
bpsFwd: 0,
bpsRev: 0,
})}
{" из "}
{formatNetflowRate({
bytes: mapTotalBytes,
bps: (mapTotalBytes * 8) / Math.max(1, mapWindowSec),
bpsFwd: 0,
bpsRev: 0,
})}
</span>
</div>
)}
<div className="flex items-center justify-between py-2 border-b border-border/50">
<span className="text-xs text-muted-foreground">Скорость</span>
<span className="text-xs font-mono font-medium">
+1
View File
@@ -113,6 +113,7 @@ async function applyOverlayHandler(req: FastifyRequest, reply: FastifyReply) {
const result = await applyFlowOverlay(parsed.data.serverId, {
publicEndpoint: parsed.data.publicEndpoint,
requestHost: requestPublicHost(req),
disableGreFastPath: parsed.data.disableGreFastPath,
})
return reply.send(result)
} catch (e) {
@@ -388,8 +388,8 @@ try {
assert.equal(def.excludeOverlayApplied, true)
assert.equal(def.excludeMeshApplied, true)
assert.ok(!def.conversationsList.some((r) => r.proto === 47))
assert.equal(def.conversationsList[0]?.service, "Google")
assert.equal(def.conversationsList[0]?.category, "Веб")
assert.equal(def.conversationsList[0]?.service, "YouTube")
assert.equal(def.conversationsList[0]?.category, "Видео / стриминг")
assert.equal(def.conversationsList[0]?.clientName, "Alice")
assert.equal(def.conversationsList[0]?.enName, "NSK-SERVHOST-RTK")
assert.equal(def.conversationsList[0]?.plane, "payload")
@@ -445,8 +445,8 @@ try {
const rev = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
const google = rev.conversationsList.find((r) => r.src === "173.194.151.65")
const cf = rev.conversationsList.find((r) => r.src === "104.18.35.51")
assert.equal(google?.service, "Google")
assert.equal(google?.category, "Веб")
assert.equal(google?.service, "YouTube")
assert.equal(google?.category, "Видео / стриминг")
assert.equal(cf?.service, "Cloudflare")
assert.equal(cf?.category, "CDN")
} finally {
@@ -32,6 +32,8 @@ assert.equal(brandByAsn(401115)?.service, "ChatGPT")
assert.equal(lookupBrand("1.1.1.1", 13335)?.service, "Cloudflare")
assert.equal(lookupBrand("104.18.35.51", 0)?.service, "Cloudflare")
assert.equal(lookupBrand("173.194.151.65", 0)?.service, "Google")
assert.equal(lookupBrand("64.233.161.1", 0)?.service, "Google")
assert.equal(lookupBrand("142.250.1.10", 0)?.service, "Google")
assert.equal(lookupBrand("8.8.8.8", 0)?.service, "Google")
assert.equal(lookupBrand("203.0.113.9", 64500), null)
assert.equal(OTHER_SERVICE, "Прочее")
@@ -41,6 +43,7 @@ assert.equal(isNamedInternetService("GRE", "Туннель"), false)
assert.equal(isNamedInternetService("DNS", "DNS"), false)
assert.equal(mapServiceNodeId("AWS"), "svc:aws")
assert.equal(mapServiceNodeId("Cloudflare"), "svc:cloudflare")
assert.equal(mapServiceNodeId("Прочее"), "svc:other")
assert.equal(brandByAsn(714)?.service, "Apple")
assert.equal(brandByAsn(714)?.category, "CDN")
@@ -70,6 +73,12 @@ assert.equal(isSteamGamePort(6, 443, 50000), false)
assert.equal(resolveFlowBrand("104.18.35.51", 32590, "VALVE-CORPORATION", 6, 443, 1)?.service, "Cloudflare")
assert.equal(resolveFlowBrand("203.0.113.9", 32590, "", 17, 27015, 50000)?.service, "Steam")
assert.equal(resolveFlowBrand("8.8.8.8", 15169, "GOOGLE", 6, 443, 51234)?.service, "Google")
assert.equal(resolveFlowBrand("173.194.160.163", 15169, "GOOGLE", 6, 443, 51234)?.service, "YouTube")
assert.equal(resolveFlowBrand("64.233.161.1", 0, "", 17, 443, 50000)?.service, "YouTube")
assert.equal(resolveFlowBrand("64.233.161.1", 0, "", 6, 80, 50000)?.service, "Google")
assert.equal(resolveFlowBrand("2001:4860:4860::8888", 15169, "GOOGLE", 17, 53, 53000)?.service, "Google")
assert.equal(resolveFlowBrand("2001:4860:4860::8888", 15169, "GOOGLE", 17, 443, 50000)?.service, "YouTube")
assert.equal(resolveRipeCountry("", 9059, ""), "IE")
assert.equal(resolveRipeCountry("", 24940, ""), "DE")
+30 -3
View File
@@ -174,6 +174,14 @@ const CIDR_BRANDS: Array<{ cidr: string; prefixLen: number; hit: BrandHit }> = [
{ cidr: "172.217.0.0/16", prefixLen: 16, hit: GOOGLE },
{ cidr: "74.125.0.0/16", prefixLen: 16, hit: GOOGLE },
{ cidr: "142.250.0.0/15", prefixLen: 15, hit: GOOGLE },
{ cidr: "64.233.0.0/16", prefixLen: 16, hit: GOOGLE },
{ cidr: "66.102.0.0/16", prefixLen: 16, hit: GOOGLE },
{ cidr: "66.249.64.0/19", prefixLen: 19, hit: GOOGLE },
{ cidr: "72.14.192.0/18", prefixLen: 18, hit: GOOGLE },
{ cidr: "108.177.0.0/16", prefixLen: 16, hit: GOOGLE },
{ cidr: "209.85.128.0/17", prefixLen: 17, hit: GOOGLE },
{ cidr: "216.58.192.0/19", prefixLen: 19, hit: GOOGLE },
{ cidr: "216.239.32.0/19", prefixLen: 19, hit: GOOGLE },
{ cidr: "208.65.152.0/22", prefixLen: 22, hit: YOUTUBE },
{ cidr: "208.117.224.0/19", prefixLen: 19, hit: YOUTUBE },
].sort((a, b) => b.prefixLen - a.prefixLen)
@@ -196,6 +204,16 @@ const HOLDER_BRANDS: Array<{ re: RegExp; hit: BrandHit }> = [
const NON_ISO = new Set(["EU", "AP", "ZZ", "XX", "A1", "A2", "O1"])
const STEAM_ASN = 32590
const GOOGLE_FRONT_ASN = new Set([15169, 396982])
function isGooglePublicDns(ip: string): boolean {
return ipInCidrV4(ip, "8.8.8.0/24") || ipInCidrV4(ip, "8.8.4.0/24")
}
function isHttpsOrQuic(proto: number, dstPort: number, srcPort: number): boolean {
if (proto !== 6 && proto !== 17) return false
return dstPort === 443 || srcPort === 443
}
export function isIsoCountry(code: string): boolean {
const c = String(code ?? "").trim().toUpperCase()
@@ -273,7 +291,15 @@ export function resolveFlowBrand(
if (cidrBrand?.service === "Cloudflare") return cidrBrand
const holderBrand = brandByHolder(holder)
if (holderBrand) return holderBrand
const fromLookup = cidrBrand || brandByAsn(asn)
const asnBrand = brandByAsn(asn)
if (
!isGooglePublicDns(ip)
&& isHttpsOrQuic(proto, dstPort, srcPort)
&& (GOOGLE_FRONT_ASN.has(asn) || cidrBrand?.service === "Google" || asnBrand?.service === "Google")
) {
return YOUTUBE
}
const fromLookup = cidrBrand || asnBrand
if (fromLookup) return fromLookup
if (asn === STEAM_ASN && isSteamGamePort(proto, dstPort, srcPort)) return STEAM
return null
@@ -300,8 +326,9 @@ export function isNamedInternetService(service: string, category: string): boole
}
export function mapServiceNodeId(label: string): string {
const slug = label
.trim()
const raw = label.trim()
if (raw === OTHER_SERVICE) return "svc:other"
const slug = raw
.toLowerCase()
.replace(/[^a-z0-9]+/g, "-")
.replace(/^-+|-+$/g, "")
@@ -33,10 +33,10 @@ const google = classifyFlowDst("173.194.160.163", 6, 443, 1, {
ok: true,
fetchedAt: Date.now(),
})
assert.equal(google.service, "Google")
assert.equal(google.category, "Веб")
assert.equal(google.service, "YouTube")
assert.equal(google.category, "Видео / стриминг")
const googleCidr = classifyFlowDst("173.194.151.65", 6, 57182, 443, null)
const googleCidr = classifyFlowDst("173.194.151.65", 6, 80, 50000, null)
assert.equal(googleCidr.service, "Google")
assert.equal(googleCidr.category, "Веб")
@@ -115,8 +115,8 @@ const googleCloud = classifyFlowDst("203.0.113.43", 6, 443, 1, {
ok: true,
fetchedAt: Date.now(),
})
assert.equal(googleCloud.service, "Google")
assert.equal(googleCloud.category, "Веб")
assert.equal(googleCloud.service, "YouTube")
assert.equal(googleCloud.category, "Видео / стриминг")
const gre = classifyFlowDst("198.51.100.1", 47, 0, 0, null)
assert.equal(gre.service, "GRE")
+13 -1
View File
@@ -4,7 +4,7 @@ import {
resetEngineForTests,
} from "./traffic-flow-engine.js"
import { factsSnapshotForTests } from "./traffic-flow-facts.js"
import { classifyInternetBrand } from "./traffic-flow-dest.js"
import { classifyInternetBrand, mapInternetBrand } from "./traffic-flow-dest.js"
import { disableCatalogFetchForTests, resetFlowCatalogForTests } from "./traffic-flow-classify.js"
import {
disableRipeEnqueueForTests,
@@ -136,6 +136,18 @@ assert.equal(classifyInternetBrand("8.8.8.8", 6, 443, 51234, {
ok: true,
fetchedAt: Date.now(),
})?.service, "Google")
assert.equal(mapInternetBrand("203.0.113.50", 6, 443, 51234, null).service, "Прочее")
assert.equal(mapInternetBrand("8.8.8.8", 47, 0, 0, null).service, "Прочее")
assert.equal(mapInternetBrand("8.8.8.8", 6, 443, 51234, {
prefix: "8.8.8.0/24",
asn: 15169,
country: "US",
lat: null,
lng: null,
holder: "GOOGLE",
ok: true,
fetchedAt: Date.now(),
}).service, "Google")
resetEngineForTests()
seedFlowTopologyForTests(null)
+15 -1
View File
@@ -1,5 +1,5 @@
import { applicationName } from "./traffic-flow-apps.js"
import { isIsoCountry, isNamedInternetService, resolveFlowBrand } from "./traffic-flow-brands.js"
import { isIsoCountry, isNamedInternetService, OTHER_SERVICE, resolveFlowBrand } from "./traffic-flow-brands.js"
import { classifyFlowDst, type FlowClassification } from "./traffic-flow-classify.js"
import { resolveFlowIp } from "./traffic-flow-geoip.js"
import { canonicalFactIface } from "./traffic-flow-ifindex.js"
@@ -57,6 +57,20 @@ export function classifyInternetBrand(
return brand
}
const OTHER_BRAND: FlowClassification = { service: OTHER_SERVICE, category: OTHER_SERVICE }
/** Бренд для карты: именованный сервис или «Прочее» (GRE/ESP не сервис). */
export function mapInternetBrand(
dst: string,
proto: number,
dstPort: number,
srcPort: number,
ripe: FlowIpMeta | null,
): FlowClassification {
if (proto === 47 || proto === 50) return OTHER_BRAND
return classifyInternetBrand(dst, proto, dstPort, srcPort, ripe) ?? OTHER_BRAND
}
export function resolveInternetDest(opts: {
src: string
dst: string
@@ -15,6 +15,7 @@ import {
lastFlushUsedTransactionForTests,
maybeRefreshIfaces,
peekPendingFlows,
capFlowRowsPerServerBucket,
resetFlowRingsForTests,
setPendingCapForTests,
setRefreshIfacesForTests,
@@ -146,4 +147,20 @@ resetFlowRingsForTests()
resetIfaceCacheForTests()
setRefreshIfacesForTests(null)
{
const minute0 = "2026-01-01T00:00:00.000Z"
const minute1 = "2026-01-01T00:01:00.000Z"
const rows: Array<{ serverId: number; bucketAt: string; bytes: number; id: string }> = []
for (let i = 1; i <= 25; i++) {
rows.push({ serverId: 1, bucketAt: minute0, bytes: i, id: `a${i}` })
rows.push({ serverId: 2, bucketAt: minute0, bytes: i, id: `b${i}` })
}
rows.push({ serverId: 1, bucketAt: minute1, bytes: 1, id: "a-min-other-minute" })
const capped = capFlowRowsPerServerBucket(rows, 20)
assert.equal(capped.filter((r) => r.serverId === 1 && r.bucketAt === minute0).length, 20)
assert.equal(capped.filter((r) => r.serverId === 2).length, 20)
assert.ok(capped.some((r) => r.id === "a-min-other-minute"), "другая минута не режется глобальным top-N")
assert.ok(!capped.some((r) => r.id === "a1" || r.id === "b1"), "мелкие 5-tuple сервера выпадают только в своём bucket")
}
console.log("traffic-flow-ingest.test.ts: ok")
+83 -25
View File
@@ -1,8 +1,7 @@
import { Worker } from "node:worker_threads"
import { gte, sql } from "drizzle-orm"
import { db, dbGet, dbQuery, pool, withAdvisoryLock } from "../db/index.js"
import { db, dbAll, dbGet, dbQuery, pool, withAdvisoryLock } from "../db/index.js"
import { dropExpiredPartitions } from "../db/partitions.js"
import { flowBuckets, servers } from "../db/schema.js"
import { servers } from "../db/schema.js"
import type { FlowPurgeDto, FlowStatsDto, FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
import { protoName, type ParsedFlowInput } from "./traffic-flow-parse.js"
import type { CollectorHeartbeat, ExporterMapPayload, MainToWorker, WorkerToMain } from "./traffic-flow-collector-ipc.js"
@@ -299,6 +298,32 @@ function mergeInto(map: Map<string, PendingFlowRow>, row: PendingFlowRow): void
map.set(key, { ...row })
}
/** Top-N разговоров на (server_id, bucket_at), как prune persist — не глобальный ORDER BY bytes. */
export function capFlowRowsPerServerBucket<T extends { serverId: number; bucketAt: string; bytes: number }>(
rows: T[],
keep: number,
): T[] {
const cap = Math.max(20, keep)
const groups = new Map<string, T[]>()
for (const row of rows) {
const k = `${row.serverId}\0${row.bucketAt}`
const list = groups.get(k)
if (list) list.push(row)
else groups.set(k, [row])
}
const out: T[] = []
for (const list of groups.values()) {
list.sort((a, b) => b.bytes - a.bytes)
out.push(...list.slice(0, cap))
}
return out
}
function isoBucketAt(v: unknown): string {
if (v instanceof Date) return v.toISOString()
return String(v ?? "")
}
export async function listLiveFlowRows(sinceIso: string): Promise<PendingFlowRow[]> {
if (worker && lastHeartbeat?.workerAlive) {
return await listStoredFlowRows(sinceIso)
@@ -306,34 +331,67 @@ export async function listLiveFlowRows(sinceIso: string): Promise<PendingFlowRow
return engineListLive(sinceIso)
}
interface StoredBucketRow {
server_id: number
bucket_at: string | Date
src: string
dst: string
proto: number
src_port: number
dst_port: number
bytes: number
packets: number
in_iface: string
out_iface: string | null
next_hop: string | null
flow_start_ms: number | null
flow_end_ms: number | null
nat_src: string | null
nat_dst: string | null
nat_src_port: number | null
nat_dst_port: number | null
}
export async function listStoredFlowRows(sinceIso: string): Promise<PendingFlowRow[]> {
const settings = await getTrafficFlowSettingsRow()
const cap = Math.max(20, settings.topN) * 60
const stored = await db.select().from(flowBuckets)
.where(gte(flowBuckets.bucketAt, sinceIso))
.orderBy(sql`${flowBuckets.bytes} DESC`)
.limit(cap)
const keep = Math.max(20, settings.topN)
const stored = await dbAll<StoredBucketRow>(`
SELECT
server_id, bucket_at, src::text AS src, dst::text AS dst, proto,
src_port, dst_port, bytes, packets, in_iface, out_iface,
next_hop::text AS next_hop, flow_start_ms, flow_end_ms,
nat_src::text AS nat_src, nat_dst::text AS nat_dst, nat_src_port, nat_dst_port
FROM (
SELECT fb.*,
ROW_NUMBER() OVER (
PARTITION BY server_id, bucket_at ORDER BY bytes DESC
) AS rn
FROM flow_buckets fb
WHERE bucket_at >= $1
) ranked
WHERE rn <= $2
`, [sinceIso, keep])
const merged = new Map<string, PendingFlowRow>()
for (const r of stored) {
mergeInto(merged, {
serverId: r.serverId,
bucketAt: r.bucketAt,
serverId: Number(r.server_id),
bucketAt: isoBucketAt(r.bucket_at),
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: r.outIface ?? "",
nextHop: r.nextHop ?? "",
flowStartMs: r.flowStartMs ?? 0,
flowEndMs: r.flowEndMs ?? 0,
natSrc: r.natSrc ?? "",
natDst: r.natDst ?? "",
natSrcPort: r.natSrcPort ?? 0,
natDstPort: r.natDstPort ?? 0,
proto: Number(r.proto) || 0,
srcPort: Number(r.src_port) || 0,
dstPort: Number(r.dst_port) || 0,
bytes: Number(r.bytes) || 0,
packets: Number(r.packets) || 0,
inIface: r.in_iface ?? "",
outIface: r.out_iface ?? "",
nextHop: r.next_hop ?? "",
flowStartMs: Number(r.flow_start_ms) || 0,
flowEndMs: Number(r.flow_end_ms) || 0,
natSrc: r.nat_src ?? "",
natDst: r.nat_dst ?? "",
natSrcPort: Number(r.nat_src_port) || 0,
natDstPort: Number(r.nat_dst_port) || 0,
})
}
if (!worker) {
@@ -342,7 +400,7 @@ export async function listStoredFlowRows(sinceIso: string): Promise<PendingFlowR
mergeInto(merged, p)
}
}
return [...merged.values()]
return capFlowRowsPerServerBucket([...merged.values()], keep)
}
export async function listFlowRowsForWindow(minutes: number): Promise<PendingFlowRow[]> {
@@ -70,6 +70,21 @@ import {
console.log("traffic-flow-map-hops.test.ts: pickMapServices ok")
}
{
const leak = pickMapServices(
[
{ id: "svc:other", label: "Прочее", category: "Прочее", bytes: 19_000, bps: 0, share: 19 / 30 },
{ id: "svc:google", label: "Google", category: "Веб", bytes: 11_000, bps: 0, share: 11 / 30 },
],
5,
)
assert.equal(leak.reduce((n, s) => n + s.bytes, 0), 30_000)
const google = leak.find((s) => s.id === "svc:google")
assert.ok(google)
assert.ok(google.share < 0.4, "доля от окна хопа, не от named-only")
assert.ok(leak.some((s) => s.id === "svc:other"), "дыра видна как Прочее")
}
if (!(await withPgOrSkip())) {
console.log("traffic-flow-map-hops.test.ts: skip")
process.exit(0)
@@ -279,12 +294,18 @@ try {
resetFlowMapHopsCacheForTests()
const six = await buildFlowMapHops({ minutes: 5, minSharePct: 5 })
assert.equal(six.totalBytes, 10_000)
assert.equal(six.namedBytes, 600)
assert.equal(six.unclassifiedBytes, 9400)
const google = six.services?.find((s) => s.id === "svc:google")
const otherSix = six.services?.find((s) => s.id === "svc:other")
assert.ok(google, "Google ≥ 5%")
assert.ok(google.share >= 0.05)
assert.ok(otherSix, "остаток — Прочее")
assert.ok(google.share >= 0.05 && google.share < 0.1)
assert.ok(otherSix.share >= 0.9)
const googleEdge = six.serviceEdges?.find((e) => e.toId === "svc:google" && e.fromId === "9")
assert.ok(googleEdge)
assert.equal(googleEdge.clientName, "Alice")
assert.equal((six.serviceEdges ?? []).reduce((n, e) => n + e.bytes, 0), 10_000)
} finally {
resetFlowRingsForTests()
resetIfaceCacheForTests()
@@ -309,9 +330,14 @@ try {
resetFlowMapHopsCacheForTests()
const four = await buildFlowMapHops({ minutes: 5, minSharePct: 5 })
assert.equal(four.totalBytes, 10_000)
assert.equal(four.namedBytes, 400)
assert.equal(four.unclassifiedBytes, 9600)
const googleFour = four.services?.find((s) => s.id === "svc:google")
assert.ok(googleFour, "единственный бренд виден при 4% от окна")
assert.ok(googleFour.share >= 0.99, "доля среди брендов ≈ 1")
const otherFour = four.services?.find((s) => s.id === "svc:other")
assert.ok(googleFour, "бренд виден при 4% от окна (MIN_NODES)")
assert.ok(otherFour, "Прочее держит остаток окна")
assert.ok(googleFour.share < 0.1, "доля от totalBytes, не от named")
assert.ok(otherFour.share > 0.9)
resetFlowMapHopsCacheForTests()
const off = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
assert.ok(off.services?.some((s) => s.id === "svc:google"), "порог 0 показывает Google")
@@ -340,10 +366,13 @@ try {
const two = await buildFlowMapHops({ minutes: 5, minSharePct: 5 })
const googleTwo = two.services?.find((s) => s.id === "svc:google")
const cfTwo = two.services?.find((s) => s.id === "svc:cloudflare")
const otherTwo = two.services?.find((s) => s.id === "svc:other")
assert.ok(googleTwo, "Google среди брендов")
assert.ok(cfTwo, "Cloudflare среди брендов")
assert.ok(googleTwo.share >= 0.05)
assert.ok(cfTwo.share >= 0.05)
assert.ok(otherTwo, "Прочее")
assert.ok(googleTwo.share > 0.03 && googleTwo.share < 0.05)
assert.ok(cfTwo.share > 0.03 && cfTwo.share < 0.05)
assert.ok(otherTwo.share > 0.9)
} finally {
resetFlowRingsForTests()
resetIfaceCacheForTests()
@@ -373,6 +402,43 @@ try {
resetRipeCacheForTests()
}
resetFlowRingsForTests()
resetIfaceCacheForTests()
resetRipeCacheForTests()
disableRipeEnqueueForTests()
seedFlowTopologyForTests(topo)
rememberServerIfaces(7, [
{ ".id": "*2", name: "gre-client" },
{ ".id": "*3", name: "gre-jh-en" },
])
ingestParsedFlowsForServerForTests(7, [
payloadFlow("64.233.161.1", 10_000),
payloadFlow("203.0.113.50", 20_000),
])
try {
resetFlowMapHopsCacheForTests()
const leak = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
const hop = leak.hops.find((h) => h.kind === "gre" && h.fromId === "7" && h.toId === "9")
assert.ok(hop, "GRE hop JH→EN")
assert.equal(hop.bytes, 30_000)
assert.equal(leak.totalBytes, 30_000)
assert.equal(leak.namedBytes, 10_000)
assert.equal(leak.unclassifiedBytes, 20_000)
const yt = leak.services?.find((s) => s.id === "svc:youtube")
const other = leak.services?.find((s) => s.id === "svc:other")
assert.ok(yt, "64.233:443 без RIPE → YouTube")
assert.ok(other, "dest без бренда → Прочее")
assert.equal(yt.bytes, 10_000)
assert.equal(other.bytes, 20_000)
assert.ok(Math.abs(yt.share - 10 / 30) < 0.01)
assert.ok(Math.abs(other.share - 20 / 30) < 0.01)
assert.equal((leak.serviceEdges ?? []).reduce((n, e) => n + e.bytes, 0), hop.bytes)
} finally {
resetFlowRingsForTests()
resetIfaceCacheForTests()
resetRipeCacheForTests()
}
const smallBrands: Array<{ ip: string; asn: number; holder: string; bytes: number; id: string }> = [
{ ip: "203.0.113.1", asn: 714, holder: "APPLE-ENGINEERING", bytes: 400, id: "svc:apple" },
{ ip: "203.0.113.2", asn: 36459, holder: "GITHUB", bytes: 390, id: "svc:github" },
@@ -554,9 +620,9 @@ ingestParsedFlowsForServerForTests(7, [
try {
resetFlowMapHopsCacheForTests()
const rev = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
assert.ok(rev.services?.some((s) => s.id === "svc:google"), "реверс Google:443 → 10.x")
assert.ok(rev.services?.some((s) => s.id === "svc:youtube"), "реверс googlevideo:443 → YouTube")
assert.ok(rev.services?.some((s) => s.id === "svc:cloudflare"), "реверс Cloudflare:443 → 10.x")
assert.ok(rev.serviceEdges?.some((e) => e.toId === "svc:google" && e.fromId === "9"))
assert.ok(rev.serviceEdges?.some((e) => e.toId === "svc:youtube" && e.fromId === "9"))
} finally {
seedFlowTopologyForTests(null)
resetFlowRingsForTests()
@@ -632,13 +698,13 @@ ingestParsedFlowsForServerForTests(7, [
try {
resetFlowMapHopsCacheForTests()
const wanOnly = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
const googleEdge = wanOnly.serviceEdges?.find((e) => e.toId === "svc:google")
assert.ok(googleEdge, "Google WAN без GRE payload")
const googleEdge = wanOnly.serviceEdges?.find((e) => e.toId === "svc:youtube")
assert.ok(googleEdge, "YouTube WAN без GRE payload")
assert.equal(googleEdge.fromId, "9", "единственный EN, даже без nextHop")
assert.ok(googleEdge.bps > 0, "скорость на hop EN→сервис")
assert.ok(!(wanOnly.serviceEdges ?? []).some((e) => e.fromId === "7"), "нет пунктира с JH")
const googlePath = wanOnly.servicePaths?.find((p) => p.serviceId === "svc:google")
assert.ok(googlePath, "путь WAN Google")
const googlePath = wanOnly.servicePaths?.find((p) => p.serviceId === "svc:youtube")
assert.ok(googlePath, "путь WAN YouTube")
assert.equal(googlePath.viaId, "7", "via = JH exporter")
assert.equal(googlePath.enId, "9", "якорь EN")
assert.ok(googlePath.bps > 0, "скорость на пути клиента")
+18 -11
View File
@@ -3,12 +3,12 @@ import type { FlowMapHop, FlowMapHopsDto, FlowMapService, FlowMapServiceEdge, Fl
import { db } from "../db/index.js"
import { userInterfaceBindings } from "../db/schema.js"
import { flowRowMatchesFilter } from "./traffic-flow-apps.js"
import { mapServiceNodeId } from "./traffic-flow-brands.js"
import { OTHER_SERVICE, mapServiceNodeId } from "./traffic-flow-brands.js"
import { dedupFlowRowsAcrossExporters, 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 { classifyInternetBrand, destCtxForIface } from "./traffic-flow-dest.js"
import { destCtxForIface, mapInternetBrand } from "./traffic-flow-dest.js"
import { pickInternetDest } from "./traffic-flow-ip.js"
import { resolveFlowIp } from "./traffic-flow-geoip.js"
import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js"
@@ -98,7 +98,7 @@ export function clampMapServiceMinSharePct(n: unknown): number {
return Math.min(100, Math.max(0, v))
}
/** Доля среди именованных брендов; порог ИЛИ топ-N, затем cap. */
/** Доля от payload окна; порог ИЛИ топ-N, затем cap. */
export function pickMapServices(ranked: FlowMapService[], minSharePct: number): FlowMapService[] {
if (minSharePct <= 0) return ranked.slice(0, MAP_SERVICE_NODE_CAP)
const minShare = minSharePct / 100
@@ -346,9 +346,9 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
natDstPort: r.natDstPort,
}),
)
if (!dest) continue
const destKey = dest || "__other__"
const client = resolveMapClient(topo, r.serverId, inName, outName)
const prevDst = dstAcc.get(dest)
const prevDst = dstAcc.get(destKey)
if (prevDst) {
prevDst.bytes += r.bytes
bumpFrom(prevDst, String(r.serverId), r.bytes, client)
@@ -361,7 +361,7 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
fromBytes: new Map(),
}
bumpFrom(acc, String(r.serverId), r.bytes, client)
dstAcc.set(dest, acc)
dstAcc.set(destKey, acc)
}
}
@@ -416,9 +416,10 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
}
for (const [dst, acc] of dstAcc) {
const ripe = resolveFlowIp(dst)
const classified = classifyInternetBrand(dst, acc.proto, acc.dstPort, acc.srcPort, ripe)
if (!classified) continue
const ripe = dst && dst !== "__other__" ? resolveFlowIp(dst) : null
const classified = dst && dst !== "__other__"
? mapInternetBrand(dst, acc.proto, acc.dstPort, acc.srcPort, ripe)
: { service: OTHER_SERVICE, category: OTHER_SERVICE }
const toId = mapServiceNodeId(classified.service)
const prevSvc = svcTotals.get(toId)
if (prevSvc) prevSvc.bytes += acc.bytes
@@ -474,7 +475,11 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
}
}
const namedBytes = [...svcTotals.values()].reduce((n, s) => n + s.bytes, 0)
const namedBytes = [...svcTotals.values()]
.filter((s) => s.label !== OTHER_SERVICE)
.reduce((n, s) => n + s.bytes, 0)
const unclassifiedBytes = Math.max(0, totalBytes - namedBytes)
const shareBase = totalBytes > 0 ? totalBytes : namedBytes
const services = pickMapServices(
[...svcTotals.entries()]
.map(([id, s]) => ({
@@ -483,7 +488,7 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
category: s.category,
bytes: s.bytes,
bps: (s.bytes * 8) / windowSec,
share: namedBytes > 0 ? s.bytes / namedBytes : 0,
share: shareBase > 0 ? s.bytes / shareBase : 0,
}))
.sort((a, b) => b.bytes - a.bytes),
minSharePct,
@@ -531,6 +536,8 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
rangeMinutes: q.minutes,
windowSec,
totalBytes,
namedBytes,
unclassifiedBytes,
services,
serviceEdges,
servicePaths,
+36 -2
View File
@@ -126,6 +126,9 @@ async function ensureIpfixFields(client: MikrotikClient): Promise<void> {
await client.post("/ip/traffic-flow/ipfix/set", body)
}
/** Максимум flows в RAM (docs: overflow обрезает новые 5-tuple). */
export const FLOW_CACHE_ENTRIES = "256k"
async function ensureTrafficFlow(
client: MikrotikClient,
collectorIp: string,
@@ -134,6 +137,7 @@ async function ensureTrafficFlow(
const body = toRosBody({
enabled: "yes",
interfaces: "all",
"cache-entries": FLOW_CACHE_ENTRIES,
"active-flow-timeout": "1m",
"inactive-flow-timeout": "15s",
})
@@ -167,6 +171,23 @@ async function ensureTrafficFlow(
await client.put("/ip/traffic-flow/target", targetBody)
}
/** GRE allow-fast-path=no: inner пакеты идут через CPU и попадают в Traffic Flow. Нагрузка на CPU. */
export async function ensureGreSlowPath(client: MikrotikClient): Promise<number> {
const rows = asRosArray<Record<string, unknown>>(await client.get("/interface/gre"))
let patched = 0
for (const row of rows) {
const id = rosRowId(row)
if (!id) continue
const current = String(row["allow-fast-path"] ?? "true").toLowerCase()
if (current === "false" || current === "no") continue
await patchRosPath(client, `/interface/gre/${encodeRosId(id)}`, toRosBody({
"allow-fast-path": "no",
}))
patched += 1
}
return patched
}
export function usablePublicHost(raw: string | undefined): string {
if (!raw) return ""
const host = raw.split(",")[0]?.trim().replace(/^\[/, "").replace(/\]:\d+$/, "").split(":")[0]?.trim() ?? ""
@@ -180,7 +201,7 @@ export function usablePublicHost(raw: string | undefined): string {
export async function applyFlowOverlay(
serverIdRaw: string | number,
opts?: { publicEndpoint?: string; requestHost?: string },
opts?: { publicEndpoint?: string; requestHost?: string; disableGreFastPath?: boolean },
): Promise<TrafficFlowOverlayResult> {
const steps: string[] = []
const keys = await ensureHostKeys()
@@ -275,7 +296,20 @@ export async function applyFlowOverlay(
}
await ensureTrafficFlow(client, settings.collectorIp, settings.flowListenPort)
steps.push(`Traffic Flow → ${settings.collectorIp}:${settings.flowListenPort} ipfix (src auto)`)
steps.push(`Traffic Flow → ${settings.collectorIp}:${settings.flowListenPort} ipfix (src auto, cache ${FLOW_CACHE_ENTRIES})`)
if (opts?.disableGreFastPath) {
try {
const n = await ensureGreSlowPath(client)
steps.push(
n > 0
? `GRE allow-fast-path=no (${n}) — inner IPFIX через CPU`
: "GRE already allow-fast-path=no",
)
} catch {
steps.push("GRE allow-fast-path не изменён (нет /interface/gre)")
}
}
const listed = await listWireGuardInterfaces({ serverId: String(server.id), includePrivateKey: false })
const created = listed.interfaces.find((i) => i.name === IFACE_NAME)
@@ -1,6 +1,6 @@
import assert from "node:assert/strict"
import { parseFlowPacket, protoName, resetFlowTemplatesForTests, templateExporterCountForTests } from "./traffic-flow-parse.js"
import { allocateOverlayAddress, FLOW_TARGET_SRC_AUTO, usablePublicHost } from "./traffic-flow-overlay.js"
import { allocateOverlayAddress, FLOW_CACHE_ENTRIES, FLOW_TARGET_SRC_AUTO, usablePublicHost } from "./traffic-flow-overlay.js"
function netflowV5One(): Buffer {
const buf = Buffer.alloc(24 + 48)
@@ -38,6 +38,7 @@ assert.equal(usablePublicHost("192.168.1.10"), "")
assert.equal(usablePublicHost("mm.example.com:443"), "mm.example.com")
assert.equal(usablePublicHost("203.0.113.10"), "203.0.113.10")
assert.equal(FLOW_TARGET_SRC_AUTO, "0.0.0.0")
assert.equal(FLOW_CACHE_ENTRIES, "256k")
resetFlowTemplatesForTests()
{
+15 -2
View File
@@ -2,7 +2,7 @@
import { useEffect, useMemo, useState } from "react"
import { toast } from "sonner"
import { FormField } from "@/components/form-kit"
import { FormField, FormToggle } from "@/components/form-kit"
import { Alert, AlertDescription, AlertTitle } from "@/components/reui/alert"
import { Frame, FramePanel } from "@/components/reui/frame"
import { CodeBlock, downloadText } from "@/components/reui-kit/code-export-sheet"
@@ -54,12 +54,14 @@ function FlowOverlaySheet({
const [result, setResult] = useState<TrafficFlowOverlayResult | null>(null)
const [copied, setCopied] = useState(false)
const [tab, setTab] = useState("linux")
const [disableGreFastPath, setDisableGreFastPath] = useState(false)
useEffect(() => {
if (!open) return
setResult(null)
setCopied(false)
setTab("linux")
setDisableGreFastPath(false)
const first = jumpHosts[0]
const nextId = first ? String(first.id) : ""
setServerId(nextId)
@@ -84,7 +86,9 @@ function FlowOverlaySheet({
if (!serverId || !endpoint.trim()) return
setBusy(true)
try {
const res = await applyTrafficFlowOverlay(backendUrl, serverId, endpoint.trim())
const res = await applyTrafficFlowOverlay(backendUrl, serverId, endpoint.trim(), {
disableGreFastPath,
})
setResult(res)
setTab(res.hostFiles[0]?.id ?? "linux")
toast.success(`wg-flow на ${res.address}`)
@@ -151,6 +155,15 @@ function FlowOverlaySheet({
autoComplete="off"
/>
</FormField>
<FormField
label="GRE slow-path (IPFIX inner)"
hint="allow-fast-path=no на GRE этого JH. Inner YouTube попадёт в Traffic Flow, но вырастет CPU. По умолчанию выкл."
>
<div className="flex items-center gap-2">
<FormToggle checked={disableGreFastPath} onChange={setDisableGreFastPath} />
<span className="text-sm text-muted-foreground">Выключить FastPath на GRE</span>
</div>
</FormField>
{result ? (
<div className="flex min-h-0 flex-col gap-4">
<Alert variant="success">
+4
View File
@@ -47,6 +47,8 @@ export const trafficFlowSettingsPatchSchema = z.object({
export const trafficFlowOverlayRequestSchema = z.object({
serverId: z.union([z.string(), z.number()]),
publicEndpoint: z.string().optional(),
/** GRE allow-fast-path=no: inner IPFIX через CPU. По умолчанию выкл (нагрузка на CPU). */
disableGreFastPath: z.boolean().optional(),
})
export const trafficFlowHostFileSchema = z.object({
@@ -316,6 +318,8 @@ export const flowMapHopsDtoSchema = z.object({
rangeMinutes: z.number().int().positive(),
windowSec: z.number().positive(),
totalBytes: z.number().nonnegative().optional(),
namedBytes: z.number().nonnegative().optional(),
unclassifiedBytes: z.number().nonnegative().optional(),
services: z.array(flowMapServiceDtoSchema).optional(),
serviceEdges: z.array(flowMapServiceEdgeDtoSchema).optional(),
servicePaths: z.array(flowMapServicePathDtoSchema).optional(),
+6 -1
View File
@@ -46,10 +46,15 @@ export async function applyTrafficFlowOverlay(
baseUrl: string,
serverId: string | number,
publicEndpoint?: string,
opts?: { disableGreFastPath?: boolean },
): Promise<TrafficFlowOverlayResult> {
return requestJson(baseUrl, "/api/traffic/flow-overlay", {
method: "POST",
body: JSON.stringify({ serverId, publicEndpoint }),
body: JSON.stringify({
serverId,
publicEndpoint,
...(opts?.disableGreFastPath ? { disableGreFastPath: true } : {}),
}),
})
}
+1 -1
View File
File diff suppressed because one or more lines are too long