feat(traffic-flow): enhance traffic flow analytics and overlay configuration
- 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 <[email protected]>
This commit is contained in:
@@ -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">
|
||||
|
||||
@@ -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")
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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, "скорость на пути клиента")
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
{
|
||||
|
||||
@@ -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">
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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 } : {}),
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
File diff suppressed because one or more lines are too long
Reference in New Issue
Block a user