fix(network-map): убрать тяжёлую классификацию из poll карты

Не сканировать RIPE и каталог по каждой строке окна; сервисы по уникальным dst и кэшу ASN. Порог доли на карте настраиваемый и отключаемый.

Co-authored-by: Cursor <[email protected]>
This commit is contained in:
Denozordec
2026-09-07 14:17:04 +07:00
co-authored by Cursor
parent db64621122
commit 0dbc0ebe81
10 changed files with 315 additions and 54 deletions
+17 -5
View File
@@ -938,6 +938,7 @@ export default function NetworkMapPage() {
const [mapHops, setMapHops] = useState<FlowMapHop[]>([]) const [mapHops, setMapHops] = useState<FlowMapHop[]>([])
const [mapServices, setMapServices] = useState<FlowMapService[]>([]) const [mapServices, setMapServices] = useState<FlowMapService[]>([])
const [mapServiceEdges, setMapServiceEdges] = useState<FlowMapServiceEdge[]>([]) const [mapServiceEdges, setMapServiceEdges] = useState<FlowMapServiceEdge[]>([])
const [mapSharePct, setMapSharePct] = useState(5)
/** FQDN из GRE outer → IPv4 (ответ POST /api/network/resolve-hosts), для матчинга с WAN. */ /** FQDN из GRE outer → IPv4 (ответ POST /api/network/resolve-hosts), для матчинга с WAN. */
const [greResolvedIpv4ByHost, setGreResolvedIpv4ByHost] = useState<Record<string, string>>({}) const [greResolvedIpv4ByHost, setGreResolvedIpv4ByHost] = useState<Record<string, string>>({})
const [dataError, setDataError] = useState<string | null>(null) const [dataError, setDataError] = useState<string | null>(null)
@@ -1032,6 +1033,7 @@ export default function NetworkMapPage() {
setMapHops([]) setMapHops([])
setMapServices(MOCK_MAP_SERVICES) setMapServices(MOCK_MAP_SERVICES)
setMapServiceEdges(MOCK_MAP_SERVICE_EDGES) setMapServiceEdges(MOCK_MAP_SERVICE_EDGES)
setMapSharePct(5)
setDataError(null) setDataError(null)
}) })
return return
@@ -1106,6 +1108,7 @@ export default function NetworkMapPage() {
setMapHops([]) setMapHops([])
setMapServices(MOCK_MAP_SERVICES) setMapServices(MOCK_MAP_SERVICES)
setMapServiceEdges(MOCK_MAP_SERVICE_EDGES) setMapServiceEdges(MOCK_MAP_SERVICE_EDGES)
setMapSharePct(5)
}) })
return return
} }
@@ -1118,25 +1121,29 @@ export default function NetworkMapPage() {
return return
} }
let cancelled = false let cancelled = false
let ac: AbortController | null = null
const tick = () => { const tick = () => {
apiFetch<FlowMapHopsDto>("/api/traffic/flow/map-hops?range=5m") ac?.abort()
ac = new AbortController()
apiFetch<FlowMapHopsDto>("/api/traffic/flow/map-hops?range=5m", { signal: ac.signal })
.then((res) => { .then((res) => {
if (cancelled) return if (cancelled) return
setMapHops(res.hops ?? []) setMapHops(res.hops ?? [])
setMapServices(res.services ?? []) setMapServices(res.services ?? [])
setMapServiceEdges(res.serviceEdges ?? []) setMapServiceEdges(res.serviceEdges ?? [])
if (res.mapServiceMinSharePct != null) setMapSharePct(res.mapServiceMinSharePct)
}) })
.catch(() => { .catch((err: unknown) => {
if (cancelled) return if (cancelled) return
setMapHops([]) const name = err instanceof Error ? err.name : ""
setMapServices([]) if (name === "AbortError") return
setMapServiceEdges([])
}) })
} }
tick() tick()
const id = window.setInterval(tick, 4000) const id = window.setInterval(tick, 4000)
return () => { return () => {
cancelled = true cancelled = true
ac?.abort()
window.clearInterval(id) window.clearInterval(id)
} }
}, [useLiveData, showNetflow, showServices, apiFetch]) }, [useLiveData, showNetflow, showServices, apiFetch])
@@ -1736,6 +1743,11 @@ export default function NetworkMapPage() {
</span> </span>
</button> </button>
))} ))}
<p className="px-3 pt-1.5 pb-1 text-[10px] text-muted-foreground leading-snug">
{mapSharePct > 0
? `Порог доли сервиса ≥ ${mapSharePct}% · Настройки → NetFlow`
: "Порог доли выключен (все бренды, макс. 20) · Настройки → NetFlow"}
</p>
{(Object.keys(nodePositions).length > 0 || Object.keys(satPositions).length > 0) && ( {(Object.keys(nodePositions).length > 0 || Object.keys(satPositions).length > 0) && (
<div className="border-t border-border/50 mt-1 pt-1"> <div className="border-t border-border/50 mt-1 pt-1">
<button <button
+8
View File
@@ -146,6 +146,7 @@ CREATE TABLE IF NOT EXISTS traffic_flow_settings (
hub_server_id INTEGER, hub_server_id INTEGER,
retention_hours INTEGER NOT NULL DEFAULT 24, retention_hours INTEGER NOT NULL DEFAULT 24,
top_n INTEGER NOT NULL DEFAULT 200, top_n INTEGER NOT NULL DEFAULT 200,
map_service_min_share_pct REAL NOT NULL DEFAULT 5,
last_datagram_at TEXT, last_datagram_at TEXT,
last_exporter_ip TEXT, last_exporter_ip TEXT,
last_error TEXT, last_error TEXT,
@@ -834,6 +835,13 @@ SELECT 1, 0, '10.255.254.1', 4739, 51821, '10.255.254.0/24'
WHERE NOT EXISTS (SELECT 1 FROM traffic_flow_settings WHERE id = 1); WHERE NOT EXISTS (SELECT 1 FROM traffic_flow_settings WHERE id = 1);
`) `)
{
const flowSettingsCols = sqlite.prepare(`PRAGMA table_info('traffic_flow_settings')`).all() as Array<{ name?: string }>
if (!flowSettingsCols.some((c) => c.name === "map_service_min_share_pct")) {
sqlite.exec(`ALTER TABLE traffic_flow_settings ADD COLUMN map_service_min_share_pct REAL NOT NULL DEFAULT 5`)
}
}
sqlite.exec(` sqlite.exec(`
INSERT INTO uptime_settings (id, enabled, interval_sec, retention_days) INSERT INTO uptime_settings (id, enabled, interval_sec, retention_days)
SELECT 1, 1, 15, 14 SELECT 1, 1, 15, 14
+1
View File
@@ -173,6 +173,7 @@ export const trafficFlowSettings = sqliteTable("traffic_flow_settings", {
hubServerId: integer("hub_server_id"), hubServerId: integer("hub_server_id"),
retentionHours: integer("retention_hours").notNull().default(24), retentionHours: integer("retention_hours").notNull().default(24),
topN: integer("top_n").notNull().default(200), topN: integer("top_n").notNull().default(200),
mapServiceMinSharePct: real("map_service_min_share_pct").notNull().default(5),
lastDatagramAt: text("last_datagram_at"), lastDatagramAt: text("last_datagram_at"),
lastExporterIp: text("last_exporter_ip"), lastExporterIp: text("last_exporter_ip"),
lastError: text("last_error"), lastError: text("last_error"),
@@ -4,7 +4,7 @@ import {
ingestParsedFlowsForServerForTests, ingestParsedFlowsForServerForTests,
resetFlowRingsForTests, resetFlowRingsForTests,
} from "./traffic-flow-ingest.js" } from "./traffic-flow-ingest.js"
import { buildFlowMapHops } from "./traffic-flow-map-hops.js" import { buildFlowMapHops, resetFlowMapHopsCacheForTests } from "./traffic-flow-map-hops.js"
import { seedFlowTopologyForTests, type FlowTopology } from "./traffic-flow-topology.js" import { seedFlowTopologyForTests, type FlowTopology } from "./traffic-flow-topology.js"
import { disableCatalogFetchForTests, resetFlowCatalogForTests } from "./traffic-flow-classify.js" import { disableCatalogFetchForTests, resetFlowCatalogForTests } from "./traffic-flow-classify.js"
import { import {
@@ -114,6 +114,7 @@ ingestParsedFlowsForServerForTests(3, [
]) ])
try { try {
resetFlowMapHopsCacheForTests()
const def = buildFlowMapHops({ minutes: 5 }) const def = buildFlowMapHops({ minutes: 5 })
assert.equal(def.excludeOverlayApplied, true) assert.equal(def.excludeOverlayApplied, true)
assert.equal(def.excludeMeshApplied, true) assert.equal(def.excludeMeshApplied, true)
@@ -142,6 +143,7 @@ try {
assert.ok(wan, "WAN hop from home-router") assert.ok(wan, "WAN hop from home-router")
assert.equal(wan.bytes, 3000) assert.equal(wan.bytes, 3000)
resetFlowMapHopsCacheForTests()
const withAll = buildFlowMapHops({ minutes: 5, excludeOverlay: false, excludeMesh: false }) const withAll = buildFlowMapHops({ minutes: 5, excludeOverlay: false, excludeMesh: false })
const overlayIface = withAll.hops.find((h) => h.iface === "gre-jh-en" && h.fromId === "7") const overlayIface = withAll.hops.find((h) => h.iface === "gre-jh-en" && h.fromId === "7")
assert.ok(overlayIface && overlayIface.bytes >= 5_000_000) assert.ok(overlayIface && overlayIface.bytes >= 5_000_000)
@@ -200,7 +202,8 @@ ingestParsedFlowsForServerForTests(7, [
payloadFlow("203.0.113.50", 9400), payloadFlow("203.0.113.50", 9400),
]) ])
try { try {
const six = buildFlowMapHops({ minutes: 5 }) resetFlowMapHopsCacheForTests()
const six = buildFlowMapHops({ minutes: 5, minSharePct: 5 })
assert.equal(six.totalBytes, 10_000) assert.equal(six.totalBytes, 10_000)
const google = six.services?.find((s) => s.id === "svc:google") const google = six.services?.find((s) => s.id === "svc:google")
assert.ok(google, "Google ≥ 5%") assert.ok(google, "Google ≥ 5%")
@@ -227,9 +230,13 @@ ingestParsedFlowsForServerForTests(7, [
payloadFlow("203.0.113.50", 9600), payloadFlow("203.0.113.50", 9600),
]) ])
try { try {
const four = buildFlowMapHops({ minutes: 5 }) resetFlowMapHopsCacheForTests()
const four = buildFlowMapHops({ minutes: 5, minSharePct: 5 })
assert.equal(four.totalBytes, 10_000) assert.equal(four.totalBytes, 10_000)
assert.ok(!(four.services ?? []).some((s) => s.id === "svc:google"), "Google < 5% hidden") assert.ok(!(four.services ?? []).some((s) => s.id === "svc:google"), "Google < 5% hidden")
resetFlowMapHopsCacheForTests()
const off = buildFlowMapHops({ minutes: 5, minSharePct: 0 })
assert.ok(off.services?.some((s) => s.id === "svc:google"), "порог 0 показывает Google 4%")
} finally { } finally {
resetFlowRingsForTests() resetFlowRingsForTests()
resetIfaceCacheForTests() resetIfaceCacheForTests()
@@ -260,7 +267,8 @@ ingestParsedFlowsForServerForTests(7, [
payloadFlow("203.0.113.50", 1000), payloadFlow("203.0.113.50", 1000),
]) ])
try { try {
const greOnly = buildFlowMapHops({ minutes: 5, excludeOverlay: false }) resetFlowMapHopsCacheForTests()
const greOnly = buildFlowMapHops({ minutes: 5, excludeOverlay: false, minSharePct: 0 })
assert.ok(!(greOnly.services ?? []).some((s) => s.label === "GRE"), "GRE is not a destination service") assert.ok(!(greOnly.services ?? []).some((s) => s.label === "GRE"), "GRE is not a destination service")
} finally { } finally {
seedFlowTopologyForTests(null) seedFlowTopologyForTests(null)
+127 -32
View File
@@ -2,20 +2,23 @@ import { eq } from "drizzle-orm"
import type { FlowMapHop, FlowMapHopsDto, FlowMapService, FlowMapServiceEdge } from "@mmapp/contracts/traffic-flow" import type { FlowMapHop, FlowMapHopsDto, FlowMapService, FlowMapServiceEdge } from "@mmapp/contracts/traffic-flow"
import { db } from "../db/index.js" import { db } from "../db/index.js"
import { servers, userInterfaceBindings } from "../db/schema.js" import { servers, userInterfaceBindings } from "../db/schema.js"
import { flowRowMatchesFilter } from "./traffic-flow-apps.js" import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
import { import {
isNamedInternetService, isNamedInternetService,
lookupBrand,
mapServiceNodeId, mapServiceNodeId,
} from "./traffic-flow-brands.js" } from "./traffic-flow-brands.js"
import { classifyFlowDst, refreshFlowCatalogInBackground } from "./traffic-flow-classify.js"
import { dedupFlowRowsMaxBytes } from "./traffic-flow-dedup.js" import { dedupFlowRowsMaxBytes } from "./traffic-flow-dedup.js"
import { getFlowListenerState, listFlowRowsForWindow } from "./traffic-flow-ingest.js" import { getFlowListenerState, listFlowRowsForWindow } from "./traffic-flow-ingest.js"
import { resolveIfaceName } from "./traffic-flow-ifaces.js" import { resolveIfaceName } from "./traffic-flow-ifaces.js"
import { classifyFlowPlane, shouldKeepPlane } from "./traffic-flow-planes.js" import { classifyFlowPlane, shouldKeepPlane } from "./traffic-flow-planes.js"
import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js" import { lookupRipeCached, type FlowIpMeta } from "./traffic-flow-ripe.js"
import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js"
import { loadFlowTopology, resolveEn } from "./traffic-flow-topology.js" import { loadFlowTopology, resolveEn } from "./traffic-flow-topology.js"
export const MAP_SERVICE_SHARE_THRESHOLD = 0.05 export const DEFAULT_MAP_SERVICE_MIN_SHARE_PCT = 5
export const MAP_SERVICE_NODE_CAP = 20
const HOPS_CACHE_TTL_MS = 2000
export interface FlowMapHopsQuery { export interface FlowMapHopsQuery {
minutes: number minutes: number
@@ -25,6 +28,8 @@ export interface FlowMapHopsQuery {
dedup?: boolean dedup?: boolean
excludeMesh?: boolean excludeMesh?: boolean
excludeOverlay?: boolean excludeOverlay?: boolean
/** Переопределение порога (тесты). Иначе из настроек NetFlow. */
minSharePct?: number
} }
interface HopAcc { interface HopAcc {
@@ -39,6 +44,39 @@ interface HopAcc {
bytesRev: number bytesRev: number
} }
interface DstAcc {
bytes: number
proto: number
dstPort: number
srcPort: number
fromBytes: Map<string, number>
}
let hopsCache: { key: string; at: number; dto: FlowMapHopsDto } | null = null
export function resetFlowMapHopsCacheForTests(): void {
hopsCache = null
}
export function clampMapServiceMinSharePct(n: unknown): number {
const v = typeof n === "number" ? n : Number(n)
if (!Number.isFinite(v)) return DEFAULT_MAP_SERVICE_MIN_SHARE_PCT
return Math.min(100, Math.max(0, v))
}
function hopsQueryKey(q: FlowMapHopsQuery, minSharePct: number): string {
return JSON.stringify({
minutes: q.minutes,
serverId: q.serverId ?? null,
userId: q.userId ?? null,
iface: q.iface ?? null,
dedup: q.dedup !== false,
excludeMesh: q.excludeMesh !== false,
excludeOverlay: q.excludeOverlay !== false,
minSharePct,
})
}
function userIfaceAllow(userId: string): Map<number, Set<string>> | null { function userIfaceAllow(userId: string): Map<number, Set<string>> | null {
if (!userId) return null if (!userId) return null
const binds = db.select().from(userInterfaceBindings).where(eq(userInterfaceBindings.userId, userId)).all() const binds = db.select().from(userInterfaceBindings).where(eq(userInterfaceBindings.userId, userId)).all()
@@ -89,8 +127,36 @@ function toHop(a: HopAcc, windowSec: number): FlowMapHop {
} }
} }
/** Hop-rates для карты сети: те же фильтры, что у общего NetFlow (dedup / mesh / overlay). */ /** Имя бренда без каталога EvoBGP — только ASN/CIDR кэш + proto. */
export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto { function classifyMapDstLite(
dst: string,
proto: number,
dstPort: number,
srcPort: number,
ripe: FlowIpMeta | null,
): { service: string; category: string } | null {
if (proto === 47 || proto === 50) return null
const app = applicationName(proto, dstPort, srcPort)
if (app === "WireGuard" || app === "DNS" || app === "SSH" || app === "BGP") return null
if (/youtube/i.test(ripe?.holder ?? "")) {
return { service: "YouTube", category: "Видео / стриминг" }
}
const brand = lookupBrand(dst, ripe?.asn ?? 0)
if (!brand || !isNamedInternetService(brand.service, brand.category)) return null
return brand
}
function resolveMinSharePct(q: FlowMapHopsQuery): number {
if (q.minSharePct != null) return clampMapServiceMinSharePct(q.minSharePct)
try {
const row = getTrafficFlowSettingsRow() as { mapServiceMinSharePct?: number }
return clampMapServiceMinSharePct(row.mapServiceMinSharePct ?? DEFAULT_MAP_SERVICE_MIN_SHARE_PCT)
} catch {
return DEFAULT_MAP_SERVICE_MIN_SHARE_PCT
}
}
function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number): FlowMapHopsDto {
const windowSec = Math.max(60, q.minutes * 60) const windowSec = Math.max(60, q.minutes * 60)
const raw = listFlowRowsForWindow(q.minutes) const raw = listFlowRowsForWindow(q.minutes)
const allow = q.userId ? userIfaceAllow(q.userId) : null const allow = q.userId ? userIfaceAllow(q.userId) : null
@@ -122,13 +188,9 @@ export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
const working = wantDedup ? dedupFlowRowsMaxBytes(matched) : matched const working = wantDedup ? dedupFlowRowsMaxBytes(matched) : matched
const hops = new Map<string, HopAcc>() const hops = new Map<string, HopAcc>()
const svcTotals = new Map<string, { label: string; category: string; bytes: number }>() const dstAcc = new Map<string, DstAcc>()
const svcEdges = new Map<string, { fromId: string; toId: string; bytes: number; bytesFwd: number; bytesRev: number }>()
const dsts = new Set<string>()
let totalBytes = 0 let totalBytes = 0
refreshFlowCatalogInBackground()
for (const r of working) { for (const r of working) {
const inRes = resolveIfaceName(r.serverId, r.inIface) const inRes = resolveIfaceName(r.serverId, r.inIface)
const outRes = resolveIfaceName(r.serverId, r.outIface) const outRes = resolveIfaceName(r.serverId, r.outIface)
@@ -215,37 +277,53 @@ export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
} }
totalBytes += r.bytes totalBytes += r.bytes
dsts.add(r.dst) const svcFromId = String((enOut ?? enIn)?.id ?? r.serverId)
const ripe = lookupRipeCached(r.dst) const prevDst = dstAcc.get(r.dst)
const classified = classifyFlowDst(r.dst, r.proto, r.dstPort, r.srcPort, ripe) if (prevDst) {
if (isNamedInternetService(classified.service, classified.category)) { prevDst.bytes += r.bytes
const toId = mapServiceNodeId(classified.service) prevDst.fromBytes.set(svcFromId, (prevDst.fromBytes.get(svcFromId) ?? 0) + r.bytes)
const prevSvc = svcTotals.get(toId) } else {
if (prevSvc) prevSvc.bytes += r.bytes dstAcc.set(r.dst, {
else svcTotals.set(toId, { label: classified.service, category: classified.category, bytes: r.bytes }) bytes: r.bytes,
proto: r.proto,
dstPort: r.dstPort,
srcPort: r.srcPort,
fromBytes: new Map([[svcFromId, r.bytes]]),
})
}
}
const svcEn = enOut ?? enIn const svcTotals = new Map<string, { label: string; category: string; bytes: number }>()
const svcFromId = svcEn ? String(svcEn.id) : fromId const svcEdges = new Map<string, { fromId: string; toId: string; bytes: number; bytesFwd: number; bytesRev: number }>()
const edgeKey = `${svcFromId}|${toId}`
for (const [dst, acc] of dstAcc) {
const ripe = lookupRipeCached(dst)
const classified = classifyMapDstLite(dst, acc.proto, acc.dstPort, acc.srcPort, ripe)
if (!classified) continue
const toId = mapServiceNodeId(classified.service)
const prevSvc = svcTotals.get(toId)
if (prevSvc) prevSvc.bytes += acc.bytes
else svcTotals.set(toId, { label: classified.service, category: classified.category, bytes: acc.bytes })
for (const [fromId, bytes] of acc.fromBytes) {
const edgeKey = `${fromId}|${toId}`
const prevEdge = svcEdges.get(edgeKey) const prevEdge = svcEdges.get(edgeKey)
if (prevEdge) { if (prevEdge) {
prevEdge.bytes += r.bytes prevEdge.bytes += bytes
prevEdge.bytesFwd += r.bytes prevEdge.bytesFwd += bytes
} else { } else {
svcEdges.set(edgeKey, { svcEdges.set(edgeKey, {
fromId: svcFromId, fromId,
toId, toId,
bytes: r.bytes, bytes,
bytesFwd: r.bytes, bytesFwd: bytes,
bytesRev: 0, bytesRev: 0,
}) })
} }
} }
} }
enqueueRipeMisses(dsts) const minShare = minSharePct / 100
let services: FlowMapService[] = [...svcTotals.entries()]
const services: FlowMapService[] = [...svcTotals.entries()]
.map(([id, s]) => ({ .map(([id, s]) => ({
id, id,
label: s.label, label: s.label,
@@ -254,8 +332,11 @@ export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
bps: (s.bytes * 8) / windowSec, bps: (s.bytes * 8) / windowSec,
share: totalBytes > 0 ? s.bytes / totalBytes : 0, share: totalBytes > 0 ? s.bytes / totalBytes : 0,
})) }))
.filter((s) => s.share >= MAP_SERVICE_SHARE_THRESHOLD)
.sort((a, b) => b.bytes - a.bytes) .sort((a, b) => b.bytes - a.bytes)
if (minSharePct > 0) {
services = services.filter((s) => s.share >= minShare)
}
services = services.slice(0, MAP_SERVICE_NODE_CAP)
const keepSvc = new Set(services.map((s) => s.id)) const keepSvc = new Set(services.map((s) => s.id))
const serviceEdges: FlowMapServiceEdge[] = [...svcEdges.values()] const serviceEdges: FlowMapServiceEdge[] = [...svcEdges.values()]
.filter((e) => keepSvc.has(e.toId)) .filter((e) => keepSvc.has(e.toId))
@@ -273,15 +354,29 @@ export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
return { return {
hops: [...hops.values()] hops: [...hops.values()]
.map((a) => toHop(a, windowSec)) .map((a) => toHop(a, windowSec))
.sort((a, b) => b.bytes - a.bytes), .sort((a, b) => a.bytes === b.bytes ? 0 : b.bytes - a.bytes),
live: listener.bound, live: listener.bound,
rangeMinutes: q.minutes, rangeMinutes: q.minutes,
windowSec, windowSec,
totalBytes, totalBytes,
services, services,
serviceEdges, serviceEdges,
mapServiceMinSharePct: minSharePct,
dedupApplied: wantDedup, dedupApplied: wantDedup,
excludeMeshApplied: excludeMesh, excludeMeshApplied: excludeMesh,
excludeOverlayApplied: excludeOverlay, excludeOverlayApplied: excludeOverlay,
} }
} }
/** Hop-rates для карты сети: те же фильтры, что у общего NetFlow (dedup / mesh / overlay). */
export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
const minSharePct = resolveMinSharePct(q)
const key = hopsQueryKey(q, minSharePct)
const now = Date.now()
if (hopsCache && hopsCache.key === key && now - hopsCache.at < HOPS_CACHE_TTL_MS) {
return hopsCache.dto
}
const dto = buildFlowMapHopsUncached(q, minSharePct)
hopsCache = { key, at: now, dto }
return dto
}
@@ -7,6 +7,7 @@ import {
lookupRipeCached, lookupRipeCached,
resetRipeCacheForTests, resetRipeCacheForTests,
ripeFetchCountForTests, ripeFetchCountForTests,
ripeLastCandidateCountForTests,
seedRipeCacheForTests, seedRipeCacheForTests,
setRipeFetchForTests, setRipeFetchForTests,
} from "./traffic-flow-ripe.js" } from "./traffic-flow-ripe.js"
@@ -100,4 +101,36 @@ await flushRipeQueueForTests()
assert.equal(lookupRipeCached("1.0.0.1")?.country, "US") assert.equal(lookupRipeCached("1.0.0.1")?.country, "US")
assert.equal(lookupRipeCached("1.0.0.1")?.asn, 13335) assert.equal(lookupRipeCached("1.0.0.1")?.asn, 13335)
resetRipeCacheForTests()
disableRipePersistForTests()
for (let i = 0; i < 3000; i++) {
const o2 = Math.floor(i / 256)
const o3 = i % 256
seedRipeCacheForTests({
prefix: `203.${o2}.${o3}.0/24`,
asn: 64500,
country: "NL",
lat: null,
lng: null,
holder: "NOISE",
ok: true,
fetchedAt: Date.now(),
})
}
seedRipeCacheForTests({
prefix: "8.8.8.0/24",
asn: 15169,
country: "US",
lat: null,
lng: null,
holder: "GOOGLE",
ok: true,
fetchedAt: Date.now(),
})
assert.equal(lookupRipeCached("8.8.8.8")?.asn, 15169)
assert.ok(
ripeLastCandidateCountForTests() < 8,
`index should not scan all prefixes, got ${ripeLastCandidateCountForTests()}`,
)
console.log("traffic-flow-ripe.test.ts: ok") console.log("traffic-flow-ripe.test.ts: ok")
+79 -13
View File
@@ -1,5 +1,5 @@
import { sqliteDatabase } from "../db/index.js" import { sqliteDatabase } from "../db/index.js"
import { ipInCidrV4, ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js" import { ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
import { resolveRipeCountry } from "./traffic-flow-brands.js" import { resolveRipeCountry } from "./traffic-flow-brands.js"
export interface FlowIpMeta { export interface FlowIpMeta {
@@ -28,6 +28,18 @@ const queue: string[] = []
const queued = new Set<string>() const queued = new Set<string>()
const recentFetches: number[] = [] const recentFetches: number[] = []
interface RipeIndexed {
entry: FlowIpMeta
net: number
mask: number
prefixLen: number
}
/** /24 → кандидаты с prefixLen ≥ 24. Более широкие префиксы — в `wideIndex`. */
const v24Index = new Map<number, RipeIndexed[]>()
const wideIndex: RipeIndexed[] = []
let lastCandidateCount = 0
let persistEnabled = true let persistEnabled = true
let enqueueEnabled = true let enqueueEnabled = true
let loaded = false let loaded = false
@@ -50,6 +62,9 @@ export function resetRipeCacheForTests(): void {
queue.length = 0 queue.length = 0
queued.clear() queued.clear()
recentFetches.length = 0 recentFetches.length = 0
v24Index.clear()
wideIndex.length = 0
lastCandidateCount = 0
loaded = persistEnabled ? false : true loaded = persistEnabled ? false : true
workerRunning = false workerRunning = false
fetchCount = 0 fetchCount = 0
@@ -58,10 +73,15 @@ export function resetRipeCacheForTests(): void {
} }
export function seedRipeCacheForTests(entry: FlowIpMeta): void { export function seedRipeCacheForTests(entry: FlowIpMeta): void {
mem.set(entry.prefix, { ...entry }) remember(entry)
loaded = true loaded = true
} }
/** Сколько CIDR смотрели в последнем lookup (для теста индекса /24). */
export function ripeLastCandidateCountForTests(): number {
return lastCandidateCount
}
export function setRipeFetchForTests(fn: typeof fetch): void { export function setRipeFetchForTests(fn: typeof fetch): void {
fetchImpl = fn fetchImpl = fn
fetchCount = 0 fetchCount = 0
@@ -87,6 +107,48 @@ function isFresh(entry: FlowIpMeta): boolean {
return Date.now() - entry.fetchedAt < ttlMs(entry.ok) return Date.now() - entry.fetchedAt < ttlMs(entry.ok)
} }
function unindexPrefix(prefix: string): void {
const parsed = parseCidrV4(prefix)
if (!parsed) return
if (parsed.prefixLen >= 24) {
const key = parsed.net >>> 8
const list = v24Index.get(key)
if (!list) return
const next = list.filter((row) => row.entry.prefix !== prefix)
if (next.length) v24Index.set(key, next)
else v24Index.delete(key)
return
}
const idx = wideIndex.findIndex((row) => row.entry.prefix === prefix)
if (idx >= 0) wideIndex.splice(idx, 1)
}
function indexEntry(entry: FlowIpMeta): void {
const parsed = parseCidrV4(entry.prefix)
if (!parsed) return
const row: RipeIndexed = {
entry,
net: parsed.net,
mask: parsed.mask,
prefixLen: parsed.prefixLen,
}
if (parsed.prefixLen >= 24) {
const key = parsed.net >>> 8
const list = v24Index.get(key)
if (list) list.push(row)
else v24Index.set(key, [row])
return
}
wideIndex.push(row)
}
function remember(entry: FlowIpMeta): void {
const prev = mem.get(entry.prefix)
if (prev) unindexPrefix(prev.prefix)
mem.set(entry.prefix, entry)
indexEntry(entry)
}
function loadSqlite(): void { function loadSqlite(): void {
if (loaded || !persistEnabled) { if (loaded || !persistEnabled) {
loaded = true loaded = true
@@ -111,7 +173,7 @@ function loadSqlite(): void {
const fetchedAt = Date.parse(r.fetched_at) const fetchedAt = Date.parse(r.fetched_at)
const asn = Number(r.asn ?? 0) || 0 const asn = Number(r.asn ?? 0) || 0
const holder = r.holder || "" const holder = r.holder || ""
mem.set(r.prefix, { remember({
prefix: r.prefix, prefix: r.prefix,
asn, asn,
country: resolveRipeCountry(r.country || "", asn, holder) || "—", country: resolveRipeCountry(r.country || "", asn, holder) || "—",
@@ -193,20 +255,24 @@ function negative(prefix: string): FlowIpMeta {
export function lookupRipeCached(ip: string): FlowIpMeta | null { export function lookupRipeCached(ip: string): FlowIpMeta | null {
loadSqlite() loadSqlite()
const trimmed = String(ip ?? "").trim() const trimmed = String(ip ?? "").trim()
lastCandidateCount = 0
if (!trimmed) return null if (!trimmed) return null
if (isNonPublicIp(trimmed)) { if (isNonPublicIp(trimmed)) {
return negative(`${trimmed.includes(":") ? trimmed : trimmed}/32`) return negative(`${trimmed.includes(":") ? trimmed : trimmed}/32`)
} }
const addr = ipv4ToInt(trimmed)
if (addr == null) return null
const bucket = v24Index.get(addr >>> 8)
const candidates = bucket ? bucket.concat(wideIndex) : wideIndex
lastCandidateCount = candidates.length
let best: FlowIpMeta | null = null let best: FlowIpMeta | null = null
let bestLen = -1 let bestLen = -1
for (const entry of mem.values()) { for (const row of candidates) {
if (!isFresh(entry)) continue if (!isFresh(row.entry)) continue
const parsed = parseCidrV4(entry.prefix) if (((addr & row.mask) >>> 0) !== row.net) continue
if (!parsed) continue if (row.prefixLen > bestLen) {
if (!ipInCidrV4(trimmed, entry.prefix)) continue best = row.entry
if (parsed.prefixLen > bestLen) { bestLen = row.prefixLen
best = entry
bestLen = parsed.prefixLen
} }
} }
return best return best
@@ -316,13 +382,13 @@ async function resolveIp(ip: string): Promise<FlowIpMeta | null> {
ok: Boolean(asn || country), ok: Boolean(asn || country),
fetchedAt: Date.now(), fetchedAt: Date.now(),
} }
mem.set(prefix, entry) remember(entry)
persist(entry) persist(entry)
return entry return entry
} catch { } catch {
const prefix = `${ip}/32` const prefix = `${ip}/32`
const entry = negative(prefix) const entry = negative(prefix)
mem.set(prefix, entry) remember(entry)
persist(entry) persist(entry)
return entry return entry
} finally { } finally {
@@ -53,6 +53,7 @@ export function toTrafficFlowSettingsDto(
hubServerId: row.hubServerId ?? null, hubServerId: row.hubServerId ?? null,
retentionHours: row.retentionHours, retentionHours: row.retentionHours,
topN: row.topN, topN: row.topN,
mapServiceMinSharePct: Number(row.mapServiceMinSharePct ?? 5),
lastDatagramAt: row.lastDatagramAt ?? null, lastDatagramAt: row.lastDatagramAt ?? null,
lastExporterIp: row.lastExporterIp ?? null, lastExporterIp: row.lastExporterIp ?? null,
lastError: row.lastError || null, lastError: row.lastError || null,
@@ -75,6 +76,9 @@ export function updateTrafficFlowSettings(patch: TrafficFlowSettingsPatch) {
hubServerId: patch.hubServerId === undefined ? row.hubServerId : patch.hubServerId, hubServerId: patch.hubServerId === undefined ? row.hubServerId : patch.hubServerId,
retentionHours: patch.retentionHours ?? row.retentionHours, retentionHours: patch.retentionHours ?? row.retentionHours,
topN: patch.topN ?? row.topN, topN: patch.topN ?? row.topN,
mapServiceMinSharePct: patch.mapServiceMinSharePct == null
? row.mapServiceMinSharePct
: Math.min(100, Math.max(0, patch.mapServiceMinSharePct)),
updatedAt: nowIso(), updatedAt: nowIso(),
}).where(eq(trafficFlowSettings.id, 1)).run() }).where(eq(trafficFlowSettings.id, 1)).run()
return getTrafficFlowSettingsRow() return getTrafficFlowSettingsRow()
@@ -47,6 +47,8 @@ function NetflowSettingsPanel({
const [endpoint, setEndpoint] = useState("") const [endpoint, setEndpoint] = useState("")
const [retention, setRetention] = useState("24") const [retention, setRetention] = useState("24")
const [topN, setTopN] = useState("200") const [topN, setTopN] = useState("200")
const [shareOn, setShareOn] = useState(true)
const [sharePct, setSharePct] = useState("5")
const [ingestOn, setIngestOn] = useState(false) const [ingestOn, setIngestOn] = useState(false)
const [purgeOpen, setPurgeOpen] = useState(false) const [purgeOpen, setPurgeOpen] = useState(false)
const [purgeBusy, setPurgeBusy] = useState(false) const [purgeBusy, setPurgeBusy] = useState(false)
@@ -62,6 +64,9 @@ function NetflowSettingsPanel({
setEndpoint(s.publicEndpoint) setEndpoint(s.publicEndpoint)
setRetention(String(s.retentionHours)) setRetention(String(s.retentionHours))
setTopN(String(s.topN)) setTopN(String(s.topN))
const pct = Number(s.mapServiceMinSharePct ?? 5)
setShareOn(pct > 0)
setSharePct(String(pct > 0 ? pct : 5))
setIngestOn(s.enabled) setIngestOn(s.enabled)
}, [backendUrl, enabled]) }, [backendUrl, enabled])
@@ -83,6 +88,9 @@ function NetflowSettingsPanel({
publicEndpoint: endpoint, publicEndpoint: endpoint,
retentionHours: Number.parseInt(retention, 10) || 24, retentionHours: Number.parseInt(retention, 10) || 24,
topN: Number.parseInt(topN, 10) || 200, topN: Number.parseInt(topN, 10) || 200,
mapServiceMinSharePct: shareOn
? Math.min(100, Math.max(1, Number.parseFloat(sharePct) || 5))
: 0,
}) })
setSettings(res.settings) setSettings(res.settings)
toast.success("Настройки NetFlow сохранены") toast.success("Настройки NetFlow сохранены")
@@ -193,6 +201,29 @@ function NetflowSettingsPanel({
<FormField label="Top-N разговоров"> <FormField label="Top-N разговоров">
<Input value={topN} onChange={(e) => setTopN(e.target.value)} inputMode="numeric" /> <Input value={topN} onChange={(e) => setTopN(e.target.value)} inputMode="numeric" />
</FormField> </FormField>
<div className="sm:col-span-2 flex flex-col gap-2">
<div className="flex items-center gap-3">
<FormToggle
checked={shareOn}
onChange={(on) => {
setShareOn(on)
if (on && (!sharePct || sharePct === "0")) setSharePct("5")
}}
/>
<span className="text-sm">Порог доли на карте</span>
</div>
<FormField
label="Минимум % окна"
hint="Узел сервиса, если доля байт окна ≥ N%. Выключить — показать все распознанные бренды (макс. 20)"
>
<Input
value={sharePct}
onChange={(e) => setSharePct(e.target.value)}
inputMode="decimal"
disabled={!shareOn}
/>
</FormField>
</div>
</div> </div>
<p className="text-xs text-muted-foreground"> <p className="text-xs text-muted-foreground">
+3
View File
@@ -21,6 +21,7 @@ export const trafficFlowSettingsDtoSchema = z.object({
hubServerId: z.number().int().positive().nullable(), hubServerId: z.number().int().positive().nullable(),
retentionHours: z.number().int().positive(), retentionHours: z.number().int().positive(),
topN: z.number().int().positive(), topN: z.number().int().positive(),
mapServiceMinSharePct: z.number().min(0).max(100),
lastDatagramAt: z.string().nullable(), lastDatagramAt: z.string().nullable(),
lastExporterIp: z.string().nullable(), lastExporterIp: z.string().nullable(),
lastError: z.string().nullable(), lastError: z.string().nullable(),
@@ -40,6 +41,7 @@ export const trafficFlowSettingsPatchSchema = z.object({
hubServerId: z.number().int().positive().nullable().optional(), hubServerId: z.number().int().positive().nullable().optional(),
retentionHours: z.number().int().positive().optional(), retentionHours: z.number().int().positive().optional(),
topN: z.number().int().positive().max(1000).optional(), topN: z.number().int().positive().max(1000).optional(),
mapServiceMinSharePct: z.number().min(0).max(100).optional(),
}) })
export const trafficFlowOverlayRequestSchema = z.object({ export const trafficFlowOverlayRequestSchema = z.object({
@@ -289,6 +291,7 @@ export const flowMapHopsDtoSchema = z.object({
totalBytes: z.number().nonnegative().optional(), totalBytes: z.number().nonnegative().optional(),
services: z.array(flowMapServiceDtoSchema).optional(), services: z.array(flowMapServiceDtoSchema).optional(),
serviceEdges: z.array(flowMapServiceEdgeDtoSchema).optional(), serviceEdges: z.array(flowMapServiceEdgeDtoSchema).optional(),
mapServiceMinSharePct: z.number().min(0).max(100).optional(),
dedupApplied: z.boolean(), dedupApplied: z.boolean(),
excludeMeshApplied: z.boolean(), excludeMeshApplied: z.boolean(),
excludeOverlayApplied: z.boolean(), excludeOverlayApplied: z.boolean(),