Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0e5fb065e2 |
@@ -14,7 +14,7 @@
|
||||
"test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts",
|
||||
"test:wireguard": "npx tsx src/services/wireguard-config.test.ts",
|
||||
"test:traffic-rate": "tsx src/services/traffic-rate.test.ts",
|
||||
"test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-hardening.test.ts && tsx src/services/traffic-flow-purge.test.ts",
|
||||
"test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-hardening.test.ts && tsx src/services/traffic-flow-purge.test.ts && tsx src/services/sqlite-write-opt.test.ts",
|
||||
"test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts"
|
||||
},
|
||||
"dependencies": {
|
||||
|
||||
+68
-1
@@ -8,18 +8,85 @@ import { env } from "../config.js"
|
||||
import * as schema from "./schema.js"
|
||||
|
||||
export const SQLITE_BUSY_TIMEOUT_MS = 5000
|
||||
/** ~16 MiB page cache (negative = KiB). */
|
||||
export const SQLITE_CACHE_SIZE_KIB = 16_000
|
||||
export const SQLITE_WAL_AUTOCHECKPOINT_PAGES = 1000
|
||||
|
||||
export interface SqliteWriteStats {
|
||||
insert: number
|
||||
update: number
|
||||
delete: number
|
||||
walCheckpoint: number
|
||||
}
|
||||
|
||||
let writeTrace: SqliteWriteStats | null = null
|
||||
|
||||
function classifyWriteSql(sql: string): keyof Omit<SqliteWriteStats, "walCheckpoint"> | null {
|
||||
const head = sql.trimStart().slice(0, 12).toUpperCase()
|
||||
if (head.startsWith("INSERT")) return "insert"
|
||||
if (head.startsWith("UPDATE")) return "update"
|
||||
if (head.startsWith("DELETE")) return "delete"
|
||||
return null
|
||||
}
|
||||
|
||||
function installSqliteWriteTrace(handle: SqliteHandle): SqliteHandle {
|
||||
const origPrepare = handle.prepare.bind(handle)
|
||||
handle.prepare = ((sql: string) => {
|
||||
const stmt = origPrepare(sql)
|
||||
const kind = classifyWriteSql(sql)
|
||||
if (!kind) return stmt
|
||||
const origRun = stmt.run.bind(stmt)
|
||||
stmt.run = ((...args: unknown[]) => {
|
||||
if (writeTrace) writeTrace[kind] += 1
|
||||
return origRun(...args)
|
||||
}) as typeof stmt.run
|
||||
return stmt
|
||||
}) as typeof handle.prepare
|
||||
|
||||
const origExec = handle.exec.bind(handle)
|
||||
handle.exec = ((sql: string) => {
|
||||
if (writeTrace) {
|
||||
for (const part of sql.split(";")) {
|
||||
const kind = classifyWriteSql(part)
|
||||
if (kind) writeTrace[kind] += 1
|
||||
}
|
||||
}
|
||||
return origExec(sql)
|
||||
}) as typeof handle.exec
|
||||
|
||||
const origPragma = handle.pragma.bind(handle)
|
||||
handle.pragma = ((source: string, options?: { simple?: boolean }) => {
|
||||
if (writeTrace && /wal_checkpoint/i.test(source)) writeTrace.walCheckpoint += 1
|
||||
return origPragma(source, options as never)
|
||||
}) as typeof handle.pragma
|
||||
|
||||
return handle
|
||||
}
|
||||
|
||||
export function countSqliteWrites<T>(fn: () => T): { result: T; stats: SqliteWriteStats } {
|
||||
const stats: SqliteWriteStats = { insert: 0, update: 0, delete: 0, walCheckpoint: 0 }
|
||||
writeTrace = stats
|
||||
try {
|
||||
return { result: fn(), stats }
|
||||
} finally {
|
||||
writeTrace = null
|
||||
}
|
||||
}
|
||||
|
||||
export function applySqlitePragmas(handle: SqliteHandle): void {
|
||||
handle.pragma("journal_mode = WAL")
|
||||
handle.pragma("foreign_keys = ON")
|
||||
handle.pragma(`busy_timeout = ${SQLITE_BUSY_TIMEOUT_MS}`)
|
||||
handle.pragma("synchronous = NORMAL")
|
||||
handle.pragma(`wal_autocheckpoint = ${SQLITE_WAL_AUTOCHECKPOINT_PAGES}`)
|
||||
handle.pragma("temp_store = MEMORY")
|
||||
handle.pragma(`cache_size = -${SQLITE_CACHE_SIZE_KIB}`)
|
||||
}
|
||||
|
||||
function openSqlite(): SqliteHandle {
|
||||
const handle = new Database(env.DATABASE_PATH)
|
||||
applySqlitePragmas(handle)
|
||||
return handle
|
||||
return installSqliteWriteTrace(handle)
|
||||
}
|
||||
|
||||
let sqlite = openSqlite()
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { desc, eq } from "drizzle-orm"
|
||||
import { db } from "../../../db/index.js"
|
||||
import { serverSnapshots, servers } from "../../../db/schema.js"
|
||||
import { invalidateFlowCatalogCache } from "../../../services/traffic-flow-topology.js"
|
||||
|
||||
export type ServerRow = typeof servers.$inferSelect
|
||||
export type SnapshotRow = typeof serverSnapshots.$inferSelect
|
||||
@@ -17,6 +18,7 @@ export function createServerRow(
|
||||
values: Omit<typeof servers.$inferInsert, "id">,
|
||||
): ServerRow {
|
||||
const [inserted] = db.insert(servers).values(values).returning().all()
|
||||
invalidateFlowCatalogCache()
|
||||
return inserted
|
||||
}
|
||||
|
||||
@@ -25,11 +27,13 @@ export function updateServerRowById(
|
||||
values: Partial<ServerRow>,
|
||||
): ServerRow {
|
||||
const [updated] = db.update(servers).set(values).where(eq(servers.id, id)).returning().all()
|
||||
invalidateFlowCatalogCache()
|
||||
return updated
|
||||
}
|
||||
|
||||
export function deleteServerRowById(id: number): void {
|
||||
db.delete(servers).where(eq(servers.id, id)).run()
|
||||
invalidateFlowCatalogCache()
|
||||
}
|
||||
|
||||
export function listSnapshotsByServerId(serverId: number, limit: number): SnapshotRow[] {
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { and, eq } from "drizzle-orm"
|
||||
import { and, count, eq } from "drizzle-orm"
|
||||
import { db } from "../../../db/index.js"
|
||||
import { appUsers, userInterfaceBindings } from "../../../db/schema.js"
|
||||
import { invalidateFlowCatalogCache } from "../../../services/traffic-flow-topology.js"
|
||||
|
||||
export type AppUserRow = typeof appUsers.$inferSelect
|
||||
export type BindingRow = typeof userInterfaceBindings.$inferSelect
|
||||
@@ -19,6 +20,7 @@ export function getUserRowByLogin(login: string): AppUserRow | undefined {
|
||||
|
||||
export function createUserRow(values: typeof appUsers.$inferInsert): AppUserRow {
|
||||
const [inserted] = db.insert(appUsers).values(values).returning().all()
|
||||
invalidateFlowCatalogCache()
|
||||
return inserted
|
||||
}
|
||||
|
||||
@@ -27,11 +29,13 @@ export function updateUserRowById(
|
||||
values: Partial<AppUserRow>,
|
||||
): AppUserRow {
|
||||
const [updated] = db.update(appUsers).set(values).where(eq(appUsers.id, id)).returning().all()
|
||||
invalidateFlowCatalogCache()
|
||||
return updated
|
||||
}
|
||||
|
||||
export function deleteUserRowById(id: string): void {
|
||||
db.delete(appUsers).where(eq(appUsers.id, id)).run()
|
||||
invalidateFlowCatalogCache()
|
||||
}
|
||||
|
||||
export function listBindingRows(): BindingRow[] {
|
||||
@@ -65,13 +69,15 @@ export function getBindingByServerIfacePeer(
|
||||
|
||||
export function createBindingRow(values: typeof userInterfaceBindings.$inferInsert): BindingRow {
|
||||
const [inserted] = db.insert(userInterfaceBindings).values(values).returning().all()
|
||||
invalidateFlowCatalogCache()
|
||||
return inserted
|
||||
}
|
||||
|
||||
export function deleteBindingRowById(id: string): void {
|
||||
db.delete(userInterfaceBindings).where(eq(userInterfaceBindings.id, id)).run()
|
||||
invalidateFlowCatalogCache()
|
||||
}
|
||||
|
||||
export function countUserRows(): number {
|
||||
return db.select().from(appUsers).all().length
|
||||
return db.select({ n: count() }).from(appUsers).all()[0]?.n ?? 0
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
|
||||
import { count } from "drizzle-orm"
|
||||
import { listCertificatesFromServers } from "../services/certificates-service.js"
|
||||
import { countWireGuardInterfaces } from "../services/wireguard-live.js"
|
||||
import { db } from "../db/index.js"
|
||||
@@ -11,27 +12,22 @@ import {
|
||||
} from "../db/schema.js"
|
||||
import { listUsers } from "../modules/users/service/users-service.js"
|
||||
|
||||
function tableCount(table: typeof servers | typeof filterRules | typeof uptimeProbes | typeof uptimeSpeedProbes | typeof recursiveRoutes): number {
|
||||
return db.select({ n: count() }).from(table).all()[0]?.n ?? 0
|
||||
}
|
||||
|
||||
const sidebarCountsRoutes: FastifyPluginAsyncZod = async (app) => {
|
||||
app.get("/sidebar-counts", async (_req, reply) => {
|
||||
const [
|
||||
serversTotal,
|
||||
filterRulesTotal,
|
||||
uptimeProbesTotal,
|
||||
uptimeSpeedProbesTotal,
|
||||
recursiveRoutesTotal,
|
||||
certificatesTotal,
|
||||
wireguardTotal,
|
||||
usersTotal,
|
||||
] = await Promise.all([
|
||||
Promise.resolve(db.select().from(servers).all().length),
|
||||
Promise.resolve(db.select().from(filterRules).all().length),
|
||||
Promise.resolve(db.select().from(uptimeProbes).all().length),
|
||||
Promise.resolve(db.select().from(uptimeSpeedProbes).all().length),
|
||||
Promise.resolve(db.select().from(recursiveRoutes).all().length),
|
||||
const serversTotal = tableCount(servers)
|
||||
const filterRulesTotal = tableCount(filterRules)
|
||||
const uptimeProbesTotal = tableCount(uptimeProbes)
|
||||
const uptimeSpeedProbesTotal = tableCount(uptimeSpeedProbes)
|
||||
const recursiveRoutesTotal = tableCount(recursiveRoutes)
|
||||
const [certificatesTotal, wireguardTotal] = await Promise.all([
|
||||
listCertificatesFromServers().then((res) => res.certificates.length),
|
||||
countWireGuardInterfaces().catch(() => 0),
|
||||
Promise.resolve(listUsers().length),
|
||||
])
|
||||
const usersTotal = listUsers().length
|
||||
|
||||
return reply.send({
|
||||
servers: serversTotal,
|
||||
|
||||
@@ -37,11 +37,12 @@ export function savePrevLiveMap(kind: PrevLiveKind, map: PrevLiveStringMap) {
|
||||
.limit(1)
|
||||
.all()[0]
|
||||
if (existing) {
|
||||
if (existing.payloadJson === payloadJson) return
|
||||
db.update(alertEnginePrevLive)
|
||||
.set({ payloadJson, updatedAt: new Date().toISOString() })
|
||||
.where(eq(alertEnginePrevLive.kind, kind))
|
||||
.run()
|
||||
} else {
|
||||
db.insert(alertEnginePrevLive).values({ kind, payloadJson, updatedAt: new Date().toISOString() }).run()
|
||||
return
|
||||
}
|
||||
db.insert(alertEnginePrevLive).values({ kind, payloadJson, updatedAt: new Date().toISOString() }).run()
|
||||
}
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
import { db, sqliteDatabase } from "../db/index.js"
|
||||
import { alertBgpPeerSamples, alertGreTunnelSamples } from "../db/schema.js"
|
||||
import { bgpPeerAlertKey, fetchBgpSessionsForAlerts } from "./bgp-peers-live.js"
|
||||
import { fetchGreTunnelLiveRows } from "./gre-tunnels-live.js"
|
||||
import type { GreBgpSnapshotRunSnapshot } from "../types/scheduler-run-snapshot.js"
|
||||
@@ -12,8 +10,8 @@ export function getGreBgpSnapshotCollectorState(): { running: boolean } {
|
||||
}
|
||||
|
||||
/**
|
||||
* Один опрос GRE + BGP по включённым серверам и запись строк в SQLite для `buildSignalSnapshot`.
|
||||
* Движок оповещений больше не дублирует эти REST-запросы.
|
||||
* Один опрос GRE + BGP по включённым серверам.
|
||||
* Снимок для алертов живёт в `scheduler_runs.result_json` (`greTunnels` / `bgpPeers`).
|
||||
*/
|
||||
export async function collectGreBgpSnapshotOnce(): Promise<GreBgpSnapshotRunSnapshot> {
|
||||
const sampledAt = new Date().toISOString()
|
||||
@@ -62,32 +60,8 @@ export async function collectGreBgpSnapshotOnce(): Promise<GreBgpSnapshotRunSnap
|
||||
const bgpRows = bgpSettled.status === "fulfilled" ? bgpSettled.value : []
|
||||
snapshot.greTunnels = greRows.map((r) => ({ targetLabel: r.targetLabel, status: r.status }))
|
||||
snapshot.bgpPeers = bgpRows.map((s) => ({ key: bgpPeerAlertKey(s), state: s.state }))
|
||||
|
||||
db.transaction((tx) => {
|
||||
for (const r of greRows) {
|
||||
tx.insert(alertGreTunnelSamples).values({
|
||||
sampledAt,
|
||||
targetLabel: r.targetLabel,
|
||||
status: r.status,
|
||||
}).run()
|
||||
snapshot.greWritten += 1
|
||||
}
|
||||
for (const s of bgpRows) {
|
||||
tx.insert(alertBgpPeerSamples).values({
|
||||
sampledAt,
|
||||
peerKey: bgpPeerAlertKey(s),
|
||||
state: s.state,
|
||||
}).run()
|
||||
snapshot.bgpWritten += 1
|
||||
}
|
||||
})
|
||||
|
||||
sqliteDatabase
|
||||
.prepare(`DELETE FROM alert_gre_tunnel_samples WHERE sampled_at < datetime('now', '-30 days')`)
|
||||
.run()
|
||||
sqliteDatabase
|
||||
.prepare(`DELETE FROM alert_bgp_peer_samples WHERE sampled_at < datetime('now', '-30 days')`)
|
||||
.run()
|
||||
snapshot.greWritten = greRows.length
|
||||
snapshot.bgpWritten = bgpRows.length
|
||||
|
||||
if (errors.length) snapshot.errors = errors
|
||||
} catch (e) {
|
||||
|
||||
@@ -5,6 +5,7 @@ import type { SnapshotInsert } from "../db/schema.js"
|
||||
import type { SnapshotRead } from "../types/server.js"
|
||||
import { MikrotikClient } from "./mikrotik.js"
|
||||
import { parseRosCpuLoadPercent, parseRosDataSizeBytes } from "./ros-metric-parse.js"
|
||||
import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js"
|
||||
|
||||
// ── pollServer ─────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -64,11 +65,13 @@ export async function pollServer(serverId: number): Promise<SnapshotRead> {
|
||||
rawIpAddresses: JSON.stringify(addresses),
|
||||
} satisfies Partial<SnapshotInsert>)
|
||||
|
||||
// Keep server.name in sync with RouterOS identity
|
||||
db.update(servers)
|
||||
.set({ name: identity.name, updatedAt: now })
|
||||
.where(eq(servers.id, serverId))
|
||||
.run()
|
||||
if ((identity.name || "") !== (server.name || "")) {
|
||||
db.update(servers)
|
||||
.set({ name: identity.name, updatedAt: now })
|
||||
.where(eq(servers.id, serverId))
|
||||
.run()
|
||||
invalidateFlowCatalogCache()
|
||||
}
|
||||
|
||||
} catch (err) {
|
||||
// Log but don't throw — we still persist the offline snapshot
|
||||
|
||||
@@ -7,7 +7,7 @@
|
||||
* 3. `uptime_speed` — ниже (BW-test, тяжёлый); локи узлов — `withBtestNodeLocks` в speed-сервисе.
|
||||
*
|
||||
* **Оповещения (`alert_engine`):** читают SQLite после джоб сбора (см. `buildSignalSnapshot`), в т.ч.
|
||||
* `gre_bgp` → `alert_gre_tunnel_samples` / `alert_bgp_peer_samples`. После успешного завершения джоб
|
||||
* `gre_bgp` → snapshot в `scheduler_runs.result_json` (`greTunnels` / `bgpPeers`). После успешного завершения джоб
|
||||
* `traffic`, `uptime_*`, `servers_rest_ping`, `gre_bgp` планируется **дополнительный** прогон движка
|
||||
* (debounce), см. [`alert-collector-hooks.ts`](./alert-collector-hooks.ts); любой другой писатель сэмплов
|
||||
* для снимка оповещений тоже должен вызывать `scheduleAlertEngineAfterDataCollectors()` после коммита.
|
||||
@@ -17,7 +17,7 @@
|
||||
*/
|
||||
import { desc, eq, lt } from "drizzle-orm"
|
||||
import { db } from "../db/index.js"
|
||||
import { schedulerRuns } from "../db/schema.js"
|
||||
import { events, schedulerRuns } from "../db/schema.js"
|
||||
import type { SchedulerRunSnapshot } from "../types/scheduler-run-snapshot.js"
|
||||
import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
|
||||
import {
|
||||
@@ -71,6 +71,20 @@ const timers = new Map<string, ReturnType<typeof setInterval>>()
|
||||
|
||||
const RUN_LOG_RETENTION_MS = 30 * 24 * 60 * 60 * 1000
|
||||
|
||||
const QUIET_SCHEDULER_OK_JOBS = new Set<SchedulerJobKey>([
|
||||
"traffic",
|
||||
"servers_rest_ping",
|
||||
"uptime_resources",
|
||||
"uptime_ping",
|
||||
"uptime_speed",
|
||||
"gre_bgp",
|
||||
"alert_engine",
|
||||
])
|
||||
|
||||
export function shouldAppendSchedulerOkEvent(jobKey: SchedulerJobKey): boolean {
|
||||
return !QUIET_SCHEDULER_OK_JOBS.has(jobKey)
|
||||
}
|
||||
|
||||
function newRunId(): string {
|
||||
return `sch-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`
|
||||
}
|
||||
@@ -84,18 +98,21 @@ function appendSchedulerRun(row: {
|
||||
durationMs: number
|
||||
result?: SchedulerRunSnapshot | null
|
||||
}) {
|
||||
db.insert(schedulerRuns).values({
|
||||
id: newRunId(),
|
||||
jobKey: row.jobKey,
|
||||
startedAt: row.startedAt,
|
||||
finishedAt: row.finishedAt,
|
||||
status: row.status,
|
||||
error: row.error,
|
||||
durationMs: row.durationMs,
|
||||
resultJson: row.result ? JSON.stringify(row.result) : null,
|
||||
}).run()
|
||||
const cutoff = new Date(Date.now() - RUN_LOG_RETENTION_MS).toISOString()
|
||||
db.delete(schedulerRuns).where(lt(schedulerRuns.finishedAt, cutoff)).run()
|
||||
db.transaction((tx) => {
|
||||
tx.insert(schedulerRuns).values({
|
||||
id: newRunId(),
|
||||
jobKey: row.jobKey,
|
||||
startedAt: row.startedAt,
|
||||
finishedAt: row.finishedAt,
|
||||
status: row.status,
|
||||
error: row.error,
|
||||
durationMs: row.durationMs,
|
||||
resultJson: row.result ? JSON.stringify(row.result) : null,
|
||||
}).run()
|
||||
tx.delete(schedulerRuns).where(lt(schedulerRuns.finishedAt, cutoff)).run()
|
||||
tx.delete(events).where(lt(events.createdAt, cutoff)).run()
|
||||
})
|
||||
}
|
||||
|
||||
async function runSchedulerJobBody(jobKey: SchedulerJobKey): Promise<void> {
|
||||
@@ -159,19 +176,21 @@ async function runSchedulerJobBody(jobKey: SchedulerJobKey): Promise<void> {
|
||||
durationMs: Date.now() - startedAt,
|
||||
result: snapshot ?? null,
|
||||
})
|
||||
appendEvent({
|
||||
level: "info",
|
||||
eventType: "scheduler.job.ok",
|
||||
sourceModule: "scheduler",
|
||||
title: "Задача планировщика завершена",
|
||||
message: `${jobKey}: выполнено за ${Date.now() - startedAt} мс`,
|
||||
entityType: "job",
|
||||
entityId: jobKey,
|
||||
payload: {
|
||||
startedAt: startedIso,
|
||||
finishedAt: finishedIso,
|
||||
},
|
||||
})
|
||||
if (shouldAppendSchedulerOkEvent(jobKey)) {
|
||||
appendEvent({
|
||||
level: "info",
|
||||
eventType: "scheduler.job.ok",
|
||||
sourceModule: "scheduler",
|
||||
title: "Задача планировщика завершена",
|
||||
message: `${jobKey}: выполнено за ${Date.now() - startedAt} мс`,
|
||||
entityType: "job",
|
||||
entityId: jobKey,
|
||||
payload: {
|
||||
startedAt: startedIso,
|
||||
finishedAt: finishedIso,
|
||||
},
|
||||
})
|
||||
}
|
||||
if (
|
||||
jobKey === "traffic" ||
|
||||
jobKey === "servers_rest_ping" ||
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { randomUUID } from "node:crypto"
|
||||
import { countSqliteWrites, sqliteDatabase } from "../db/index.js"
|
||||
import { events } from "../db/schema.js"
|
||||
import { db } from "../db/index.js"
|
||||
import {
|
||||
attachEngineSqlite,
|
||||
bumpPacketMeta,
|
||||
configureEngine,
|
||||
flushPending,
|
||||
ingestParsedFlowsForServerForTests,
|
||||
resetEngineForTests,
|
||||
setEngineError,
|
||||
} from "./traffic-flow-engine.js"
|
||||
import { collectGreBgpSnapshotOnce } from "./gre-bgp-snapshot-collector.js"
|
||||
import { shouldAppendSchedulerOkEvent, executeSchedulerJob } from "./scheduler.js"
|
||||
import { savePrevLiveMap, loadPrevLiveMap } from "./alert-engine/prev-live-store.js"
|
||||
import {
|
||||
invalidateFlowCatalogCache,
|
||||
loadFlowTopology,
|
||||
seedFlowTopologyForTests,
|
||||
} from "./traffic-flow-topology.js"
|
||||
|
||||
resetEngineForTests()
|
||||
attachEngineSqlite(sqliteDatabase)
|
||||
seedFlowTopologyForTests(null)
|
||||
invalidateFlowCatalogCache()
|
||||
|
||||
const idle1 = countSqliteWrites(() => {
|
||||
flushPending()
|
||||
})
|
||||
assert.equal(idle1.stats.walCheckpoint, 0)
|
||||
assert.ok(idle1.stats.update >= 1, "first idle flush persists listener stats")
|
||||
|
||||
const idle2 = countSqliteWrites(() => {
|
||||
flushPending()
|
||||
})
|
||||
assert.equal(idle2.stats.update, 0, "unchanged listener stats skip UPDATE")
|
||||
assert.equal(idle2.stats.walCheckpoint, 0)
|
||||
|
||||
bumpPacketMeta("203.0.113.9")
|
||||
const changed = countSqliteWrites(() => {
|
||||
flushPending()
|
||||
})
|
||||
assert.equal(changed.stats.update, 1, "changed packets persist once")
|
||||
assert.equal(changed.stats.walCheckpoint, 0)
|
||||
|
||||
setEngineError("boom")
|
||||
const errWrite = countSqliteWrites(() => {
|
||||
flushPending()
|
||||
})
|
||||
assert.equal(errWrite.stats.update, 1)
|
||||
setEngineError("")
|
||||
flushPending()
|
||||
|
||||
resetEngineForTests()
|
||||
configureEngine({ topN: 20 })
|
||||
ingestParsedFlowsForServerForTests(9, [{
|
||||
src: "10.1.1.1",
|
||||
dst: "8.8.8.8",
|
||||
proto: 6,
|
||||
srcPort: 40000,
|
||||
dstPort: 443,
|
||||
bytes: 100,
|
||||
packets: 1,
|
||||
inIface: "2",
|
||||
outIface: "",
|
||||
}])
|
||||
const withData = countSqliteWrites(() => {
|
||||
flushPending()
|
||||
})
|
||||
assert.equal(withData.stats.walCheckpoint, 0, "flush with data must not TRUNCATE WAL")
|
||||
assert.ok(withData.stats.insert >= 1, "flow upsert writes")
|
||||
|
||||
sqliteDatabase.prepare(`DELETE FROM flow_buckets WHERE server_id = 9`).run()
|
||||
sqliteDatabase.prepare(`DELETE FROM flow_minute_stats WHERE server_id = 9`).run()
|
||||
sqliteDatabase.prepare(`DELETE FROM flow_minute_dims WHERE server_id = 9`).run()
|
||||
sqliteDatabase.prepare(`DELETE FROM flow_daily_dims WHERE server_id = 9`).run()
|
||||
|
||||
const plan = sqliteDatabase.prepare(`
|
||||
EXPLAIN QUERY PLAN
|
||||
SELECT id FROM flow_buckets WHERE bucket_at >= ? ORDER BY bytes DESC LIMIT 100
|
||||
`).all("2000-01-01T00:00:00.000Z") as Array<{ detail?: string }>
|
||||
const planText = plan.map((p) => String(p.detail ?? "")).join(" | ")
|
||||
assert.ok(planText.length > 0, "EXPLAIN QUERY PLAN returned rows")
|
||||
|
||||
invalidateFlowCatalogCache()
|
||||
seedFlowTopologyForTests(null)
|
||||
const topoA = loadFlowTopology()
|
||||
const topoB = loadFlowTopology()
|
||||
assert.equal(topoA, topoB, "topology cache returns same object")
|
||||
invalidateFlowCatalogCache()
|
||||
const topoC = loadFlowTopology()
|
||||
assert.notEqual(topoA, topoC, "invalidate rebuilds topology")
|
||||
|
||||
assert.equal(shouldAppendSchedulerOkEvent("traffic"), false)
|
||||
assert.equal(shouldAppendSchedulerOkEvent("alert_engine"), false)
|
||||
assert.equal(shouldAppendSchedulerOkEvent("gre_bgp"), false)
|
||||
assert.equal(shouldAppendSchedulerOkEvent("backups"), true)
|
||||
assert.equal(shouldAppendSchedulerOkEvent("certificates_renew"), true)
|
||||
assert.equal(shouldAppendSchedulerOkEvent("internet_path"), true)
|
||||
|
||||
savePrevLiveMap("gre", { a: "up" })
|
||||
const prevSame = countSqliteWrites(() => {
|
||||
savePrevLiveMap("gre", { a: "up" })
|
||||
})
|
||||
assert.equal(prevSame.stats.update, 0)
|
||||
assert.equal(prevSame.stats.insert, 0)
|
||||
savePrevLiveMap("gre", { a: "down" })
|
||||
assert.equal(loadPrevLiveMap("gre").a, "down")
|
||||
savePrevLiveMap("gre", { a: "up" })
|
||||
|
||||
const greBefore = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_gre_tunnel_samples`).get() as { n: number }
|
||||
const bgpBefore = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_bgp_peer_samples`).get() as { n: number }
|
||||
await collectGreBgpSnapshotOnce()
|
||||
const greAfter = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_gre_tunnel_samples`).get() as { n: number }
|
||||
const bgpAfter = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_bgp_peer_samples`).get() as { n: number }
|
||||
assert.equal(greAfter.n, greBefore.n)
|
||||
assert.equal(bgpAfter.n, bgpBefore.n)
|
||||
|
||||
const oldId = `evt-old-${randomUUID()}`
|
||||
db.insert(events).values({
|
||||
id: oldId,
|
||||
createdAt: new Date(Date.now() - 40 * 24 * 60 * 60 * 1000).toISOString(),
|
||||
level: "info",
|
||||
eventType: "test.retention",
|
||||
sourceModule: "system",
|
||||
title: "old",
|
||||
message: "old",
|
||||
}).run()
|
||||
const eventsBefore = sqliteDatabase.prepare(
|
||||
`SELECT COUNT(*) AS n FROM events WHERE event_type = 'scheduler.job.ok' AND entity_id = 'alert_engine'`,
|
||||
).get() as { n: number }
|
||||
await executeSchedulerJob("alert_engine")
|
||||
const oldGone = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM events WHERE id = ?`).get(oldId) as { n: number }
|
||||
assert.equal(oldGone.n, 0, "events older than 30 days are purged")
|
||||
const eventsAfter = sqliteDatabase.prepare(
|
||||
`SELECT COUNT(*) AS n FROM events WHERE event_type = 'scheduler.job.ok' AND entity_id = 'alert_engine'`,
|
||||
).get() as { n: number }
|
||||
assert.equal(eventsAfter.n, eventsBefore.n, "quiet jobs do not append scheduler.job.ok")
|
||||
|
||||
resetEngineForTests()
|
||||
console.log("sqlite-write-opt.test.ts: ok")
|
||||
console.log("EXPLAIN listStoredFlowRows:", planText)
|
||||
@@ -1,6 +1,6 @@
|
||||
import { eq } from "drizzle-orm"
|
||||
import { db, sqliteDatabase } from "../db/index.js"
|
||||
import { appUsers, servers, userInterfaceBindings } from "../db/schema.js"
|
||||
import { appUsers, userInterfaceBindings } from "../db/schema.js"
|
||||
import type {
|
||||
FlowAnalyticsDto,
|
||||
FlowBreakdownRow,
|
||||
@@ -21,7 +21,7 @@ import {
|
||||
listFlowRowsForWindow,
|
||||
type PendingFlowRow,
|
||||
} from "./traffic-flow-ingest.js"
|
||||
import { MAX_PENDING, RING_OVERLAY } from "./traffic-flow-engine.js"
|
||||
import { flowDataEpoch, MAX_PENDING, RING_OVERLAY } from "./traffic-flow-engine.js"
|
||||
import { resolveIfaceName } from "./traffic-flow-ifaces.js"
|
||||
import { getTrafficFlowSettingsRow, listHostPeers } from "./traffic-flow-settings.js"
|
||||
import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
|
||||
@@ -33,6 +33,7 @@ import { classifyFlowPlane, flowBps, shouldKeepPlane } from "./traffic-flow-plan
|
||||
import { pickInternetPeer } from "./traffic-flow-ip.js"
|
||||
import {
|
||||
enGreIfaceNames,
|
||||
getServerCatalog,
|
||||
latestWireBps,
|
||||
loadFlowTopology,
|
||||
resolveClient,
|
||||
@@ -142,14 +143,46 @@ function topLabel(map: Map<string, { bytes: number; packets: number; label?: str
|
||||
}
|
||||
|
||||
export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
|
||||
const key = analyticsQueryKey(q)
|
||||
const now = Date.now()
|
||||
if (analyticsCache && analyticsCache.key === key && now - analyticsCache.at < ANALYTICS_CACHE_TTL_MS) {
|
||||
return analyticsCache.dto
|
||||
}
|
||||
const dto = buildFlowAnalyticsUncached(q)
|
||||
analyticsCache = { key, at: now, dto }
|
||||
return dto
|
||||
}
|
||||
|
||||
export function resetFlowAnalyticsCacheForTests(): void {
|
||||
analyticsCache = null
|
||||
}
|
||||
|
||||
function analyticsQueryKey(q: FlowAnalyticsQuery): string {
|
||||
return JSON.stringify({
|
||||
epoch: flowDataEpoch(),
|
||||
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,
|
||||
skipHeavy: Boolean(q.skipHeavy),
|
||||
})
|
||||
}
|
||||
|
||||
const ANALYTICS_CACHE_TTL_MS = 2000
|
||||
let analyticsCache: { key: string; at: number; dto: FlowAnalyticsDto } | null = null
|
||||
|
||||
function buildFlowAnalyticsUncached(q: FlowAnalyticsQuery): FlowAnalyticsDto {
|
||||
const settings = getTrafficFlowSettingsRow()
|
||||
const top = Math.min(50, Math.max(10, settings.topN))
|
||||
const windowSec = Math.max(60, q.minutes * 60)
|
||||
const raw = listFlowRowsForWindow(q.minutes)
|
||||
const allow = q.userId ? userIfaceAllow(q.userId) : null
|
||||
const serverRows = db.select().from(servers).all()
|
||||
const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host]))
|
||||
const countryById = new Map(serverRows.map((s) => [s.id, (s.country || "").toUpperCase() || "UN"]))
|
||||
const catalog = getServerCatalog()
|
||||
const nameById = new Map([...catalog.byId].map(([id, s]) => [id, s.name]))
|
||||
const countryById = new Map([...catalog.byId].map(([id, s]) => [id, s.country]))
|
||||
const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : ""
|
||||
const wantDedup = q.dedup !== false && !ifaceFilter
|
||||
const excludeMesh = q.excludeMesh !== false
|
||||
@@ -478,19 +511,19 @@ export function listFlowExporters(minutes: number): FlowExportersDto {
|
||||
const { bytes, sessions } = summarizeByServer(rows)
|
||||
const ids = new Set<number>([...bytes.keys()])
|
||||
for (const p of listHostPeers()) ids.add(p.serverId)
|
||||
const serverRows = db.select().from(servers).all()
|
||||
const catalog = getServerCatalog()
|
||||
const emptySeries = Array(60).fill(0) as number[]
|
||||
const exporters = serverRows
|
||||
const exporters = catalog.list
|
||||
.filter((s) => ids.has(s.id))
|
||||
.map((s) => {
|
||||
const ring = getRingMbps(s.id, "__all__")
|
||||
const total = bytes.get(s.id) ?? 0
|
||||
return {
|
||||
id: String(s.id),
|
||||
name: s.name || s.host,
|
||||
name: s.name,
|
||||
subtitle: s.host,
|
||||
site: s.site || "—",
|
||||
country: s.country || "UN",
|
||||
site: s.site,
|
||||
country: s.country,
|
||||
status: snapshotStatus(s.id),
|
||||
rxNow: ring.rxNow || (total * 8) / Math.max(60, minutes * 60) / 1_000_000,
|
||||
txNow: ring.txNow,
|
||||
|
||||
@@ -4,7 +4,8 @@ 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, lookupRipeCached } from "./traffic-flow-ripe.js"
|
||||
import { enqueueRipeMisses, lookupRipeCached, pruneRipeSqlite } from "./traffic-flow-ripe.js"
|
||||
import { invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js"
|
||||
import { isIsoCountry } from "./traffic-flow-brands.js"
|
||||
import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js"
|
||||
import { pickInternetPeer } from "./traffic-flow-ip.js"
|
||||
@@ -94,10 +95,27 @@ let dropped = 0
|
||||
let rowsStored = 0
|
||||
let lastFlushUsedTransaction = false
|
||||
let lastPruneAt = 0
|
||||
let lastPassiveCheckpointAt = Date.now()
|
||||
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
|
||||
const PASSIVE_CHECKPOINT_MS = 60_000
|
||||
|
||||
function bumpDataEpoch(): void {
|
||||
dataEpoch += 1
|
||||
}
|
||||
|
||||
export function flowDataEpoch(): number {
|
||||
return dataEpoch
|
||||
}
|
||||
|
||||
function nowIso(): string {
|
||||
return new Date().toISOString()
|
||||
@@ -234,6 +252,7 @@ export function getEngineStats(): EngineStats {
|
||||
}
|
||||
|
||||
export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): void {
|
||||
if (flows.length) bumpDataEpoch()
|
||||
const bucketAt = minuteBucketIso()
|
||||
const ripeMisses: string[] = []
|
||||
for (const raw of flows) {
|
||||
@@ -422,7 +441,16 @@ export function applyRingSnapshot(rows: Array<{ key: string; inBps: number[]; ou
|
||||
}
|
||||
}
|
||||
|
||||
function persistListenerStats(handle: SqliteHandle): void {
|
||||
function persistListenerStats(handle: SqliteHandle): boolean {
|
||||
if (
|
||||
lastPersistedStats
|
||||
&& lastPersistedStats.packetsReceived === packetsReceived
|
||||
&& lastPersistedStats.lastDatagramAt === lastDatagramAt
|
||||
&& lastPersistedStats.lastExporterIp === lastExporterIp
|
||||
&& lastPersistedStats.lastError === lastError
|
||||
) {
|
||||
return false
|
||||
}
|
||||
handle.prepare(`
|
||||
UPDATE traffic_flow_settings
|
||||
SET packets_received = @packetsReceived,
|
||||
@@ -438,6 +466,25 @@ function persistListenerStats(handle: SqliteHandle): void {
|
||||
lastError,
|
||||
updatedAt: nowIso(),
|
||||
})
|
||||
lastPersistedStats = {
|
||||
packetsReceived,
|
||||
lastDatagramAt,
|
||||
lastExporterIp,
|
||||
lastError,
|
||||
}
|
||||
invalidateTrafficFlowSettingsCache()
|
||||
return true
|
||||
}
|
||||
|
||||
function maybePassiveCheckpoint(handle: SqliteHandle): void {
|
||||
const now = Date.now()
|
||||
if (now - lastPassiveCheckpointAt < PASSIVE_CHECKPOINT_MS) return
|
||||
lastPassiveCheckpointAt = now
|
||||
try {
|
||||
handle.pragma("wal_checkpoint(PASSIVE)")
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
|
||||
function upsertMinuteAndDaily(handle: SqliteHandle): void {
|
||||
@@ -586,6 +633,7 @@ function pruneStored(handle: SqliteHandle): void {
|
||||
}
|
||||
}
|
||||
}
|
||||
pruneRipeSqlite(now)
|
||||
}
|
||||
|
||||
function topNPending(rows: PendingFlowRow[]): PendingFlowRow[] {
|
||||
@@ -616,6 +664,7 @@ export function flushPending(): void {
|
||||
persistListenerStats(handle)
|
||||
if (pending.size === 0 && minuteRollup.size === 0 && minuteDims.size === 0) {
|
||||
pruneStored(handle)
|
||||
maybePassiveCheckpoint(handle)
|
||||
lastFlushUsedTransaction = false
|
||||
return
|
||||
}
|
||||
@@ -665,6 +714,7 @@ export function flushPending(): void {
|
||||
tx(rows)
|
||||
lastFlushUsedTransaction = true
|
||||
rowsStored += rows.length
|
||||
bumpDataEpoch()
|
||||
} catch {
|
||||
for (const r of rows) {
|
||||
try {
|
||||
@@ -697,11 +747,7 @@ export function flushPending(): void {
|
||||
/* rollup best-effort */
|
||||
}
|
||||
pruneStored(handle)
|
||||
try {
|
||||
handle.pragma("wal_checkpoint(TRUNCATE)")
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
maybePassiveCheckpoint(handle)
|
||||
}
|
||||
|
||||
export function lastFlushUsedTransactionForTests(): boolean {
|
||||
@@ -736,6 +782,9 @@ export function resetEngineForTests(): void {
|
||||
rowsStored = 0
|
||||
lastFlushUsedTransaction = false
|
||||
lastPruneAt = 0
|
||||
lastPassiveCheckpointAt = Date.now()
|
||||
lastPersistedStats = null
|
||||
bumpDataEpoch()
|
||||
pendingCap = MAX_PENDING
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
import assert from "node:assert/strict"
|
||||
import { SQLITE_BUSY_TIMEOUT_MS, sqliteDatabase } from "../db/index.js"
|
||||
import {
|
||||
SQLITE_BUSY_TIMEOUT_MS,
|
||||
SQLITE_CACHE_SIZE_KIB,
|
||||
SQLITE_WAL_AUTOCHECKPOINT_PAGES,
|
||||
sqliteDatabase,
|
||||
} from "../db/index.js"
|
||||
import {
|
||||
MAX_FLOW_LIVE_SUBSCRIBERS,
|
||||
resetFlowLiveSlotsForTests,
|
||||
@@ -11,6 +16,16 @@ const busy = sqliteDatabase.pragma("busy_timeout") as Array<{ busy_timeout: numb
|
||||
const busyValue = Array.isArray(busy) ? Number(Object.values(busy[0] ?? {})[0]) : Number(busy)
|
||||
assert.equal(busyValue, SQLITE_BUSY_TIMEOUT_MS)
|
||||
|
||||
function pragmaNum(name: string): number {
|
||||
const rows = sqliteDatabase.pragma(name) as Array<Record<string, number>>
|
||||
const row = Array.isArray(rows) ? rows[0] : rows
|
||||
return Number(Object.values(row ?? {})[0])
|
||||
}
|
||||
|
||||
assert.equal(pragmaNum("wal_autocheckpoint"), SQLITE_WAL_AUTOCHECKPOINT_PAGES)
|
||||
assert.equal(pragmaNum("cache_size"), -SQLITE_CACHE_SIZE_KIB)
|
||||
assert.equal(pragmaNum("temp_store"), 2)
|
||||
|
||||
resetFlowLiveSlotsForTests()
|
||||
for (let i = 0; i < MAX_FLOW_LIVE_SUBSCRIBERS; i++) {
|
||||
assert.equal(tryAcquireFlowLiveSlot(), true)
|
||||
|
||||
@@ -33,6 +33,7 @@ import {
|
||||
} from "./traffic-flow-settings.js"
|
||||
import { applicationName } from "./traffic-flow-apps.js"
|
||||
import { resolveIfaceName } from "./traffic-flow-ifaces.js"
|
||||
import { getServerCatalog } from "./traffic-flow-topology.js"
|
||||
|
||||
export type { PendingFlowRow }
|
||||
|
||||
@@ -332,8 +333,8 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto {
|
||||
const settings = getTrafficFlowSettingsRow()
|
||||
const runtime = getFlowRuntimeCounters()
|
||||
const rows = listFlowRowsForWindow(minutes)
|
||||
const serverRows = db.select().from(servers).all()
|
||||
const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host]))
|
||||
const catalog = getServerCatalog()
|
||||
const nameById = new Map([...catalog.byId].map(([id, s]) => [id, s.name]))
|
||||
const agg = new Map<string, FlowTalkerDto & { rawBytes: number }>()
|
||||
const protoBytes = new Map<number, number>()
|
||||
const srcs = new Set<string>()
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { eq } from "drizzle-orm"
|
||||
import type { FlowMapHop, FlowMapHopsDto, FlowMapService, FlowMapServiceEdge, FlowMapServicePath } from "@mmapp/contracts/traffic-flow"
|
||||
import { db } from "../db/index.js"
|
||||
import { servers, userInterfaceBindings } from "../db/schema.js"
|
||||
import { userInterfaceBindings } from "../db/schema.js"
|
||||
import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
|
||||
import {
|
||||
isNamedInternetService,
|
||||
@@ -15,7 +15,8 @@ import { classifyFlowPlane, shouldKeepPlane } from "./traffic-flow-planes.js"
|
||||
import { pickInternetPeer } from "./traffic-flow-ip.js"
|
||||
import { lookupRipeCached, type FlowIpMeta } from "./traffic-flow-ripe.js"
|
||||
import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js"
|
||||
import { loadFlowTopology, resolveClient, resolveEn } from "./traffic-flow-topology.js"
|
||||
import { loadFlowTopology, resolveClient, resolveEn, getServerCatalog } from "./traffic-flow-topology.js"
|
||||
import { flowDataEpoch } from "./traffic-flow-engine.js"
|
||||
|
||||
export const DEFAULT_MAP_SERVICE_MIN_SHARE_PCT = 5
|
||||
export const MAP_SERVICE_NODE_CAP = 20
|
||||
@@ -100,6 +101,7 @@ export function clampMapServiceMinSharePct(n: unknown): number {
|
||||
|
||||
function hopsQueryKey(q: FlowMapHopsQuery, minSharePct: number): string {
|
||||
return JSON.stringify({
|
||||
epoch: flowDataEpoch(),
|
||||
minutes: q.minutes,
|
||||
serverId: q.serverId ?? null,
|
||||
userId: q.userId ?? null,
|
||||
@@ -194,8 +196,8 @@ function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number): Flo
|
||||
const windowSec = Math.max(60, q.minutes * 60)
|
||||
const raw = listFlowRowsForWindow(q.minutes)
|
||||
const allow = q.userId ? userIfaceAllow(q.userId) : null
|
||||
const serverRows = db.select().from(servers).all()
|
||||
const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host]))
|
||||
const catalog = getServerCatalog()
|
||||
const nameById = new Map([...catalog.byId].map(([id, s]) => [id, s.name]))
|
||||
const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : ""
|
||||
const wantDedup = q.dedup !== false && !ifaceFilter
|
||||
const excludeMesh = q.excludeMesh !== false
|
||||
|
||||
@@ -21,6 +21,7 @@ import {
|
||||
} from "./traffic-flow-settings.js"
|
||||
import { refreshFlowExporterMap, startTrafficFlowListener } from "./traffic-flow-ingest.js"
|
||||
import { listTrafficFlowHostFiles } from "./traffic-flow-host-files.js"
|
||||
import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js"
|
||||
|
||||
const IFACE_NAME = "wg-flow"
|
||||
const JH_LISTEN_PORT = 13232
|
||||
@@ -288,6 +289,7 @@ export async function applyFlowOverlay(
|
||||
mgmtTunnelIp: address,
|
||||
updatedAt: new Date().toISOString(),
|
||||
}).where(eq(servers.id, server.id)).run()
|
||||
invalidateFlowCatalogCache()
|
||||
|
||||
upsertHostPeer({
|
||||
serverId: server.id,
|
||||
|
||||
@@ -43,8 +43,8 @@ try {
|
||||
`).run(serverId)
|
||||
sqliteDatabase.prepare(`
|
||||
INSERT INTO flow_ip_meta (prefix, asn, country, holder, ok, fetched_at)
|
||||
VALUES ('8.8.8.0/24', 15169, 'US', 'Google', 1, '2026-01-01T00:00:00.000Z')
|
||||
`).run()
|
||||
VALUES ('8.8.8.0/24', 15169, 'US', 'Google', 1, ?)
|
||||
`).run(new Date().toISOString())
|
||||
sqliteDatabase.prepare(`
|
||||
UPDATE traffic_flow_settings SET packets_received = 42, last_exporter_ip = '10.255.254.3' WHERE id = 1
|
||||
`).run()
|
||||
|
||||
@@ -239,6 +239,20 @@ function persistAsn(asn: number, holder: string): void {
|
||||
}
|
||||
}
|
||||
|
||||
/** Удаляет просроченный RIPE-кэш с диска (hit 24h / negative 6h). */
|
||||
export function pruneRipeSqlite(nowMs = Date.now()): void {
|
||||
if (!persistEnabled) return
|
||||
try {
|
||||
const hitCutoff = new Date(nowMs - HIT_TTL_MS).toISOString()
|
||||
const negCutoff = new Date(nowMs - NEG_TTL_MS).toISOString()
|
||||
sqliteDatabase.prepare(`DELETE FROM flow_ip_meta WHERE ok != 0 AND fetched_at < ?`).run(hitCutoff)
|
||||
sqliteDatabase.prepare(`DELETE FROM flow_ip_meta WHERE ok = 0 AND fetched_at < ?`).run(negCutoff)
|
||||
sqliteDatabase.prepare(`DELETE FROM flow_asn_meta WHERE fetched_at < ?`).run(hitCutoff)
|
||||
} catch {
|
||||
/* table may not exist in isolated tests */
|
||||
}
|
||||
}
|
||||
|
||||
function negative(prefix: string): FlowIpMeta {
|
||||
return {
|
||||
prefix,
|
||||
|
||||
@@ -4,10 +4,42 @@ import { trafficFlowSettings } from "../db/schema.js"
|
||||
import type { FlowHostPeer, TrafficFlowSettingsDto, TrafficFlowSettingsPatch } from "@mmapp/contracts/traffic-flow"
|
||||
import { generateWireGuardKeyPair } from "./wg-keys.js"
|
||||
|
||||
let settingsRowCache: ReturnType<typeof readSettingsRow> | null = null
|
||||
|
||||
function nowIso() {
|
||||
return new Date().toISOString()
|
||||
}
|
||||
|
||||
export function invalidateTrafficFlowSettingsCache(): void {
|
||||
settingsRowCache = null
|
||||
}
|
||||
|
||||
function readSettingsRow() {
|
||||
return db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0]
|
||||
}
|
||||
|
||||
export function getTrafficFlowSettingsRow() {
|
||||
if (settingsRowCache) return settingsRowCache
|
||||
const row = readSettingsRow()
|
||||
if (row) {
|
||||
settingsRowCache = row
|
||||
return row
|
||||
}
|
||||
const now = nowIso()
|
||||
db.insert(trafficFlowSettings).values({
|
||||
id: 1,
|
||||
enabled: false,
|
||||
collectorIp: "10.255.254.1",
|
||||
flowListenPort: 4739,
|
||||
wgListenPort: 51821,
|
||||
prefix: "10.255.254.0/24",
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
}).run()
|
||||
settingsRowCache = readSettingsRow()
|
||||
return settingsRowCache!
|
||||
}
|
||||
|
||||
function parsePeers(raw: string): FlowHostPeer[] {
|
||||
try {
|
||||
const parsed = JSON.parse(raw) as unknown
|
||||
@@ -20,23 +52,6 @@ function parsePeers(raw: string): FlowHostPeer[] {
|
||||
}
|
||||
}
|
||||
|
||||
export function getTrafficFlowSettingsRow() {
|
||||
const row = db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0]
|
||||
if (row) return row
|
||||
const now = nowIso()
|
||||
db.insert(trafficFlowSettings).values({
|
||||
id: 1,
|
||||
enabled: false,
|
||||
collectorIp: "10.255.254.1",
|
||||
flowListenPort: 4739,
|
||||
wgListenPort: 51821,
|
||||
prefix: "10.255.254.0/24",
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
}).run()
|
||||
return db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0]
|
||||
}
|
||||
|
||||
export function toTrafficFlowSettingsDto(
|
||||
listener: { bound: boolean; address: string | null },
|
||||
): TrafficFlowSettingsDto {
|
||||
@@ -81,6 +96,7 @@ export function updateTrafficFlowSettings(patch: TrafficFlowSettingsPatch) {
|
||||
: Math.min(100, Math.max(0, patch.mapServiceMinSharePct)),
|
||||
updatedAt: nowIso(),
|
||||
}).where(eq(trafficFlowSettings.id, 1)).run()
|
||||
invalidateTrafficFlowSettingsCache()
|
||||
return getTrafficFlowSettingsRow()
|
||||
}
|
||||
|
||||
@@ -95,6 +111,7 @@ export function ensureHostKeys(): { publicKey: string; created: boolean } {
|
||||
hostPrivateKey: keys.privateKey,
|
||||
updatedAt: nowIso(),
|
||||
}).where(eq(trafficFlowSettings.id, 1)).run()
|
||||
invalidateTrafficFlowSettingsCache()
|
||||
return { publicKey: keys.publicKey, created: true }
|
||||
}
|
||||
|
||||
@@ -106,6 +123,7 @@ export function upsertHostPeer(peer: FlowHostPeer) {
|
||||
peersJson: JSON.stringify(peers),
|
||||
updatedAt: nowIso(),
|
||||
}).where(eq(trafficFlowSettings.id, 1)).run()
|
||||
invalidateTrafficFlowSettingsCache()
|
||||
}
|
||||
|
||||
export function recordFlowPacket(exporterIp: string) {
|
||||
@@ -116,6 +134,7 @@ export function recordFlowPacket(exporterIp: string) {
|
||||
packetsReceived: row.packetsReceived + 1,
|
||||
updatedAt: nowIso(),
|
||||
}).where(eq(trafficFlowSettings.id, 1)).run()
|
||||
invalidateTrafficFlowSettingsCache()
|
||||
}
|
||||
|
||||
export function recordFlowListenerError(message: string) {
|
||||
@@ -123,6 +142,7 @@ export function recordFlowListenerError(message: string) {
|
||||
lastError: message,
|
||||
updatedAt: nowIso(),
|
||||
}).where(eq(trafficFlowSettings.id, 1)).run()
|
||||
invalidateTrafficFlowSettingsCache()
|
||||
}
|
||||
|
||||
export function enableTrafficFlowIngest() {
|
||||
@@ -130,6 +150,7 @@ export function enableTrafficFlowIngest() {
|
||||
enabled: true,
|
||||
updatedAt: nowIso(),
|
||||
}).where(eq(trafficFlowSettings.id, 1)).run()
|
||||
invalidateTrafficFlowSettingsCache()
|
||||
}
|
||||
|
||||
export function listHostPeers(): FlowHostPeer[] {
|
||||
@@ -144,4 +165,5 @@ export function resetFlowIngestCounters(): void {
|
||||
lastError: "",
|
||||
updatedAt: nowIso(),
|
||||
}).where(eq(trafficFlowSettings.id, 1)).run()
|
||||
invalidateTrafficFlowSettingsCache()
|
||||
}
|
||||
|
||||
@@ -27,7 +27,44 @@ export interface FlowTopology {
|
||||
plane: PlaneTopology
|
||||
}
|
||||
|
||||
export interface ServerCatalogEntry {
|
||||
id: number
|
||||
name: string
|
||||
country: string
|
||||
host: string
|
||||
type: string
|
||||
site: string
|
||||
}
|
||||
|
||||
const CATALOG_TTL_MS = 5_000
|
||||
|
||||
let seeded: FlowTopology | null = null
|
||||
let topologyCache: { at: number; topo: FlowTopology } | null = null
|
||||
let serverCatalogCache: { at: number; list: ServerCatalogEntry[]; byId: Map<number, ServerCatalogEntry> } | null = null
|
||||
|
||||
export function invalidateFlowCatalogCache(): void {
|
||||
topologyCache = null
|
||||
serverCatalogCache = null
|
||||
}
|
||||
|
||||
export function getServerCatalog(): { list: ServerCatalogEntry[]; byId: Map<number, ServerCatalogEntry> } {
|
||||
const now = Date.now()
|
||||
if (serverCatalogCache && now - serverCatalogCache.at < CATALOG_TTL_MS) {
|
||||
return serverCatalogCache
|
||||
}
|
||||
const rows = db.select().from(servers).all()
|
||||
const list: ServerCatalogEntry[] = rows.map((s) => ({
|
||||
id: s.id,
|
||||
name: s.name || s.host,
|
||||
country: (s.country || "").toUpperCase() || "UN",
|
||||
host: s.host,
|
||||
type: s.type,
|
||||
site: s.site || "—",
|
||||
}))
|
||||
const byId = new Map(list.map((s) => [s.id, s]))
|
||||
serverCatalogCache = { at: now, list, byId }
|
||||
return serverCatalogCache
|
||||
}
|
||||
|
||||
function parseWanUplinks(raw: string): Array<{ iface?: string; ip?: string }> {
|
||||
try {
|
||||
@@ -44,6 +81,8 @@ function ifaceKey(serverId: number, name: string): string {
|
||||
|
||||
export function loadFlowTopology(): FlowTopology {
|
||||
if (seeded) return seeded
|
||||
const now = Date.now()
|
||||
if (topologyCache && now - topologyCache.at < CATALOG_TTL_MS) return topologyCache.topo
|
||||
const serverRows = db.select().from(servers).all()
|
||||
const users = db.select().from(appUsers).all()
|
||||
const binds = db.select().from(userInterfaceBindings).all()
|
||||
@@ -82,7 +121,7 @@ export function loadFlowTopology(): FlowTopology {
|
||||
for (const h of hosts) jhHosts.add(h)
|
||||
}
|
||||
}
|
||||
return {
|
||||
const topo: FlowTopology = {
|
||||
clientIfaces,
|
||||
clientByIface,
|
||||
enNodes,
|
||||
@@ -95,10 +134,13 @@ export function loadFlowTopology(): FlowTopology {
|
||||
jhHosts,
|
||||
},
|
||||
}
|
||||
topologyCache = { at: Date.now(), topo }
|
||||
return topo
|
||||
}
|
||||
|
||||
export function seedFlowTopologyForTests(topo: FlowTopology | null): void {
|
||||
seeded = topo
|
||||
invalidateFlowCatalogCache()
|
||||
}
|
||||
|
||||
export function resolveClient(
|
||||
|
||||
Reference in New Issue
Block a user