Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
db21e1217c |
@@ -0,0 +1,63 @@
|
||||
# Локальные GeoLite2-базы (Country + ASN) для потоков по странам и ASN
|
||||
|
||||
## Контекст
|
||||
|
||||
Сейчас страна и ASN для netflow-потоков резолвятся через внешний RIPEstat API (`backend/src/services/traffic-flow-ripe.ts`): лимит 30 новых префиксов/мин, очередь на 90, кэш в PG `flow_ip_meta`. Новые IP «дозревают» с задержкой, IPv6 не покрывается (кэш индексируется только по IPv4). Локальные mmdb-базы дают мгновенный синхронный lookup всех IP без внешних вызовов.
|
||||
|
||||
Решения (подтверждены):
|
||||
- Источник — **MaxMind GeoLite2 через P3TERX-зеркало**: `https://github.com/P3TERX/GeoLite.mmdb/raw/download/GeoLite2-Country.mmdb` и `.../GeoLite2-ASN.mmdb`. Без регистрации, ключей и tar-распаковки. Точность по стране у GeoLite2 и IPinfo паритетная (<1% ошибок у обоих, arXiv 2026); выбран P3TERX за надёжность зеркала (5.2k звёзд) и преемственность: текущий RIPE-путь и так читает GeoLite (`maxmind-geo-lite`), история в кэше остаётся консистентной.
|
||||
- **RIPEstat остаётся fallback** (до первой загрузки баз / если lookup не дал результата).
|
||||
- City-базу не качаем (lat/lng фронтенд не использует).
|
||||
|
||||
## Изменения
|
||||
|
||||
### 1. Зависимость
|
||||
- `npm install -w mikrotik-manager-backend maxmind` — sync-чтение mmdb, встроенные TS-типы, без транзитивных зависимостей, Node 22 ок.
|
||||
|
||||
### 2. Новый сервис `backend/src/services/traffic-flow-geoip.ts`
|
||||
(по конвенциям окружения traffic-flow-*: контракт → маршрут → сервис, без БД-логики в маршрутах)
|
||||
- Каталог: `backend/storage/geoip/` (конвенция `storage/backups`), файлы `GeoLite2-Country.mmdb`, `GeoLite2-ASN.mmdb`.
|
||||
- `initGeoip()` — открыть ридеры best-effort при старте (из `index.ts` рядом с `startTrafficFlowListener`), независимо от настроек автообновления: файлы есть — работают.
|
||||
- `lookupGeoip(ip): FlowIpMeta | null` — синхронно: `country.iso_code` (fallback `registered_country.iso_code`) с валидацией `isIsoCountry`, ASN = `autonomous_system_number`, holder = `autonomous_system_organization`; приватные IP → negative-запись как в RIPE (`isNonPublicIp`); в PG не пишем (lookup и так быстрый). IPv6 поддержан ридером.
|
||||
- `resolveFlowIp(ip)` — фасад: `lookupGeoip(ip) ?? lookupRipeCached(ip)`; главный экспорт для потребителей.
|
||||
- `geoipStatus()` — loaded, даты сборки баз (метаданные mmdb). Тест-хук `setGeoipReadersForTests`. Смена ридеров после обновления — атомарная замена ссылок.
|
||||
|
||||
### 3. Коллектор `backend/src/services/geoip-update-collector.ts`
|
||||
`collectGeoipUpdateOnce()` по образцу `certificate-renew-collector.ts`:
|
||||
1. Conditional GET с ETag/If-None-Match из настроек → 304 = skip (фолбэк-сравнение: размер/содержимое).
|
||||
2. Скачивание в `*.tmp` через глобальный `fetch` + AbortController с таймаутом (внешний HTTP из service-слоя — по правилу fastify-backend-drizzle).
|
||||
3. Валидация: открыть ридер из tmp-файла, пробой 8.8.8.8 (страна US, ASN 15169).
|
||||
4. `fs.rename` атомарная подмена, старый файл → `*.prev` (откат, если новый ридер не открылся).
|
||||
5. Перезагрузка ридеров, статус в настройках; snapshot для `scheduler_runs` (checked/downloaded/skipped/bytes/error).
|
||||
|
||||
### 4. Планировщик (`backend/src/services/scheduler.ts`)
|
||||
- `JOB_KEYS` += `geoip_update`; case в `runSchedulerJobBody`; блок в `refreshScheduler()` по образцу `certificates_renew`: интервал `Math.max(6ч, updateIntervalSec*1000)`, по умолчанию 7 дней (upstream обновляется еженедельно) + немедленный первый запуск при включённой настройке.
|
||||
|
||||
### 5. Схема и миграция
|
||||
- `backend/src/db/schema.ts`: singleton `geoip_settings` — `enabled` (default true), `updateIntervalSec` (default 604800), `lastCheckAt`, `lastSuccessAt`, `lastError`, `countryBuildAt`, `asnBuildAt`, `etagsJson` (jsonb), `createdAt/updatedAt`.
|
||||
- Миграция: `npm run db:generate` → файл в `backend/drizzle/`.
|
||||
|
||||
### 6. API + контракты
|
||||
- `packages/contracts/src/geoip.ts`: zod-схемы настроек/статуса (все входы — Zod, по правилам проекта).
|
||||
- Новый `backend/src/routes/geoip.ts`, регистрация в `index.ts` с prefix `/api`:
|
||||
- `GET /api/geoip` — настройки + статус (ready, даты сборки, последняя проверка/ошибка);
|
||||
- `PUT /api/geoip` — сохранить настройки, затем `refreshScheduler()`;
|
||||
- `POST /api/geoip/update` — запустить загрузку сейчас (409, если уже идёт; флаг-гард как в коллекторах).
|
||||
|
||||
### 7. Интеграция в пайплайн (geoip-first, RIPE-fallback)
|
||||
- `traffic-flow-engine.ts` (`queueParsedFlows`, ~строка 336): `lookupRipeCached` → `resolveFlowIp`. Логика misses не меняется: при готовом mmdb публичные IP (v4+v6) резолвятся сразу, очередь RIPE пустеет; до скачивания баз — прежнее поведение.
|
||||
- Остальные вызовы `lookupRipeCached` → `resolveFlowIp` (grep: как минимум `traffic-flow-analytics.ts` ~258–271).
|
||||
- `classifyFlowDst`/бренды не трогаем: holder из mmdb (org name) встаёт в существующие `HOLDER_BRANDS`-регулярки как есть.
|
||||
|
||||
### 8. Frontend (по next-shadcn-production / ui-guardian: только переиспользование)
|
||||
- Секция «GeoIP-базы (GeoLite2)» внутри существующей `components/traffic/netflow-settings-panel.tsx`: статус (готово/не скачано, даты сборки Country/ASN, последняя проверка, ошибка), тумблер автообновления, интервал, кнопка «Обновить сейчас» с индикатором. Только уже используемые в панели примитивы (Switch/Button/поля) — никаких новых визуальных паттернов и Card-shell. API-клиент через существующие http-хелперы.
|
||||
|
||||
### 9. Хаускипинг, тесты, проверки
|
||||
- `backend/.gitignore`: `storage/geoip/`.
|
||||
- Тесты `backend/src/services/traffic-flow-geoip.test.ts` + скрипт `test:geoip` (по образцу `test:traffic-flow`): приоритет фасада (geoip hit → RIPE не зовётся; miss → fallback), negative на приватных IP, фильтрация EU/ZZ через `isIsoCountry`, коллектор с мокнутым fetch (304-skip, битый файл → подмены нет, `.prev` сохранён), dims по стране/ASN с засеянным ридером.
|
||||
- Проверки после реализации (обязательно по правилам): типы/сборка бэка (`npm run build -w mikrotik-manager-backend`), типы фронта при правке UI (`npx tsc --noEmit`), `npm run test:geoip` и `test:traffic-flow`; предупреждения не игнорировать.
|
||||
- Коммит: `feat(netflow): <subject по-русски>` — новая пользовательская фича (мгновенные страна/ASN в потоках), по commit-messages-ru.
|
||||
- README: короткий раздел о GeoIP; примечание, что в Docker `storage/geoip` ephemeral без тома — базы перекачаются после пересоздания контейнера (~17 МБ); при желании смонтировать volume.
|
||||
|
||||
## Что это даёт
|
||||
Страна и ASN появляются у потока мгновенно при ingest (включая IPv6), без ограничения скорости RIPE; dims `country`/`asn` в `flow_daily_dims`, аналитика (карта, топы, monthly) становятся полными сразу. Внешняя зависимость от stat.ripe.net остаётся только как fallback до первой загрузки баз.
|
||||
@@ -1216,6 +1216,9 @@ export default function NetworkMapPage() {
|
||||
// ── Interaction ─────────────────────────────────────────────────────────────
|
||||
const [selected, setSelected] = useState<Server | null>(null)
|
||||
const [selectedService, setSelectedService] = useState<FlowMapService | null>(null)
|
||||
const liveSelectedService = selectedService
|
||||
? (mapServices.find((s) => s.id === selectedService.id) ?? selectedService)
|
||||
: null
|
||||
const [highlightedPath, setHighlightedPath] = useState<{ viaId: string; enId: string; serviceId: string } | null>(null)
|
||||
const [selWanIdx, setSelWanIdx] = useState<number | null>(null)
|
||||
const [hoveredId, setHoveredId] = useState<string | null>(null)
|
||||
@@ -2702,16 +2705,16 @@ export default function NetworkMapPage() {
|
||||
})()}
|
||||
</div>
|
||||
</>
|
||||
) : selectedService ? (
|
||||
) : liveSelectedService ? (
|
||||
<>
|
||||
<div className="flex items-start gap-2 px-4 py-3 border-b">
|
||||
<div className="mt-0.5">
|
||||
<ServiceBrandIcon label={selectedService.label} size={22} />
|
||||
<ServiceBrandIcon label={liveSelectedService.label} size={22} />
|
||||
</div>
|
||||
<div className="flex-1 min-w-0">
|
||||
<p className="font-mono font-semibold text-sm truncate">{selectedService.label}</p>
|
||||
<p className="font-mono font-semibold text-sm truncate">{liveSelectedService.label}</p>
|
||||
<p className="text-xs text-muted-foreground mt-0.5">
|
||||
Конечный сервис · {selectedService.category}
|
||||
Конечный сервис · {liveSelectedService.category}
|
||||
</p>
|
||||
</div>
|
||||
<button
|
||||
@@ -2726,15 +2729,15 @@ export default function NetworkMapPage() {
|
||||
<div className="flex flex-col gap-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">{serviceSharePct(selectedService.share)}</span>
|
||||
<span className="text-xs font-mono font-medium text-cyan-400">{serviceSharePct(liveSelectedService.share)}</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">
|
||||
{formatNetflowRate({
|
||||
bytes: selectedService.bytes,
|
||||
bps: selectedService.bps,
|
||||
bpsFwd: selectedService.bps,
|
||||
bytes: liveSelectedService.bytes,
|
||||
bps: liveSelectedService.bps,
|
||||
bpsFwd: liveSelectedService.bps,
|
||||
bpsRev: 0,
|
||||
})}
|
||||
</span>
|
||||
@@ -2742,34 +2745,34 @@ export default function NetworkMapPage() {
|
||||
</div>
|
||||
<div>
|
||||
<p className="text-xs font-semibold text-muted-foreground uppercase tracking-wider mb-2">Выход</p>
|
||||
<div className="flex flex-col gap-1.5">
|
||||
{visibleServiceEdges.filter((e) => e.toId === selectedService.id).map((e) => {
|
||||
<div className="flex flex-col gap-3">
|
||||
{visibleServiceEdges.filter((e) => e.toId === liveSelectedService.id).map((e) => {
|
||||
const src = mapServers.find((s) => s.id === e.fromId)
|
||||
const enPaths = mapServicePaths
|
||||
.filter((p) => p.serviceId === liveSelectedService.id && p.enId === e.fromId)
|
||||
.slice()
|
||||
.sort((a, b) => b.bps - a.bps)
|
||||
return (
|
||||
<div key={`${e.fromId}|${e.toId}`} className="flex items-center justify-between text-xs">
|
||||
<span className="font-mono truncate">{src?.name ?? e.fromId}</span>
|
||||
<span className="font-mono text-emerald-400 tabular-nums">
|
||||
{formatNetflowRate({ bytes: e.bytes, bps: e.bps, bpsFwd: e.bpsFwd, bpsRev: e.bpsRev })}
|
||||
</span>
|
||||
<div key={`${e.fromId}|${e.toId}`} className="flex flex-col gap-1.5">
|
||||
<div className="flex items-center justify-between text-xs">
|
||||
<span className="font-mono truncate">{src?.name ?? e.fromId}</span>
|
||||
<span className="font-mono text-emerald-400 tabular-nums">
|
||||
{formatNetflowRate({ bytes: e.bytes, bps: e.bps, bpsFwd: e.bpsFwd, bpsRev: e.bpsRev })}
|
||||
</span>
|
||||
</div>
|
||||
<ServicePathList
|
||||
paths={enPaths}
|
||||
servers={mapServers}
|
||||
services={mapServices}
|
||||
highlight={highlightedPath}
|
||||
viaMode="via"
|
||||
onToggle={togglePathHighlight}
|
||||
/>
|
||||
</div>
|
||||
)
|
||||
})}
|
||||
</div>
|
||||
</div>
|
||||
<div>
|
||||
<p className="text-xs font-semibold text-muted-foreground uppercase tracking-wider mb-2">Пути</p>
|
||||
<ServicePathList
|
||||
paths={mapServicePaths
|
||||
.filter((p) => p.serviceId === selectedService.id)
|
||||
.slice()
|
||||
.sort((a, b) => b.bps - a.bps)}
|
||||
servers={mapServers}
|
||||
services={mapServices}
|
||||
highlight={highlightedPath}
|
||||
viaMode="via"
|
||||
onToggle={togglePathHighlight}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
</>
|
||||
) : selected ? (
|
||||
@@ -3002,7 +3005,11 @@ export default function NetworkMapPage() {
|
||||
<p className="text-xs font-semibold text-muted-foreground uppercase tracking-wider mb-2">Пути</p>
|
||||
<ServicePathList
|
||||
paths={mapServicePaths
|
||||
.filter((p) => p.viaId === selected.id || p.enId === selected.id)
|
||||
.filter((p) => (
|
||||
selected.type === "exit-node"
|
||||
? p.enId === selected.id
|
||||
: p.viaId === selected.id
|
||||
))
|
||||
.slice()
|
||||
.sort((a, b) => b.bps - a.bps)}
|
||||
servers={mapServers}
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js"
|
||||
import {
|
||||
dedupFlowRowsAcrossExporters,
|
||||
dedupFlowRowsMaxBytes,
|
||||
flowConversationKey,
|
||||
flowTupleKey,
|
||||
} from "./traffic-flow-dedup.js"
|
||||
|
||||
const a = {
|
||||
serverId: 7,
|
||||
@@ -22,4 +27,13 @@ assert.equal(flowTupleKey(a), flowTupleKey(b))
|
||||
const sameIface = dedupFlowRowsMaxBytes([a, { ...a, bytes: 3_000, packets: 2 }])
|
||||
assert.equal(sameIface[0]?.bytes, 15_000)
|
||||
|
||||
const jh = { ...a, serverId: 7, bytes: 9_000 }
|
||||
const en = { ...a, serverId: 9, bytes: 11_000, inIface: "1" }
|
||||
assert.equal(flowConversationKey(jh), flowConversationKey(en))
|
||||
assert.notEqual(flowTupleKey(jh), flowTupleKey(en))
|
||||
const across = dedupFlowRowsAcrossExporters([en, jh], (x, y) => (x.serverId === 7 ? x : y))
|
||||
assert.equal(across.length, 1)
|
||||
assert.equal(across[0]?.serverId, 7)
|
||||
assert.equal(across[0]?.bytes, 9_000)
|
||||
|
||||
console.log("traffic-flow-dedup.test.ts: ok")
|
||||
|
||||
@@ -14,6 +14,32 @@ export function flowTupleKey(r: Pick<FlowTupleRow, "serverId" | "src" | "dst" |
|
||||
return `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}`
|
||||
}
|
||||
|
||||
/** Один разговор на всех экспортёрах (JH+EN), без serverId. */
|
||||
export function flowConversationKey(r: Pick<FlowTupleRow, "src" | "dst" | "proto" | "srcPort" | "dstPort">): string {
|
||||
return `${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}`
|
||||
}
|
||||
|
||||
/**
|
||||
* Схлопнуть копии одного 5-tuple с разных серверов.
|
||||
* `prefer` выбирает ряд (клиент на JH важнее голого EN).
|
||||
*/
|
||||
export function dedupFlowRowsAcrossExporters<T extends FlowTupleRow>(
|
||||
rows: T[],
|
||||
prefer: (a: T, b: T) => T,
|
||||
): T[] {
|
||||
const byConv = new Map<string, T>()
|
||||
for (const row of rows) {
|
||||
const key = flowConversationKey(row)
|
||||
const prev = byConv.get(key)
|
||||
if (!prev) {
|
||||
byConv.set(key, row)
|
||||
continue
|
||||
}
|
||||
byConv.set(key, prefer(prev, row))
|
||||
}
|
||||
return [...byConv.values()]
|
||||
}
|
||||
|
||||
function ifaceKey(r: FlowTupleRow): string {
|
||||
return `${flowTupleKey(r)}|${r.inIface}`
|
||||
}
|
||||
|
||||
@@ -584,4 +584,98 @@ try {
|
||||
resetFlowCatalogForTests()
|
||||
}
|
||||
|
||||
resetFlowRingsForTests()
|
||||
resetIfaceCacheForTests()
|
||||
resetRipeCacheForTests()
|
||||
disableRipeEnqueueForTests()
|
||||
seedFlowTopologyForTests(topo)
|
||||
rememberServerIfaces(7, [
|
||||
{ ".id": "*2", name: "gre-client" },
|
||||
{ ".id": "*3", name: "gre-jh-en" },
|
||||
])
|
||||
rememberServerIfaces(9, [
|
||||
{ ".id": "*1", name: "ether1" },
|
||||
])
|
||||
googleRipe()
|
||||
ingestParsedFlowsForServerForTests(7, [payloadFlow("8.8.8.8", 12_000)])
|
||||
ingestParsedFlowsForServerForTests(9, [{
|
||||
src: "10.100.1.17",
|
||||
dst: "8.8.8.8",
|
||||
proto: 6,
|
||||
srcPort: 51234,
|
||||
dstPort: 443,
|
||||
bytes: 12_000,
|
||||
packets: 10,
|
||||
inIface: "1",
|
||||
outIface: "1",
|
||||
nextHop: "",
|
||||
}])
|
||||
try {
|
||||
resetFlowMapHopsCacheForTests()
|
||||
const dual = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
|
||||
const gre = dual.hops.find((h) => h.kind === "gre" && h.fromId === "7" && h.toId === "9")
|
||||
assert.ok(gre, "GRE JH→EN сохранён")
|
||||
assert.equal(gre.bytes, 12_000)
|
||||
const googlePaths = (dual.servicePaths ?? []).filter((p) => p.serviceId === "svc:google")
|
||||
assert.equal(googlePaths.length, 1, "один путь без копии EN")
|
||||
assert.equal(googlePaths[0]?.clientId, "u1")
|
||||
assert.equal(googlePaths[0]?.viaId, "7")
|
||||
const googleEdge = dual.serviceEdges?.find((e) => e.toId === "svc:google" && e.fromId === "9")
|
||||
assert.ok(googleEdge)
|
||||
assert.equal(googleEdge.bytes, 12_000)
|
||||
assert.equal(googleEdge.bps, googlePaths[0]?.bps)
|
||||
} finally {
|
||||
seedFlowTopologyForTests(null)
|
||||
resetFlowRingsForTests()
|
||||
resetIfaceCacheForTests()
|
||||
resetRipeCacheForTests()
|
||||
resetFlowCatalogForTests()
|
||||
}
|
||||
|
||||
resetFlowRingsForTests()
|
||||
resetIfaceCacheForTests()
|
||||
resetRipeCacheForTests()
|
||||
disableRipeEnqueueForTests()
|
||||
seedFlowTopologyForTests(topo)
|
||||
rememberServerIfaces(7, [
|
||||
{ ".id": "*2", name: "gre-client" },
|
||||
{ ".id": "*3", name: "gre-jh-en" },
|
||||
])
|
||||
googleRipe()
|
||||
ingestParsedFlowsForServerForTests(7, [
|
||||
payloadFlow("8.8.8.8", 8_000),
|
||||
{
|
||||
src: "8.8.8.8",
|
||||
dst: "10.100.1.17",
|
||||
proto: 6,
|
||||
srcPort: 443,
|
||||
dstPort: 51234,
|
||||
bytes: 4_000,
|
||||
packets: 8,
|
||||
inIface: "3",
|
||||
outIface: "2",
|
||||
nextHop: "",
|
||||
},
|
||||
])
|
||||
try {
|
||||
resetFlowMapHopsCacheForTests()
|
||||
const bothDir = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
|
||||
const gre = bothDir.hops.find((h) => h.kind === "gre" && h.fromId === "7" && h.toId === "9")
|
||||
assert.ok(gre, "GRE-hop при fwd/rev")
|
||||
const googlePaths = (bothDir.servicePaths ?? []).filter((p) => p.serviceId === "svc:google")
|
||||
assert.equal(googlePaths.length, 1, "fwd+rev — один клиент")
|
||||
assert.equal(googlePaths[0]?.clientId, "u1")
|
||||
assert.ok(!(bothDir.servicePaths ?? []).some((p) => p.serviceId === "svc:google" && p.clientId === "—"))
|
||||
const googleEdge = bothDir.serviceEdges?.find((e) => e.toId === "svc:google" && e.fromId === "9")
|
||||
assert.ok(googleEdge)
|
||||
assert.equal(googleEdge.bytes, 12_000)
|
||||
assert.equal(googleEdge.bps, googlePaths[0]?.bps)
|
||||
} finally {
|
||||
seedFlowTopologyForTests(null)
|
||||
resetFlowRingsForTests()
|
||||
resetIfaceCacheForTests()
|
||||
resetRipeCacheForTests()
|
||||
resetFlowCatalogForTests()
|
||||
}
|
||||
|
||||
console.log("traffic-flow-map-hops.test.ts: ok")
|
||||
|
||||
@@ -8,7 +8,7 @@ import {
|
||||
mapServiceNodeId,
|
||||
resolveFlowBrand,
|
||||
} from "./traffic-flow-brands.js"
|
||||
import { dedupFlowRowsMaxBytes } from "./traffic-flow-dedup.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"
|
||||
@@ -16,7 +16,7 @@ import { pickInternetPeer } from "./traffic-flow-ip.js"
|
||||
import { type FlowIpMeta } from "./traffic-flow-ripe.js"
|
||||
import { resolveFlowIp } from "./traffic-flow-geoip.js"
|
||||
import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js"
|
||||
import { loadFlowTopology, resolveClient, resolveEn, getServerCatalog } from "./traffic-flow-topology.js"
|
||||
import { loadFlowTopology, resolveClient, resolveEn, getServerCatalog, type FlowTopology } from "./traffic-flow-topology.js"
|
||||
import { flowDataEpoch } from "./traffic-flow-engine.js"
|
||||
|
||||
export const DEFAULT_MAP_SERVICE_MIN_SHARE_PCT = 5
|
||||
@@ -141,6 +141,16 @@ function ifaceUsable(name: string): boolean {
|
||||
return Boolean(name) && name !== "—"
|
||||
}
|
||||
|
||||
function resolveMapClient(
|
||||
topo: FlowTopology,
|
||||
serverId: number,
|
||||
inName: string,
|
||||
outName: string,
|
||||
) {
|
||||
return resolveClient(topo, serverId, inName)
|
||||
?? (ifaceUsable(outName) ? resolveClient(topo, serverId, outName) : null)
|
||||
}
|
||||
|
||||
function bump(acc: Map<string, HopAcc>, key: string, seed: Omit<HopAcc, "bytes" | "bytesFwd" | "bytesRev">, bytes: number, dir: "fwd" | "rev" | "both"): void {
|
||||
const prev = acc.get(key)
|
||||
const addFwd = dir === "fwd" || dir === "both" ? bytes : 0
|
||||
@@ -238,6 +248,21 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
|
||||
const enIds = new Set(topo.enNodes.map((n) => n.id))
|
||||
let totalBytes = 0
|
||||
|
||||
function rowClient(r: (typeof working)[number]) {
|
||||
const inName = resolveIfaceName(r.serverId, r.inIface).name
|
||||
const outName = resolveIfaceName(r.serverId, r.outIface).name
|
||||
return resolveMapClient(topo, r.serverId, inName, outName)
|
||||
}
|
||||
|
||||
const payloadRows = wantDedup
|
||||
? dedupFlowRowsAcrossExporters(working, (a, b) => {
|
||||
const aCli = Boolean(rowClient(a))
|
||||
const bCli = Boolean(rowClient(b))
|
||||
if (aCli !== bCli) return aCli ? a : b
|
||||
return a.bytes >= b.bytes ? a : b
|
||||
})
|
||||
: working
|
||||
|
||||
for (const r of working) {
|
||||
const inRes = resolveIfaceName(r.serverId, r.inIface)
|
||||
const outRes = resolveIfaceName(r.serverId, r.outIface)
|
||||
@@ -323,10 +348,14 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
|
||||
}, r.bytes, "fwd")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (const r of payloadRows) {
|
||||
const inName = resolveIfaceName(r.serverId, r.inIface).name
|
||||
const outName = resolveIfaceName(r.serverId, r.outIface).name
|
||||
totalBytes += r.bytes
|
||||
const peer = pickInternetPeer(r.src, r.dst, r.srcPort, r.dstPort)
|
||||
const client = resolveClient(topo, r.serverId, inName)
|
||||
const client = resolveMapClient(topo, r.serverId, inName, outName)
|
||||
const prevDst = dstAcc.get(peer)
|
||||
if (prevDst) {
|
||||
prevDst.bytes += r.bytes
|
||||
@@ -428,10 +457,15 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
|
||||
const enName = nodeName(fromId)
|
||||
const viaName = nodeName(exporterId)
|
||||
for (const [clientId, c] of from.clients) {
|
||||
const pathKey = `${clientId}|${exporterId}|${fromId}|${toId}`
|
||||
const pathKey = `${clientId}|${fromId}|${toId}`
|
||||
const prevPath = svcPaths.get(pathKey)
|
||||
if (prevPath) {
|
||||
prevPath.bytes += c.bytes
|
||||
if (exporterId !== fromId && prevPath.viaId === fromId) {
|
||||
prevPath.viaId = exporterId
|
||||
prevPath.viaName = viaName
|
||||
}
|
||||
if (prevPath.clientName === "—" && c.name !== "—") prevPath.clientName = c.name
|
||||
} else {
|
||||
svcPaths.set(pathKey, {
|
||||
clientId,
|
||||
|
||||
File diff suppressed because one or more lines are too long
Reference in New Issue
Block a user