306 lines
11 KiB
TypeScript
306 lines
11 KiB
TypeScript
import { and, asc, desc, eq, gte } from "drizzle-orm"
|
|
import { db } from "../db/index.js"
|
|
import { servers, trafficSamples, trafficSettings } from "../db/schema.js"
|
|
import { decodeTrafficSampleFlags, encodeTrafficFlags, trafficCounterContinuity } from "../db/traffic-flags.js"
|
|
import type { TrafficRunSnapshot } from "../types/scheduler-run-snapshot.js"
|
|
import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
|
|
import { MikrotikClient } from "./mikrotik.js"
|
|
import { bpsToMbps, shouldIncludeIface } from "./traffic-rate.js"
|
|
import { observedTrafficRate, parseTrafficCounter, parseTrafficFlag, nativeTrafficCounterIdentity } from "./traffic-observation.js"
|
|
import { rememberServerIfaces } from "./traffic-flow-ifindex.js"
|
|
import { recordTrafficCollectionAttempt } from "./traffic-collection-state.js"
|
|
import { parseTrafficPeers } from "./traffic-peer-observation.js"
|
|
|
|
interface RosIfaceTraffic {
|
|
".id"?: string
|
|
name?: string
|
|
ifindex?: string
|
|
running?: string
|
|
disabled?: string
|
|
"rx-byte"?: string
|
|
"tx-byte"?: string
|
|
"rx-bits-per-second"?: string
|
|
"tx-bits-per-second"?: string
|
|
}
|
|
|
|
interface RosWgPeerTraffic {
|
|
interface?: string
|
|
name?: string
|
|
comment?: string
|
|
"public-key"?: string
|
|
rx?: string
|
|
tx?: string
|
|
disabled?: string
|
|
}
|
|
|
|
function waveKey(interfaceName: string, peerPublicKey = ""): string {
|
|
return `${interfaceName}\0${peerPublicKey}`
|
|
}
|
|
|
|
function sampleRate(
|
|
prevWave: Map<string, { rxBytes: number; txBytes: number; sampledAt: string; nativeCounterIdentity: string|null }>,
|
|
key: string,
|
|
rxBytes: number,
|
|
txBytes: number,
|
|
nowMs: number,
|
|
maxIntervalMs: number,
|
|
identity: string|null,
|
|
): { rxBps: number; txBps: number; measured: boolean } {
|
|
return observedTrafficRate(identity !== null && prevWave.get(key)?.nativeCounterIdentity === identity ? prevWave.get(key) : undefined, rxBytes, txBytes, nowMs, maxIntervalMs)
|
|
}
|
|
|
|
export interface TrafficCollectorState {
|
|
running: boolean
|
|
lastRunAt: string | null
|
|
lastError: string | null
|
|
}
|
|
|
|
let collecting = false
|
|
|
|
function toNum(raw: unknown): number {
|
|
return parseTrafficCounter(raw)
|
|
}
|
|
|
|
async function getSettingsRow() {
|
|
const row = (await db.select().from(trafficSettings).where(eq(trafficSettings.id, 1)).limit(1))[0]
|
|
if (row) return row
|
|
const now = new Date().toISOString()
|
|
await db.insert(trafficSettings).values({
|
|
id: 1,
|
|
enabled: true,
|
|
intervalSec: 30,
|
|
retentionDays: 14,
|
|
createdAt: now,
|
|
updatedAt: now,
|
|
})
|
|
return (await db.select().from(trafficSettings).where(eq(trafficSettings.id, 1)).limit(1))[0]
|
|
}
|
|
|
|
async function readPreviousWave(serverId: number): Promise<Map<string, { rxBytes: number; txBytes: number; sampledAt: string; nativeCounterIdentity: string|null }>> {
|
|
const last = (await db
|
|
.select({ sampledAt: trafficSamples.sampledAt })
|
|
.from(trafficSamples)
|
|
.where(eq(trafficSamples.serverId, serverId))
|
|
.orderBy(desc(trafficSamples.sampledAt))
|
|
.limit(1))[0]
|
|
if (!last) return new Map()
|
|
const rows = await db
|
|
.select({
|
|
interfaceName: trafficSamples.interfaceName,
|
|
peerPublicKey: trafficSamples.peerPublicKey,
|
|
rxBytes: trafficSamples.rxBytes,
|
|
txBytes: trafficSamples.txBytes,
|
|
flags: trafficSamples.flags,
|
|
nativeCounterIdentity: trafficSamples.nativeCounterIdentity,
|
|
sampledAt: trafficSamples.sampledAt,
|
|
})
|
|
.from(trafficSamples)
|
|
.where(and(
|
|
eq(trafficSamples.serverId, serverId),
|
|
eq(trafficSamples.sampledAt, last.sampledAt),
|
|
))
|
|
return new Map(rows.filter(r => decodeTrafficSampleFlags(r.flags).countersKnown)
|
|
.map((r) => [`${r.interfaceName}\0${r.peerPublicKey ?? ""}`, r]))
|
|
}
|
|
|
|
export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
|
|
const sampledAt = new Date().toISOString()
|
|
if (collecting) {
|
|
return {
|
|
v: SCHEDULER_RUN_SNAPSHOT_VERSION,
|
|
job: "traffic",
|
|
sampledAt,
|
|
skipped: true,
|
|
servers: [],
|
|
}
|
|
}
|
|
collecting = true
|
|
const startedAt = Date.now()
|
|
const now = sampledAt
|
|
const snapshot: TrafficRunSnapshot = {
|
|
v: SCHEDULER_RUN_SNAPSHOT_VERSION,
|
|
job: "traffic",
|
|
sampledAt: now,
|
|
servers: [],
|
|
}
|
|
|
|
try {
|
|
const settings = await getSettingsRow()
|
|
const maxIntervalMs = Math.max(5, settings?.intervalSec ?? 30) * 3000
|
|
const enabledServers = await db.select().from(servers).where(eq(servers.enabled, true))
|
|
|
|
for (const srv of enabledServers) {
|
|
const name = srv.name || srv.host
|
|
try {
|
|
await recordTrafficCollectionAttempt(srv, "pending", now)
|
|
const client = MikrotikClient.fromServer(srv)
|
|
const ifaces = await client.get<RosIfaceTraffic[]>("/interface")
|
|
rememberServerIfaces(srv.id, ifaces)
|
|
const prevWave = await readPreviousWave(srv.id)
|
|
const nowMs = Date.parse(now)
|
|
if (!Array.isArray(ifaces) || ifaces.some(i => typeof i.name !== "string" || !i.name.trim()) || new Set(ifaces.map(i => i.name)).size !== ifaces.length)
|
|
throw new Error("RouterOS вернул некорректный каталог интерфейсов")
|
|
let sumRxMbps = 0
|
|
let sumTxMbps = 0
|
|
let includedInterfaces = 0
|
|
let totalsMeasured = true
|
|
const rows = ifaces.map((i) => {
|
|
const interfaceName = i.name!
|
|
const nativeCounterIdentity = nativeTrafficCounterIdentity(srv.updatedAt,i[".id"])
|
|
const continuityKnown = nativeCounterIdentity !== null && prevWave.get(waveKey(interfaceName))?.nativeCounterIdentity === nativeCounterIdentity
|
|
const rxBytes = toNum(i["rx-byte"])
|
|
const txBytes = toNum(i["tx-byte"])
|
|
const running = parseTrafficFlag(i.running)
|
|
const disabled = parseTrafficFlag(i.disabled)
|
|
const { rxBps, txBps, measured } = sampleRate(prevWave, waveKey(interfaceName), rxBytes, txBytes, nowMs, maxIntervalMs, nativeCounterIdentity)
|
|
if (shouldIncludeIface(interfaceName, running, disabled)) {
|
|
includedInterfaces++
|
|
totalsMeasured &&= measured
|
|
sumRxMbps += bpsToMbps(rxBps)
|
|
sumTxMbps += bpsToMbps(txBps)
|
|
}
|
|
return {
|
|
serverId: srv.id,
|
|
interfaceName,
|
|
peerPublicKey: "",
|
|
sampledAt: now,
|
|
rxBytes,
|
|
txBytes,
|
|
rxBps: Math.round(rxBps),
|
|
txBps: Math.round(txBps),
|
|
flags: encodeTrafficFlags(running, disabled, true, true, continuityKnown),
|
|
nativeCounterIdentity,
|
|
}
|
|
})
|
|
let peerWarning: string | undefined
|
|
try {
|
|
const peers = await client.get<RosWgPeerTraffic[]>("/interface/wireguard/peers")
|
|
for (const p of parseTrafficPeers(peers)) {
|
|
const { interfaceName, peerPublicKey, rxBytes, txBytes, disabled } = p
|
|
const nativeCounterIdentity = nativeTrafficCounterIdentity(srv.updatedAt,p.nativeId)
|
|
const continuityKnown = nativeCounterIdentity !== null && prevWave.get(waveKey(interfaceName,peerPublicKey))?.nativeCounterIdentity === nativeCounterIdentity
|
|
const running = !disabled
|
|
const { rxBps, txBps } = sampleRate(
|
|
prevWave,
|
|
waveKey(interfaceName, peerPublicKey),
|
|
rxBytes,
|
|
txBytes,
|
|
nowMs,
|
|
maxIntervalMs,
|
|
nativeCounterIdentity,
|
|
)
|
|
rows.push({
|
|
serverId: srv.id,
|
|
interfaceName,
|
|
peerPublicKey,
|
|
sampledAt: now,
|
|
rxBytes,
|
|
txBytes,
|
|
rxBps: Math.round(rxBps),
|
|
txBps: Math.round(txBps),
|
|
flags: encodeTrafficFlags(running, disabled, true, true, continuityKnown),
|
|
nativeCounterIdentity,
|
|
})
|
|
}
|
|
} catch (error) {
|
|
peerWarning = `Измерения WireGuard peers недоступны: ${error instanceof Error ? error.message : String(error)}`
|
|
}
|
|
if (rows.length > 0) {
|
|
await db.insert(trafficSamples).values(rows)
|
|
}
|
|
await recordTrafficCollectionAttempt(srv, "ok", now, peerWarning)
|
|
snapshot.servers.push({
|
|
serverId: srv.id,
|
|
name,
|
|
host: srv.host,
|
|
ok: true,
|
|
...(peerWarning ? { warning: peerWarning } : {}),
|
|
interfaces: ifaces.length,
|
|
...(includedInterfaces > 0 && totalsMeasured ? {
|
|
sumRxMbps,
|
|
sumTxMbps,
|
|
} : {}),
|
|
})
|
|
} catch (err) {
|
|
await recordTrafficCollectionAttempt(srv, "error", now, err instanceof Error ? err.message : String(err))
|
|
snapshot.servers.push({
|
|
serverId: srv.id,
|
|
name,
|
|
host: srv.host,
|
|
ok: false,
|
|
error: err instanceof Error ? err.message : String(err),
|
|
})
|
|
}
|
|
}
|
|
|
|
await db.update(trafficSettings).set({
|
|
lastCollectedAt: now,
|
|
lastDurationMs: Date.now() - startedAt,
|
|
lastError: snapshot.servers.filter(s => !s.ok || s.warning).map(s => `${s.name}: ${s.warning ?? s.error ?? "Ошибка опроса"}`).join("; "),
|
|
updatedAt: now,
|
|
}).where(eq(trafficSettings.id, 1))
|
|
} catch (e) {
|
|
const msg = e instanceof Error ? e.message : String(e)
|
|
snapshot.fatalError = msg
|
|
await db.update(trafficSettings).set({
|
|
lastCollectedAt: now,
|
|
lastDurationMs: Date.now() - startedAt,
|
|
lastError: msg,
|
|
updatedAt: now,
|
|
}).where(eq(trafficSettings.id, 1))
|
|
} finally {
|
|
collecting = false
|
|
}
|
|
return snapshot
|
|
}
|
|
|
|
/** Таймеры трафика ведёт планировщик. */
|
|
export function stopTrafficCollector() {}
|
|
|
|
export function restartTrafficCollector() {}
|
|
|
|
export async function getTrafficCollectorState(): Promise<TrafficCollectorState> {
|
|
const s = await getSettingsRow()
|
|
return {
|
|
running: Boolean(s.enabled),
|
|
lastRunAt: s.lastCollectedAt ?? null,
|
|
lastError: s.lastError || null,
|
|
}
|
|
}
|
|
|
|
/** Идёт активный прогон `collectTrafficOnce` (запись в PostgreSQL ещё не завершена). */
|
|
export function isTrafficCollecting(): boolean {
|
|
return collecting
|
|
}
|
|
|
|
export async function getTrafficSettings() {
|
|
return await getSettingsRow()
|
|
}
|
|
|
|
export async function updateTrafficSettings(patch: {
|
|
enabled?: boolean
|
|
intervalSec?: number
|
|
retentionDays?: number
|
|
}) {
|
|
const prev = await getSettingsRow()
|
|
const next = {
|
|
enabled: patch.enabled ?? prev.enabled,
|
|
intervalSec: patch.intervalSec ?? prev.intervalSec,
|
|
retentionDays: patch.retentionDays ?? prev.retentionDays,
|
|
updatedAt: new Date().toISOString(),
|
|
}
|
|
await db.update(trafficSettings).set(next).where(eq(trafficSettings.id, 1))
|
|
return await getSettingsRow()
|
|
}
|
|
|
|
export async function readServerSamplesInRange(serverId: number, sinceIso: string) {
|
|
const rows = await db.select()
|
|
.from(trafficSamples)
|
|
.where(and(
|
|
eq(trafficSamples.serverId, serverId),
|
|
gte(trafficSamples.sampledAt, sinceIso),
|
|
))
|
|
.orderBy(asc(trafficSamples.sampledAt))
|
|
return rows.map((r) => ({ ...r, ...decodeTrafficSampleFlags(r.flags), continuityKnown: trafficCounterContinuity(r.flags) }))
|
|
}
|