DenozordecandCursor
0c0dfa1df7
Docker images / prepare-release (push) Successful in 8s
Docker images / backend-test (push) Successful in 2m18s
Docker images / frontend-image (push) Successful in 3m25s
Docker images / updater-image (push) Successful in 48s
Docker images / backend-image (push) Successful in 3m9s
Docker images / notify-webhook (push) Skipped
Docker images / publish-release (push) Successful in 14s
Co-authored-by: Cursor <[email protected]>
1015 lines
32 KiB
TypeScript
1015 lines
32 KiB
TypeScript
import { dbAll, dbQuery, pool } from "../db/index.js"
|
||
import { ensurePartitionFor, specForParent } from "../db/partitions.js"
|
||
import { normalizeParsedFlow, parseFlowPacket, protoName, type ParsedFlow, type ParsedFlowInput } from "./traffic-flow-parse.js"
|
||
import { classifyFlowPlaneLite } from "./traffic-flow-planes.js"
|
||
import { pickServerIdForExporter, type OverlayPeerRef } from "./traffic-flow-map-exporter.js"
|
||
import { applicationName } from "./traffic-flow-apps.js"
|
||
import { classifyFlowDst } from "./traffic-flow-classify.js"
|
||
import { enqueueRipeMisses, pruneRipeSqlite } from "./traffic-flow-ripe.js"
|
||
import { resolveFlowIp } from "./traffic-flow-geoip.js"
|
||
import { invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js"
|
||
import { isIsoCountry } from "./traffic-flow-brands.js"
|
||
import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js"
|
||
import { canonicalFactIface } from "./traffic-flow-ifindex.js"
|
||
import { shouldWriteFlowFact } from "./traffic-flow-facts-filter.js"
|
||
import { pickInternetPeer } from "./traffic-flow-ip.js"
|
||
import {
|
||
getServerCatalog,
|
||
loadFlowTopology,
|
||
peekFlowTopology,
|
||
peekServerCatalog,
|
||
} from "./traffic-flow-topology.js"
|
||
import {
|
||
bumpFlowFact,
|
||
factsPendingSize,
|
||
flushFlowFacts,
|
||
hourBucketIso,
|
||
resetFactsForTests,
|
||
} from "./traffic-flow-facts.js"
|
||
|
||
export const TICK_MS = 2_000
|
||
export const PERSIST_MS = 10_000
|
||
export const STATS_PERSIST_MS = 15_000
|
||
export const RING_LEN = 60
|
||
export const MAX_PENDING = 50_000
|
||
export const DAILY_ASN_TOP = 500
|
||
export const DAILY_RETENTION_DAYS = 396
|
||
export const MINUTE_RETENTION_HOURS = 48
|
||
|
||
let pendingCap = MAX_PENDING
|
||
const ensuredParts = new Set<string>()
|
||
|
||
async function ensureParentPartition(parent: string, ts: string): Promise<void> {
|
||
const spec = specForParent(parent)
|
||
if (!spec) return
|
||
const iso = ts.length === 10 ? `${ts}T00:00:00Z` : ts
|
||
const key = `${parent}:${iso.slice(0, 10)}`
|
||
if (ensuredParts.has(key)) return
|
||
await ensurePartitionFor(pool, parent, spec.kind, new Date(iso))
|
||
ensuredParts.add(key)
|
||
}
|
||
|
||
export interface PendingFlowRow {
|
||
serverId: number
|
||
bucketAt: string
|
||
src: string
|
||
dst: string
|
||
proto: number
|
||
srcPort: number
|
||
dstPort: number
|
||
bytes: number
|
||
packets: number
|
||
inIface: string
|
||
outIface: string
|
||
nextHop: string
|
||
flowStartMs: number
|
||
flowEndMs: number
|
||
}
|
||
|
||
function inetOrNull(value: string | null | undefined): string | null {
|
||
const s = String(value ?? "").trim()
|
||
return s.length > 0 ? s : null
|
||
}
|
||
|
||
export function isValidFlowInet(value: string): boolean {
|
||
const s = value.trim()
|
||
if (!s) return false
|
||
const v4 = /^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/.exec(s)
|
||
if (v4) {
|
||
return v4.slice(1).every((octet) => {
|
||
const n = Number(octet)
|
||
return Number.isInteger(n) && n >= 0 && n <= 255
|
||
})
|
||
}
|
||
if (!s.includes(":")) return false
|
||
if (!/^[0-9a-fA-F:]+$/.test(s)) return false
|
||
const parts = s.split(":")
|
||
if (parts.length < 3 || parts.length > 8) return false
|
||
return parts.every((p) => p.length <= 4)
|
||
}
|
||
|
||
function clampProto(n: number): number {
|
||
if (!Number.isFinite(n)) return 0
|
||
return Math.max(0, Math.min(255, Math.trunc(n)))
|
||
}
|
||
|
||
function sanitizeFlowRow(r: PendingFlowRow): PendingFlowRow | null {
|
||
const src = (r.src || "").trim() || "0.0.0.0"
|
||
const dst = (r.dst || "").trim() || "0.0.0.0"
|
||
if (!isValidFlowInet(src) || !isValidFlowInet(dst)) return null
|
||
const next = inetOrNull(r.nextHop)
|
||
return {
|
||
...r,
|
||
src,
|
||
dst,
|
||
nextHop: next && isValidFlowInet(next) ? next : "",
|
||
proto: clampProto(r.proto),
|
||
}
|
||
}
|
||
|
||
function flowUpsertParams(r: PendingFlowRow) {
|
||
return {
|
||
serverId: r.serverId,
|
||
bucketAt: r.bucketAt,
|
||
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: inetOrNull(r.nextHop),
|
||
flowStartMs: r.flowStartMs,
|
||
flowEndMs: r.flowEndMs,
|
||
}
|
||
}
|
||
|
||
export interface EngineStats {
|
||
packetsReceived: number
|
||
lastExporterIp: string | null
|
||
lastError: string
|
||
lastDatagramAt: string | null
|
||
dropped: number
|
||
rowsStored: number
|
||
pendingSize: number
|
||
}
|
||
|
||
interface PendingEntry {
|
||
serverId: number
|
||
bucketAt: string
|
||
flow: ParsedFlow
|
||
bytes: number
|
||
packets: number
|
||
}
|
||
|
||
interface MinuteRollup {
|
||
bytes: number
|
||
packets: number
|
||
srcs: Set<string>
|
||
dsts: Set<string>
|
||
conversations: number
|
||
}
|
||
|
||
interface DimAcc {
|
||
bytes: number
|
||
packets: number
|
||
}
|
||
|
||
export interface ExporterResolveCtx {
|
||
overlayPrefix: string
|
||
byTunnelIp: Map<string, number>
|
||
peers: OverlayPeerRef[]
|
||
hostIps: Map<string, number>
|
||
}
|
||
|
||
let topN = 200
|
||
let retentionHours = 24
|
||
|
||
const pending = new Map<string, PendingEntry>()
|
||
const recent = new Map<string, PendingFlowRow>()
|
||
const tickAccum = new Map<string, { inBytes: number; outBytes: number }>()
|
||
const rings = new Map<string, { inBps: number[]; outBps: number[] }>()
|
||
const minuteRollup = new Map<string, MinuteRollup>()
|
||
const minuteDims = new Map<string, DimAcc>()
|
||
|
||
let packetsReceived = 0
|
||
let lastExporterIp: string | null = null
|
||
let lastError = ""
|
||
let lastDatagramAt: string | null = null
|
||
let dropped = 0
|
||
let rowsStored = 0
|
||
let lastFlushUsedTransaction = false
|
||
let lastPruneAt = 0
|
||
let lastPassiveCheckpointAt = 0
|
||
let lastPersistAt = 0
|
||
let lastFlushedMinute = ""
|
||
let lastStatsPersistAt = 0
|
||
let dataEpoch = 0
|
||
let lastPersistedStats: {
|
||
packetsReceived: number
|
||
lastDatagramAt: string | null
|
||
lastExporterIp: string | null
|
||
lastError: string
|
||
} | null = null
|
||
let exporterCtx: ExporterResolveCtx | null = null
|
||
|
||
const PRUNE_MS = 5 * 60_000
|
||
const LIVE_WINDOW_MS = 15 * 60_000
|
||
|
||
function bumpDataEpoch(): void {
|
||
dataEpoch += 1
|
||
}
|
||
|
||
export function flowDataEpoch(): number {
|
||
return dataEpoch
|
||
}
|
||
|
||
function nowIso(): string {
|
||
return new Date().toISOString()
|
||
}
|
||
|
||
export function minuteBucketIso(at = Date.now()): string {
|
||
const d = new Date(at)
|
||
d.setSeconds(0, 0)
|
||
return d.toISOString()
|
||
}
|
||
|
||
function dayKey(bucketAt: string): string {
|
||
return bucketAt.slice(0, 10)
|
||
}
|
||
|
||
export const RING_PAYLOAD = "__all__"
|
||
export const RING_OVERLAY = "__overlay__"
|
||
export const RING_MESH = "__mesh__"
|
||
|
||
function ringKey(serverId: number, iface: string): string {
|
||
return `${serverId}\0${iface || RING_PAYLOAD}`
|
||
}
|
||
|
||
function pendingKey(serverId: number, bucketAt: string, flow: ParsedFlow): string {
|
||
return `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}\0${flow.inIface}`
|
||
}
|
||
|
||
function rowKey(row: PendingFlowRow): string {
|
||
return `${row.serverId}|${row.bucketAt}|${row.src}|${row.dst}|${row.proto}|${row.srcPort}|${row.dstPort}|${row.inIface}`
|
||
}
|
||
|
||
function rollupKey(serverId: number, bucketAt: string): string {
|
||
return `${serverId}\0${bucketAt}`
|
||
}
|
||
|
||
function dimKey(serverId: number, bucketAt: string, dim: string, key: string): string {
|
||
return `${serverId}\0${bucketAt}\0${dim}\0${key}`
|
||
}
|
||
|
||
function bumpTick(key: string, inBytes: number, outBytes: number): void {
|
||
const prev = tickAccum.get(key) ?? { inBytes: 0, outBytes: 0 }
|
||
prev.inBytes += inBytes
|
||
prev.outBytes += outBytes
|
||
tickAccum.set(key, prev)
|
||
}
|
||
|
||
function addToTick(serverId: number, flow: ParsedFlow, bytes: number): void {
|
||
const plane = classifyFlowPlaneLite(flow)
|
||
if (plane === "mgmt") return
|
||
const bucket = plane === "overlay" ? RING_OVERLAY : plane === "client_mesh" ? RING_MESH : RING_PAYLOAD
|
||
bumpTick(ringKey(serverId, bucket), bytes, 0)
|
||
if (flow.inIface) bumpTick(ringKey(serverId, flow.inIface), bytes, 0)
|
||
if (flow.outIface && flow.outIface !== flow.inIface) bumpTick(ringKey(serverId, flow.outIface), 0, bytes)
|
||
}
|
||
|
||
function emptyRing(): { inBps: number[]; outBps: number[] } {
|
||
return { inBps: Array(RING_LEN).fill(0), outBps: Array(RING_LEN).fill(0) }
|
||
}
|
||
|
||
function bumpDim(serverId: number, bucketAt: string, dim: string, key: string, bytes: number, packets: number): void {
|
||
if (!key) return
|
||
const k = dimKey(serverId, bucketAt, dim, key)
|
||
const prev = minuteDims.get(k)
|
||
if (prev) {
|
||
prev.bytes += bytes
|
||
prev.packets += packets
|
||
return
|
||
}
|
||
minuteDims.set(k, { bytes, packets })
|
||
}
|
||
|
||
function bumpRollup(serverId: number, bucketAt: string, flow: ParsedFlow, bytes: number, packets: number): void {
|
||
const k = rollupKey(serverId, bucketAt)
|
||
let acc = minuteRollup.get(k)
|
||
if (!acc) {
|
||
acc = { bytes: 0, packets: 0, srcs: new Set(), dsts: new Set(), conversations: 0 }
|
||
minuteRollup.set(k, acc)
|
||
}
|
||
acc.bytes += bytes
|
||
acc.packets += packets
|
||
if (flow.src) acc.srcs.add(flow.src)
|
||
if (flow.dst) acc.dsts.add(flow.dst)
|
||
acc.conversations += 1
|
||
}
|
||
|
||
export function attachEngineSqlite(_handle?: unknown): void {
|
||
/* engine uses pg pool via dbQuery/dbAll */
|
||
}
|
||
|
||
export function setPendingCapForTests(n: number | null): void {
|
||
pendingCap = n == null ? MAX_PENDING : Math.max(1, n)
|
||
}
|
||
|
||
export function configureEngine(opts: { topN?: number; retentionHours?: number }): void {
|
||
if (opts.topN != null) topN = Math.max(20, opts.topN)
|
||
if (opts.retentionHours != null) retentionHours = Math.max(1, opts.retentionHours)
|
||
}
|
||
|
||
export function setExporterResolveCtx(ctx: ExporterResolveCtx | null): void {
|
||
exporterCtx = ctx
|
||
}
|
||
|
||
export function resolveServerId(exporterIp: string): number | null {
|
||
if (!exporterCtx) return null
|
||
return pickServerIdForExporter({
|
||
exporterIp,
|
||
overlayPrefix: exporterCtx.overlayPrefix,
|
||
byTunnelIp: exporterCtx.byTunnelIp,
|
||
peers: exporterCtx.peers,
|
||
hostIps: exporterCtx.hostIps,
|
||
})
|
||
}
|
||
|
||
export function bumpPacketMeta(exporterIp: string): void {
|
||
packetsReceived += 1
|
||
lastExporterIp = exporterIp
|
||
lastDatagramAt = nowIso()
|
||
}
|
||
|
||
export function setEngineError(message: string): void {
|
||
lastError = message
|
||
}
|
||
|
||
export function getEngineStats(): EngineStats {
|
||
return {
|
||
packetsReceived,
|
||
lastExporterIp,
|
||
lastError,
|
||
lastDatagramAt,
|
||
dropped,
|
||
rowsStored,
|
||
pendingSize: pending.size,
|
||
}
|
||
}
|
||
|
||
export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): void {
|
||
if (flows.length) bumpDataEpoch()
|
||
const bucketAt = minuteBucketIso()
|
||
const hourAt = hourBucketIso()
|
||
const ripeMisses: string[] = []
|
||
const topo = peekFlowTopology()
|
||
const catalog = peekServerCatalog()
|
||
if (!topo) void loadFlowTopology().catch(() => {})
|
||
if (!catalog) void getServerCatalog().catch(() => {})
|
||
const serverType = catalog?.byId.get(serverId)?.type
|
||
for (const raw of flows) {
|
||
const flow = normalizeParsedFlow(raw)
|
||
addToTick(serverId, flow, flow.bytes)
|
||
bumpRollup(serverId, bucketAt, flow, flow.bytes, flow.packets)
|
||
const peer = pickInternetPeer(flow.src, flow.dst, flow.srcPort, flow.dstPort)
|
||
const ripe = resolveFlowIp(peer)
|
||
if (peer && !ripe) ripeMisses.push(peer)
|
||
const classified = classifyFlowDst(peer, flow.proto, flow.dstPort, flow.srcPort, ripe)
|
||
const app = applicationName(flow.proto, flow.dstPort, flow.srcPort)
|
||
const country = ripe?.ok && isIsoCountry(ripe.country)
|
||
? ripe.country
|
||
: (ripe?.ok ? "" : "unknown")
|
||
const asnKey = ripe?.ok && ripe.asn ? String(ripe.asn) : "unknown"
|
||
bumpDim(serverId, bucketAt, "proto", protoName(flow.proto), flow.bytes, flow.packets)
|
||
bumpDim(serverId, bucketAt, "app", app, flow.bytes, flow.packets)
|
||
bumpDim(serverId, bucketAt, "iface", flow.inIface || "__unknown__", flow.bytes, flow.packets)
|
||
bumpDim(serverId, bucketAt, "category", classified.category, flow.bytes, flow.packets)
|
||
bumpDim(serverId, bucketAt, "service", classified.service, flow.bytes, flow.packets)
|
||
if (country) bumpDim(serverId, bucketAt, "country", country, flow.bytes, flow.packets)
|
||
bumpDim(serverId, bucketAt, "asn", asnKey, flow.bytes, flow.packets)
|
||
if (shouldWriteFlowFact({
|
||
serverId,
|
||
serverType,
|
||
inIface: flow.inIface,
|
||
outIface: flow.outIface,
|
||
proto: flow.proto,
|
||
srcPort: flow.srcPort,
|
||
dstPort: flow.dstPort,
|
||
src: flow.src,
|
||
dst: flow.dst,
|
||
topo,
|
||
})) {
|
||
bumpFlowFact({
|
||
serverId,
|
||
bucketAt: hourAt,
|
||
iface: canonicalFactIface(serverId, flow.inIface),
|
||
country: country || "XX",
|
||
service: classified.service,
|
||
asn: ripe?.ok && ripe.asn ? ripe.asn : 0,
|
||
bytes: flow.bytes,
|
||
packets: flow.packets,
|
||
})
|
||
}
|
||
|
||
const key = pendingKey(serverId, bucketAt, flow)
|
||
const prev = pending.get(key)
|
||
if (prev) {
|
||
prev.bytes += flow.bytes
|
||
prev.packets += flow.packets
|
||
if (flow.outIface && !prev.flow.outIface) prev.flow.outIface = flow.outIface
|
||
if (flow.nextHop && !prev.flow.nextHop) prev.flow.nextHop = flow.nextHop
|
||
if (flow.flowStartMs && (!prev.flow.flowStartMs || flow.flowStartMs < prev.flow.flowStartMs)) {
|
||
prev.flow.flowStartMs = flow.flowStartMs
|
||
}
|
||
if (flow.flowEndMs > (prev.flow.flowEndMs ?? 0)) prev.flow.flowEndMs = flow.flowEndMs
|
||
continue
|
||
}
|
||
if (pending.size >= pendingCap) {
|
||
dropped += 1
|
||
continue
|
||
}
|
||
pending.set(key, {
|
||
serverId,
|
||
bucketAt,
|
||
flow: { ...flow },
|
||
bytes: flow.bytes,
|
||
packets: flow.packets,
|
||
})
|
||
}
|
||
if (ripeMisses.length) enqueueRipeMisses(ripeMisses)
|
||
}
|
||
|
||
export function ingestDatagram(msg: Buffer, exporterIp: string): boolean {
|
||
bumpPacketMeta(exporterIp)
|
||
const flows = parseFlowPacket(msg, exporterIp)
|
||
if (!flows.length) return true
|
||
const serverId = resolveServerId(exporterIp)
|
||
if (serverId == null) {
|
||
setEngineError(
|
||
`IPFIX от ${exporterIp}: нет jump-host с адресом wg-flow. Docker SNAT (172.x) при нескольких JH не различим.`,
|
||
)
|
||
return false
|
||
}
|
||
setEngineError("")
|
||
maybeRefreshIfaces(serverId)
|
||
queueParsedFlows(serverId, flows)
|
||
return true
|
||
}
|
||
|
||
function toPendingRow(row: PendingEntry): PendingFlowRow {
|
||
const flow = normalizeParsedFlow(row.flow)
|
||
return {
|
||
serverId: row.serverId,
|
||
bucketAt: row.bucketAt,
|
||
src: flow.src || "0.0.0.0",
|
||
dst: flow.dst || "0.0.0.0",
|
||
proto: flow.proto,
|
||
srcPort: flow.srcPort,
|
||
dstPort: flow.dstPort,
|
||
bytes: row.bytes,
|
||
packets: row.packets,
|
||
inIface: flow.inIface,
|
||
outIface: flow.outIface,
|
||
nextHop: flow.nextHop,
|
||
flowStartMs: flow.flowStartMs,
|
||
flowEndMs: flow.flowEndMs,
|
||
}
|
||
}
|
||
|
||
function mergeInto(map: Map<string, PendingFlowRow>, row: PendingFlowRow): void {
|
||
const key = rowKey(row)
|
||
const prev = map.get(key)
|
||
if (prev) {
|
||
prev.bytes += row.bytes
|
||
prev.packets += row.packets
|
||
if (row.outIface && !prev.outIface) prev.outIface = row.outIface
|
||
if (row.nextHop && !prev.nextHop) prev.nextHop = row.nextHop
|
||
if (row.flowStartMs && (!prev.flowStartMs || row.flowStartMs < prev.flowStartMs)) prev.flowStartMs = row.flowStartMs
|
||
if (row.flowEndMs > (prev.flowEndMs ?? 0)) prev.flowEndMs = row.flowEndMs
|
||
return
|
||
}
|
||
map.set(key, { ...row })
|
||
}
|
||
|
||
function pruneRecent(sinceMs = Date.now() - LIVE_WINDOW_MS): void {
|
||
const cutoff = new Date(sinceMs).toISOString()
|
||
for (const [key, row] of recent) {
|
||
if (row.bucketAt < cutoff) recent.delete(key)
|
||
}
|
||
while (recent.size > MAX_PENDING) {
|
||
const first = recent.keys().next().value
|
||
if (first == null) break
|
||
recent.delete(first)
|
||
}
|
||
}
|
||
|
||
export function peekPendingFlows(): PendingFlowRow[] {
|
||
return [...pending.values()].map(toPendingRow)
|
||
}
|
||
|
||
export function listLiveFlowRows(sinceIso: string): PendingFlowRow[] {
|
||
const merged = new Map<string, PendingFlowRow>()
|
||
for (const row of recent.values()) {
|
||
if (row.bucketAt < sinceIso) continue
|
||
mergeInto(merged, row)
|
||
}
|
||
for (const row of peekPendingFlows()) {
|
||
if (row.bucketAt < sinceIso) continue
|
||
mergeInto(merged, row)
|
||
}
|
||
return [...merged.values()]
|
||
}
|
||
|
||
export function rollFlowRings(): void {
|
||
const keys = new Set([...tickAccum.keys(), ...rings.keys()])
|
||
const sec = TICK_MS / 1000
|
||
for (const key of keys) {
|
||
const acc = tickAccum.get(key) ?? { inBytes: 0, outBytes: 0 }
|
||
tickAccum.delete(key)
|
||
const inBps = (acc.inBytes * 8) / sec
|
||
const outBps = (acc.outBytes * 8) / sec
|
||
let ring = rings.get(key)
|
||
if (!ring) {
|
||
ring = emptyRing()
|
||
rings.set(key, ring)
|
||
}
|
||
ring.inBps.push(inBps)
|
||
ring.inBps.shift()
|
||
ring.outBps.push(outBps)
|
||
ring.outBps.shift()
|
||
const silent = ring.inBps.every((v) => v === 0) && ring.outBps.every((v) => v === 0)
|
||
if (silent && !tickAccum.has(key)) rings.delete(key)
|
||
}
|
||
}
|
||
|
||
export function getRingMbps(serverId: number, iface = RING_PAYLOAD): {
|
||
rx: number[]
|
||
tx: number[]
|
||
rxNow: number
|
||
txNow: number
|
||
} {
|
||
const ring = rings.get(ringKey(serverId, iface))
|
||
const scale = 1_000_000
|
||
if (!ring) {
|
||
return { rx: Array(RING_LEN).fill(0), tx: Array(RING_LEN).fill(0), rxNow: 0, txNow: 0 }
|
||
}
|
||
return {
|
||
rx: ring.inBps.map((b) => b / scale),
|
||
tx: ring.outBps.map((b) => b / scale),
|
||
rxNow: (ring.inBps[RING_LEN - 1] ?? 0) / scale,
|
||
txNow: (ring.outBps[RING_LEN - 1] ?? 0) / scale,
|
||
}
|
||
}
|
||
|
||
export function snapshotRings(): Array<{ key: string; inBps: number[]; outBps: number[] }> {
|
||
return [...rings.entries()].map(([key, ring]) => ({
|
||
key,
|
||
inBps: [...ring.inBps],
|
||
outBps: [...ring.outBps],
|
||
}))
|
||
}
|
||
|
||
export function applyRingSnapshot(rows: Array<{ key: string; inBps: number[]; outBps: number[] }>): void {
|
||
rings.clear()
|
||
for (const row of rows) {
|
||
rings.set(row.key, { inBps: row.inBps, outBps: row.outBps })
|
||
}
|
||
}
|
||
|
||
async function persistListenerStats(force = false): Promise<boolean> {
|
||
if (
|
||
lastPersistedStats
|
||
&& lastPersistedStats.packetsReceived === packetsReceived
|
||
&& lastPersistedStats.lastDatagramAt === lastDatagramAt
|
||
&& lastPersistedStats.lastExporterIp === lastExporterIp
|
||
&& lastPersistedStats.lastError === lastError
|
||
) {
|
||
return false
|
||
}
|
||
const errorChanged = lastPersistedStats?.lastError !== lastError
|
||
const exporterChanged = lastPersistedStats?.lastExporterIp !== lastExporterIp
|
||
const now = Date.now()
|
||
if (!force && !errorChanged && !exporterChanged && now - lastStatsPersistAt < STATS_PERSIST_MS) {
|
||
return false
|
||
}
|
||
await dbQuery(`
|
||
UPDATE traffic_flow_settings
|
||
SET packets_received = @packetsReceived,
|
||
last_datagram_at = @lastDatagramAt,
|
||
last_exporter_ip = @lastExporterIp,
|
||
last_error = @lastError,
|
||
updated_at = @updatedAt
|
||
WHERE id = 1
|
||
`, {
|
||
packetsReceived,
|
||
lastDatagramAt,
|
||
lastExporterIp,
|
||
lastError,
|
||
updatedAt: nowIso(),
|
||
})
|
||
lastPersistedStats = {
|
||
packetsReceived,
|
||
lastDatagramAt,
|
||
lastExporterIp,
|
||
lastError,
|
||
}
|
||
lastStatsPersistAt = now
|
||
invalidateTrafficFlowSettingsCache()
|
||
return true
|
||
}
|
||
|
||
async function upsertMinuteAndDaily(): Promise<void> {
|
||
for (const [k, acc] of minuteRollup) {
|
||
const [serverIdRaw, bucketAt] = k.split("\0")
|
||
await ensureParentPartition("flow_minute_stats", bucketAt ?? "")
|
||
await dbQuery(`
|
||
INSERT INTO flow_minute_stats (
|
||
server_id, bucket_at, bytes, packets, unique_src, unique_dst, conversations
|
||
) VALUES (
|
||
@serverId, @bucketAt, @bytes, @packets, @uniqueSrc, @uniqueDst, @conversations
|
||
)
|
||
ON CONFLICT(server_id, bucket_at) DO UPDATE SET
|
||
bytes = flow_minute_stats.bytes + excluded.bytes,
|
||
packets = flow_minute_stats.packets + excluded.packets,
|
||
unique_src = GREATEST(flow_minute_stats.unique_src, excluded.unique_src),
|
||
unique_dst = GREATEST(flow_minute_stats.unique_dst, excluded.unique_dst),
|
||
conversations = flow_minute_stats.conversations + excluded.conversations
|
||
`, {
|
||
serverId: Number(serverIdRaw),
|
||
bucketAt,
|
||
bytes: acc.bytes,
|
||
packets: acc.packets,
|
||
uniqueSrc: acc.srcs.size,
|
||
uniqueDst: acc.dsts.size,
|
||
conversations: acc.conversations,
|
||
})
|
||
}
|
||
for (const [k, acc] of minuteDims) {
|
||
const [serverIdRaw, bucketAt, dim, key] = k.split("\0")
|
||
await ensureParentPartition("flow_minute_dims", bucketAt ?? "")
|
||
await ensureParentPartition("flow_daily_dims", dayKey(bucketAt ?? ""))
|
||
await dbQuery(`
|
||
INSERT INTO flow_minute_dims (server_id, bucket_at, dim, key, bytes, packets)
|
||
VALUES (@serverId, @bucketAt, @dim, @key, @bytes, @packets)
|
||
ON CONFLICT(server_id, bucket_at, dim, key) DO UPDATE SET
|
||
bytes = flow_minute_dims.bytes + excluded.bytes,
|
||
packets = flow_minute_dims.packets + excluded.packets
|
||
`, {
|
||
serverId: Number(serverIdRaw),
|
||
bucketAt,
|
||
dim,
|
||
key,
|
||
bytes: acc.bytes,
|
||
packets: acc.packets,
|
||
})
|
||
if (dim === "country" || dim === "service" || dim === "asn") {
|
||
await dbQuery(`
|
||
INSERT INTO flow_daily_dims (server_id, day, dim, key, bytes, packets)
|
||
VALUES (@serverId, @day, @dim, @key, @bytes, @packets)
|
||
ON CONFLICT(server_id, day, dim, key) DO UPDATE SET
|
||
bytes = flow_daily_dims.bytes + excluded.bytes,
|
||
packets = flow_daily_dims.packets + excluded.packets
|
||
`, {
|
||
serverId: Number(serverIdRaw),
|
||
day: dayKey(bucketAt ?? ""),
|
||
dim,
|
||
key,
|
||
bytes: acc.bytes,
|
||
packets: acc.packets,
|
||
})
|
||
}
|
||
}
|
||
minuteRollup.clear()
|
||
minuteDims.clear()
|
||
}
|
||
|
||
async function capDailyAsn(): Promise<void> {
|
||
const today = nowIso().slice(0, 10)
|
||
const rows = await dbAll<{ serverId: number; key: string; bytes: number; packets: number }>(`
|
||
SELECT server_id AS "serverId", key, bytes, packets
|
||
FROM flow_daily_dims
|
||
WHERE day = ? AND dim = 'asn'
|
||
ORDER BY server_id, bytes DESC
|
||
`, [today])
|
||
const byServer = new Map<number, typeof rows>()
|
||
for (const row of rows) {
|
||
const list = byServer.get(row.serverId) ?? []
|
||
list.push(row)
|
||
byServer.set(row.serverId, list)
|
||
}
|
||
for (const [serverId, list] of byServer) {
|
||
if (list.length <= DAILY_ASN_TOP) continue
|
||
let otherBytes = 0
|
||
let otherPackets = 0
|
||
for (const row of list.slice(DAILY_ASN_TOP)) {
|
||
if (row.key === "other") continue
|
||
otherBytes += row.bytes
|
||
otherPackets += row.packets
|
||
await dbQuery(
|
||
`DELETE FROM flow_daily_dims WHERE server_id = ? AND day = ? AND dim = 'asn' AND key = ?`,
|
||
[serverId, today, row.key],
|
||
)
|
||
}
|
||
if (otherBytes > 0) {
|
||
await ensureParentPartition("flow_daily_dims", today)
|
||
await dbQuery(`
|
||
INSERT INTO flow_daily_dims (server_id, day, dim, key, bytes, packets)
|
||
VALUES (?, ?, 'asn', 'other', ?, ?)
|
||
ON CONFLICT(server_id, day, dim, key) DO UPDATE SET
|
||
bytes = flow_daily_dims.bytes + excluded.bytes,
|
||
packets = flow_daily_dims.packets + excluded.packets
|
||
`, [serverId, today, otherBytes, otherPackets])
|
||
}
|
||
}
|
||
}
|
||
|
||
async function pruneStored(): Promise<void> {
|
||
const now = Date.now()
|
||
if (now - lastPruneAt < PRUNE_MS) return
|
||
lastPruneAt = now
|
||
const flowCutoff = new Date(now - retentionHours * 3600_000).toISOString()
|
||
const minuteCutoff = new Date(now - MINUTE_RETENTION_HOURS * 3600_000).toISOString()
|
||
const dailyCutoff = new Date(now - DAILY_RETENTION_DAYS * 86400_000).toISOString().slice(0, 10)
|
||
await dbQuery(`DELETE FROM flow_buckets WHERE bucket_at < ?`, [flowCutoff])
|
||
await dbQuery(`DELETE FROM flow_minute_stats WHERE bucket_at < ?`, [minuteCutoff])
|
||
await dbQuery(`DELETE FROM flow_minute_dims WHERE bucket_at < ?`, [minuteCutoff])
|
||
await dbQuery(`DELETE FROM flow_daily_dims WHERE day < ?`, [dailyCutoff])
|
||
|
||
const keep = Math.max(20, topN)
|
||
await dbQuery(`
|
||
DELETE FROM flow_buckets fb
|
||
USING (
|
||
SELECT server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface
|
||
FROM (
|
||
SELECT server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY server_id, bucket_at ORDER BY bytes DESC
|
||
) AS rn
|
||
FROM flow_buckets
|
||
) ranked
|
||
WHERE rn > ?
|
||
) drop_rows
|
||
WHERE fb.server_id = drop_rows.server_id
|
||
AND fb.bucket_at = drop_rows.bucket_at
|
||
AND fb.src = drop_rows.src
|
||
AND fb.dst = drop_rows.dst
|
||
AND fb.proto = drop_rows.proto
|
||
AND fb.src_port = drop_rows.src_port
|
||
AND fb.dst_port = drop_rows.dst_port
|
||
AND fb.in_iface = drop_rows.in_iface
|
||
`, [keep])
|
||
await pruneRipeSqlite(now)
|
||
}
|
||
|
||
function topNPending(rows: PendingFlowRow[]): PendingFlowRow[] {
|
||
const keep = Math.max(20, topN)
|
||
const groups = new Map<string, PendingFlowRow[]>()
|
||
for (const row of rows) {
|
||
const k = `${row.serverId}\0${row.bucketAt}`
|
||
const list = groups.get(k) ?? []
|
||
list.push(row)
|
||
groups.set(k, list)
|
||
}
|
||
const out: PendingFlowRow[] = []
|
||
for (const list of groups.values()) {
|
||
list.sort((a, b) => b.bytes - a.bytes)
|
||
out.push(...list.slice(0, keep))
|
||
}
|
||
return out
|
||
}
|
||
|
||
function persistDue(force: boolean, hasWork: boolean): boolean {
|
||
if (force) return true
|
||
if (!hasWork) return false
|
||
if (minuteBucketIso() !== lastFlushedMinute) return true
|
||
return Date.now() - lastPersistAt >= PERSIST_MS
|
||
}
|
||
|
||
async function upsertFlowBucketsBatch(rows: PendingFlowRow[]): Promise<void> {
|
||
if (rows.length === 0) return
|
||
const days = new Set(rows.map((r) => r.bucketAt))
|
||
for (const bucketAt of days) await ensureParentPartition("flow_buckets", bucketAt)
|
||
await pool.query({
|
||
text: `
|
||
INSERT INTO flow_buckets (
|
||
server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms
|
||
)
|
||
SELECT *
|
||
FROM UNNEST(
|
||
$1::bigint[],
|
||
$2::timestamptz[],
|
||
$3::inet[],
|
||
$4::inet[],
|
||
$5::smallint[],
|
||
$6::int[],
|
||
$7::int[],
|
||
$8::bigint[],
|
||
$9::bigint[],
|
||
$10::text[],
|
||
$11::text[],
|
||
$12::inet[],
|
||
$13::bigint[],
|
||
$14::bigint[]
|
||
) AS t(server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms)
|
||
ON CONFLICT (server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
|
||
DO UPDATE SET
|
||
bytes = flow_buckets.bytes + excluded.bytes,
|
||
packets = flow_buckets.packets + excluded.packets,
|
||
out_iface = CASE WHEN excluded.out_iface != '' THEN excluded.out_iface ELSE flow_buckets.out_iface END,
|
||
next_hop = COALESCE(excluded.next_hop, flow_buckets.next_hop),
|
||
flow_start_ms = CASE
|
||
WHEN excluded.flow_start_ms > 0 AND (flow_buckets.flow_start_ms = 0 OR excluded.flow_start_ms < flow_buckets.flow_start_ms)
|
||
THEN excluded.flow_start_ms ELSE flow_buckets.flow_start_ms END,
|
||
flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms)
|
||
`,
|
||
values: [
|
||
rows.map((r) => r.serverId),
|
||
rows.map((r) => r.bucketAt),
|
||
rows.map((r) => r.src),
|
||
rows.map((r) => r.dst),
|
||
rows.map((r) => r.proto),
|
||
rows.map((r) => r.srcPort),
|
||
rows.map((r) => r.dstPort),
|
||
rows.map((r) => r.bytes),
|
||
rows.map((r) => r.packets),
|
||
rows.map((r) => r.inIface),
|
||
rows.map((r) => r.outIface),
|
||
rows.map((r) => inetOrNull(r.nextHop)),
|
||
rows.map((r) => r.flowStartMs),
|
||
rows.map((r) => r.flowEndMs),
|
||
],
|
||
})
|
||
}
|
||
|
||
const FLOW_UPSERT_SQL = `
|
||
INSERT INTO flow_buckets (
|
||
server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms
|
||
) VALUES (
|
||
@serverId, @bucketAt, @src, @dst, @proto, @srcPort, @dstPort, @bytes, @packets, @inIface, @outIface, @nextHop, @flowStartMs, @flowEndMs
|
||
)
|
||
ON CONFLICT(server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
|
||
DO UPDATE SET
|
||
bytes = flow_buckets.bytes + excluded.bytes,
|
||
packets = flow_buckets.packets + excluded.packets,
|
||
out_iface = CASE WHEN excluded.out_iface != '' THEN excluded.out_iface ELSE flow_buckets.out_iface END,
|
||
next_hop = COALESCE(excluded.next_hop, flow_buckets.next_hop),
|
||
flow_start_ms = CASE
|
||
WHEN excluded.flow_start_ms > 0 AND (flow_buckets.flow_start_ms = 0 OR excluded.flow_start_ms < flow_buckets.flow_start_ms)
|
||
THEN excluded.flow_start_ms ELSE flow_buckets.flow_start_ms END,
|
||
flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms)
|
||
`
|
||
|
||
async function upsertFlowBuckets(rows: PendingFlowRow[]): Promise<number> {
|
||
if (rows.length === 0) return 0
|
||
try {
|
||
await upsertFlowBucketsBatch(rows)
|
||
return rows.length
|
||
} catch (err) {
|
||
const message = err instanceof Error ? err.message : String(err)
|
||
setEngineError(`flow_buckets: ${message}`)
|
||
let stored = 0
|
||
for (const r of rows) {
|
||
try {
|
||
await ensureParentPartition("flow_buckets", r.bucketAt)
|
||
await dbQuery(FLOW_UPSERT_SQL, flowUpsertParams(r))
|
||
stored += 1
|
||
} catch (rowErr) {
|
||
if (stored === 0 && lastError.startsWith("flow_buckets:")) {
|
||
const rowMsg = rowErr instanceof Error ? rowErr.message : String(rowErr)
|
||
setEngineError(`flow_buckets: ${rowMsg}`)
|
||
}
|
||
}
|
||
}
|
||
return stored
|
||
}
|
||
}
|
||
|
||
export async function flushPending(opts?: { force?: boolean }): Promise<void> {
|
||
pruneRecent()
|
||
rollFlowRings()
|
||
const force = Boolean(opts?.force)
|
||
const hasWork = pending.size > 0 || minuteRollup.size > 0 || minuteDims.size > 0 || factsPendingSize() > 0
|
||
const due = persistDue(force, hasWork)
|
||
try {
|
||
await persistListenerStats(force)
|
||
} catch {
|
||
/* settings row may be absent in unit tests */
|
||
}
|
||
|
||
if (!hasWork) {
|
||
if (force) {
|
||
try {
|
||
await pruneStored()
|
||
} catch {
|
||
/* prune best-effort */
|
||
}
|
||
}
|
||
lastFlushUsedTransaction = false
|
||
return
|
||
}
|
||
if (!due) {
|
||
lastFlushUsedTransaction = false
|
||
return
|
||
}
|
||
|
||
const sanitized: PendingFlowRow[] = []
|
||
let skippedInet = 0
|
||
for (const row of topNPending([...pending.values()].map(toPendingRow))) {
|
||
const clean = sanitizeFlowRow(row)
|
||
if (!clean) {
|
||
skippedInet += 1
|
||
continue
|
||
}
|
||
sanitized.push(clean)
|
||
}
|
||
pending.clear()
|
||
for (const row of sanitized) mergeInto(recent, row)
|
||
if (skippedInet > 0) {
|
||
setEngineError(`flow_buckets: пропуск ${skippedInet} строк с невалидным IP`)
|
||
}
|
||
|
||
lastFlushUsedTransaction = false
|
||
lastPersistAt = Date.now()
|
||
lastFlushedMinute = minuteBucketIso()
|
||
try {
|
||
const stored = await upsertFlowBuckets(sanitized)
|
||
lastFlushUsedTransaction = stored === sanitized.length
|
||
rowsStored += stored
|
||
if (stored > 0) bumpDataEpoch()
|
||
} catch (err) {
|
||
const message = err instanceof Error ? err.message : String(err)
|
||
setEngineError(`flow_buckets: ${message}`)
|
||
}
|
||
try {
|
||
await upsertMinuteAndDaily()
|
||
await capDailyAsn()
|
||
} catch {
|
||
/* rollup best-effort */
|
||
}
|
||
try {
|
||
await flushFlowFacts()
|
||
} catch {
|
||
/* statistics cube best-effort */
|
||
}
|
||
try {
|
||
await pruneStored()
|
||
} catch {
|
||
/* prune best-effort */
|
||
}
|
||
}
|
||
|
||
export function lastFlushUsedTransactionForTests(): boolean {
|
||
return lastFlushUsedTransaction
|
||
}
|
||
|
||
export async function flushPendingForTests(): Promise<void> {
|
||
await flushPending({ force: true })
|
||
}
|
||
|
||
export function onEngineTick(): void {
|
||
void flushPending()
|
||
}
|
||
|
||
export function ingestParsedFlowsForServerForTests(serverId: number, flows: ParsedFlowInput[]): void {
|
||
queueParsedFlows(serverId, flows)
|
||
rollFlowRings()
|
||
}
|
||
|
||
export function resetEngineForTests(): void {
|
||
pending.clear()
|
||
recent.clear()
|
||
tickAccum.clear()
|
||
rings.clear()
|
||
minuteRollup.clear()
|
||
minuteDims.clear()
|
||
resetFactsForTests()
|
||
packetsReceived = 0
|
||
lastExporterIp = null
|
||
lastError = ""
|
||
lastDatagramAt = null
|
||
dropped = 0
|
||
rowsStored = 0
|
||
lastFlushUsedTransaction = false
|
||
lastPruneAt = 0
|
||
lastPassiveCheckpointAt = Date.now()
|
||
lastPersistAt = 0
|
||
lastFlushedMinute = ""
|
||
lastStatsPersistAt = 0
|
||
lastPersistedStats = null
|
||
bumpDataEpoch()
|
||
pendingCap = MAX_PENDING
|
||
}
|
||
|
||
export function pendingSizeForTests(): number {
|
||
return pending.size
|
||
}
|
||
|
||
export function droppedForTests(): number {
|
||
return dropped
|
||
}
|
||
|
||
/** Снимок минутных dims (dim → key → bytes) для тестов обогащения потоков. */
|
||
export function minuteDimsSnapshotForTests(): Map<string, Map<string, { bytes: number; packets: number }>> {
|
||
const out = new Map<string, Map<string, { bytes: number; packets: number }>>()
|
||
for (const [k, acc] of minuteDims) {
|
||
// dimKey: serverId\0bucketAt\0dim\0key
|
||
const parts = k.split("\0")
|
||
const dim = parts[2] ?? ""
|
||
const key = parts.slice(3).join("\0")
|
||
let byKey = out.get(dim)
|
||
if (!byKey) {
|
||
byKey = new Map()
|
||
out.set(dim, byKey)
|
||
}
|
||
const prev = byKey.get(key)
|
||
byKey.set(key, {
|
||
bytes: (prev?.bytes ?? 0) + acc.bytes,
|
||
packets: (prev?.packets ?? 0) + acc.packets,
|
||
})
|
||
}
|
||
return out
|
||
}
|