366 lines
15 KiB
TypeScript
366 lines
15 KiB
TypeScript
import { and, asc, eq } from "drizzle-orm"
|
|
import { db } from "../db/index.js"
|
|
import { servers, uptimeSpeedProbes, uptimeSpeedTestRuns } from "../db/schema.js"
|
|
import type { SpeedRunSnapshot, SpeedScheduledRunSnapshot } from "../types/scheduler-run-snapshot.js"
|
|
import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
|
|
import type { RosIpAddress } from "../types/server.js"
|
|
import { withBtestNodeLocks } from "./btest-node-locks.js"
|
|
import { MikrotikClient } from "./mikrotik.js"
|
|
import { parsePing } from "./uptime-collector.js"
|
|
import { assertBandwidthTestCompleted } from "./bandwidth-test-status.js"
|
|
import { assertSpeedProbeContext, persistSpeedRun, SpeedRunPersistenceUnknown, type SpeedProbeContext } from "./speed-run-persistence.js"
|
|
import { readBandwidthTestRates } from "./bandwidth-test-rates.js"
|
|
|
|
type SpeedWriteTransaction = Parameters<Parameters<typeof db.transaction>[0]>[0]
|
|
function speedProbeContextWhere(target: SpeedProbeContext) {
|
|
return and(eq(uptimeSpeedProbes.id, target.id), eq(uptimeSpeedProbes.srcServerId, target.srcServerId), eq(uptimeSpeedProbes.dstServerId, target.dstServerId),
|
|
eq(uptimeSpeedProbes.srcInterface, target.srcInterface), eq(uptimeSpeedProbes.dstInterface, target.dstInterface),
|
|
eq(uptimeSpeedProbes.protocol, target.protocol), eq(uptimeSpeedProbes.direction, target.direction), eq(uptimeSpeedProbes.durationSec, target.durationSec))
|
|
}
|
|
|
|
function firstUsableIpOnInterface(rows: RosIpAddress[], iface: string): string {
|
|
const name = iface.trim()
|
|
const candidates = rows.filter((a) => {
|
|
if (String(a.interface ?? "").trim() !== name) return false
|
|
if (String(a.disabled ?? "false").toLowerCase() === "true") return false
|
|
if (String(a.invalid ?? "false").toLowerCase() === "true") return false
|
|
const addr = String(a.address ?? "").trim()
|
|
return addr.includes("/")
|
|
})
|
|
candidates.sort((a, b) => {
|
|
const ad = String(a.dynamic ?? "false").toLowerCase() === "true" ? 1 : 0
|
|
const bd = String(b.dynamic ?? "false").toLowerCase() === "true" ? 1 : 0
|
|
return ad - bd
|
|
})
|
|
const raw = candidates[0]?.address
|
|
return raw ? String(raw).split("/")[0]?.trim() ?? "" : ""
|
|
}
|
|
|
|
export interface SpeedTestRunParams {
|
|
runId?: string
|
|
probeId?: string
|
|
srcServerId: number
|
|
dstServerId: number
|
|
srcInterface?: string
|
|
dstInterface?: string
|
|
protocol?: "tcp" | "udp"
|
|
direction?: "transmit" | "receive" | "both"
|
|
durationSec?: number
|
|
}
|
|
|
|
export interface SpeedTestRunResult {
|
|
createdAt: string
|
|
runId: string
|
|
srcInterface: string | null
|
|
dstInterface: string | null
|
|
address: string
|
|
srcAddress: string
|
|
dstAddress: string
|
|
srcInterfaceAddress: string | null
|
|
dstInterfaceAddress: string | null
|
|
protocol: "tcp" | "udp"
|
|
direction: "transmit" | "receive" | "both"
|
|
durationSec: number
|
|
txAvgMbps: number | null
|
|
rxAvgMbps: number | null
|
|
txSeries: number[]
|
|
rxSeries: number[]
|
|
raw: Array<Record<string, string>>
|
|
afterBtPing: {
|
|
rttMs: number | null
|
|
lossPct: number | null
|
|
error: string | null
|
|
}
|
|
}
|
|
|
|
function humanizeSpeedError(raw: string): string {
|
|
return /timed out|socket hang up|ECONNRESET/i.test(raw)
|
|
? "BTTest timeout/socket hang up: проверь доступность destination, /tool bandwidth-server, firewall и корректность интерфейсов"
|
|
: raw
|
|
}
|
|
|
|
/** Полный цикл bandwidth-test + after-ping + запись в БД. Вызывать внутри withBtestNodeLocks при необходимости — здесь lock уже снаружи. */
|
|
export async function executeBandwidthSpeedTestInner(params: SpeedTestRunParams): Promise<SpeedTestRunResult> {
|
|
const probeId = String(params.probeId ?? "").trim()
|
|
const srcId = params.srcServerId
|
|
const dstId = params.dstServerId
|
|
const src = (await db.select().from(servers).where(eq(servers.id, srcId)).limit(1))[0]
|
|
const dst = (await db.select().from(servers).where(eq(servers.id, dstId)).limit(1))[0]
|
|
if (!src || !dst) throw new Error("Server not found")
|
|
assertMonitoringServersEnabled(src, dst)
|
|
|
|
const protocol = params.protocol === "udp" ? "udp" : "tcp"
|
|
const direction = params.direction === "transmit" || params.direction === "receive" ? params.direction : "both"
|
|
const durationSec = Math.max(3, Number.parseInt(String(params.durationSec ?? ""), 10) || 10)
|
|
const srcInterface = String(params.srcInterface ?? "").trim()
|
|
const dstInterface = String(params.dstInterface ?? "").trim()
|
|
const target: SpeedProbeContext = { id: probeId, srcServerId: srcId, dstServerId: dstId, srcInterface, dstInterface, protocol, direction, durationSec }
|
|
if (probeId) {
|
|
const probe = (await db.select().from(uptimeSpeedProbes).where(eq(uptimeSpeedProbes.id, probeId)).limit(1))[0]
|
|
assertSpeedProbeContext(probe, target)
|
|
}
|
|
const srcClient = MikrotikClient.fromServer(src)
|
|
const dstAddressRows = dstInterface
|
|
? await MikrotikClient.fromServer(dst).getIpAddresses().catch(() => [])
|
|
: []
|
|
const srcAddressRows = srcInterface
|
|
? await srcClient.getIpAddresses().catch(() => [])
|
|
: []
|
|
const dstAddress = dstInterface ? firstUsableIpOnInterface(dstAddressRows, dstInterface) : dst.host
|
|
if (dstInterface && !dstAddress) {
|
|
throw new Error(`На интерфейсе назначения '${dstInterface}' нет активного IP (проверьте /ip/address — не disabled).`)
|
|
}
|
|
const initialAddress = dstAddress || dst.host
|
|
const srcAddress = srcInterface ? firstUsableIpOnInterface(srcAddressRows, srcInterface) : src.host
|
|
const srcInterfaceAddress = srcInterface ? (srcAddress || null) : null
|
|
const dstInterfaceAddress = dstInterface ? (dstAddress || null) : null
|
|
|
|
const rows = await srcClient.bandwidthTest({
|
|
address: initialAddress,
|
|
user: dst.username,
|
|
password: dst.password,
|
|
protocol,
|
|
direction,
|
|
durationSec,
|
|
})
|
|
assertBandwidthTestCompleted(rows)
|
|
const { txAvgMbps: txFinal, rxAvgMbps: rxFinal, txSeries, rxSeries } = readBandwidthTestRates(rows)
|
|
|
|
await new Promise<void>((resolve) => setTimeout(resolve, 3000))
|
|
|
|
const pingNow = new Date().toISOString()
|
|
let afterBtPing: SpeedTestRunResult["afterBtPing"] = { rttMs: null, lossPct: null, error: null }
|
|
try {
|
|
const pingRows = await srcClient.ping(initialAddress, 4, srcInterface || undefined, { interval: "1s" })
|
|
const pr = parsePing(pingRows as Parameters<typeof parsePing>[0])
|
|
afterBtPing = { rttMs: pr.avgRtt, lossPct: pr.loss, error: null }
|
|
} catch (pe) {
|
|
afterBtPing = {
|
|
rttMs: null,
|
|
lossPct: null,
|
|
error: pe instanceof Error ? pe.message : String(pe),
|
|
}
|
|
}
|
|
|
|
const runId = String(params.runId ?? "").trim() || `sr-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`
|
|
try {
|
|
await persistSpeedRun<SpeedWriteTransaction, SpeedProbeContext>(probeId ? target : null, {
|
|
transaction: work => db.transaction(work),
|
|
updateProbe: async (tx, target) => {
|
|
const rows = await tx.update(uptimeSpeedProbes).set({
|
|
lastRunAt: pingNow,
|
|
lastTxAvgMbps: txFinal,
|
|
lastRxAvgMbps: rxFinal,
|
|
lastStatus: "done",
|
|
lastError: "",
|
|
lastPingRttMs: afterBtPing.rttMs,
|
|
lastPingLossPct: afterBtPing.lossPct,
|
|
lastPingAt: pingNow,
|
|
lastPingError: afterBtPing.error ?? "",
|
|
updatedAt: pingNow,
|
|
}).where(speedProbeContextWhere(target)).returning({ id: uptimeSpeedProbes.id })
|
|
return rows.length === 1
|
|
},
|
|
insertRun: async (tx, associatedProbeId) => {
|
|
await tx.insert(uptimeSpeedTestRuns).values({
|
|
id: runId,
|
|
probeId: associatedProbeId,
|
|
srcServerId: srcId,
|
|
dstServerId: dstId,
|
|
srcInterface: srcInterface || "",
|
|
dstInterface: dstInterface || "",
|
|
srcAddress: srcAddress || src.host,
|
|
dstAddress: initialAddress,
|
|
srcInterfaceAddress,
|
|
dstInterfaceAddress,
|
|
protocol,
|
|
direction,
|
|
durationSec,
|
|
txAvgMbps: txFinal,
|
|
rxAvgMbps: rxFinal,
|
|
pingRttMs: afterBtPing.rttMs,
|
|
pingLossPct: afterBtPing.lossPct,
|
|
pingError: afterBtPing.error ?? null,
|
|
status: "done",
|
|
error: null,
|
|
createdAt: pingNow,
|
|
})
|
|
},
|
|
})
|
|
|
|
} catch (error) { throw new SpeedRunPersistenceUnknown(runId, error) }
|
|
|
|
return {
|
|
createdAt: pingNow,
|
|
runId,
|
|
srcInterface: srcInterface || null,
|
|
dstInterface: dstInterface || null,
|
|
address: initialAddress,
|
|
srcAddress: srcAddress || src.host,
|
|
dstAddress: initialAddress,
|
|
srcInterfaceAddress,
|
|
dstInterfaceAddress,
|
|
protocol,
|
|
direction,
|
|
durationSec,
|
|
txAvgMbps: txFinal,
|
|
rxAvgMbps: rxFinal,
|
|
txSeries,
|
|
rxSeries,
|
|
raw: rows,
|
|
afterBtPing,
|
|
}
|
|
}
|
|
|
|
export async function runBandwidthSpeedTest(params: SpeedTestRunParams): Promise<SpeedTestRunResult> {
|
|
const srcId = params.srcServerId
|
|
const dstId = params.dstServerId
|
|
if (srcId === dstId) throw new Error("Source and destination must be different")
|
|
return await withBtestNodeLocks([srcId, dstId], () => executeBandwidthSpeedTestInner(params))
|
|
}
|
|
|
|
export async function recordSpeedTestFailure(params: {
|
|
probeId?: string
|
|
runId?: string
|
|
srcServerId: number
|
|
dstServerId: number
|
|
srcInterface?: string
|
|
dstInterface?: string
|
|
protocol?: "tcp" | "udp"
|
|
direction?: "transmit" | "receive" | "both"
|
|
durationSec?: number
|
|
message: string
|
|
}) {
|
|
const probeId = String(params.probeId ?? "").trim()
|
|
const message = humanizeSpeedError(params.message)
|
|
await persistSpeedRun<SpeedWriteTransaction, SpeedProbeContext>(probeId ? {
|
|
id: probeId, srcServerId: params.srcServerId, dstServerId: params.dstServerId,
|
|
srcInterface: String(params.srcInterface ?? "").trim(), dstInterface: String(params.dstInterface ?? "").trim(),
|
|
protocol: params.protocol === "udp" ? "udp" : "tcp",
|
|
direction: params.direction === "transmit" || params.direction === "receive" ? params.direction : "both",
|
|
durationSec: Math.max(3, Number.parseInt(String(params.durationSec ?? ""), 10) || 10),
|
|
} : null, {
|
|
transaction: work => db.transaction(work),
|
|
updateProbe: async (tx, target) => {
|
|
const rows = await tx.update(uptimeSpeedProbes).set({
|
|
lastRunAt: new Date().toISOString(),
|
|
lastStatus: "error",
|
|
lastError: message,
|
|
updatedAt: new Date().toISOString(),
|
|
}).where(speedProbeContextWhere(target)).returning({ id: uptimeSpeedProbes.id })
|
|
return rows.length === 1
|
|
},
|
|
insertRun: async (tx, associatedProbeId) => {
|
|
await tx.insert(uptimeSpeedTestRuns).values({
|
|
id: String(params.runId ?? "").trim() || `sr-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
|
|
probeId: associatedProbeId,
|
|
srcServerId: params.srcServerId,
|
|
dstServerId: params.dstServerId,
|
|
srcInterface: String(params.srcInterface ?? "").trim(),
|
|
dstInterface: String(params.dstInterface ?? "").trim(),
|
|
srcAddress: null,
|
|
dstAddress: null,
|
|
srcInterfaceAddress: null,
|
|
dstInterfaceAddress: null,
|
|
protocol: params.protocol === "udp" ? "udp" : "tcp",
|
|
direction: params.direction === "transmit" || params.direction === "receive" ? params.direction : "both",
|
|
durationSec: Math.max(3, Number.parseInt(String(params.durationSec ?? ""), 10) || 10),
|
|
txAvgMbps: null,
|
|
rxAvgMbps: null,
|
|
pingRttMs: null,
|
|
pingLossPct: null,
|
|
pingError: null,
|
|
status: "error",
|
|
error: message,
|
|
createdAt: new Date().toISOString(),
|
|
})
|
|
},
|
|
})
|
|
}
|
|
|
|
async function serverDisplayName(id: number): Promise<string> {
|
|
const s = (await db.select().from(servers).where(eq(servers.id, id)).limit(1))[0]
|
|
return s ? (s.name || s.host) : `#${id}`
|
|
}
|
|
|
|
/** Фоновый прогон всех включённых speed-проб (по очереди). */
|
|
export async function runScheduledSpeedProbesOnce(): Promise<SpeedScheduledRunSnapshot> {
|
|
const sampledAt = new Date().toISOString()
|
|
const runs: SpeedRunSnapshot[] = []
|
|
const rows = await db.select().from(uptimeSpeedProbes).orderBy(asc(uptimeSpeedProbes.sortOrder))
|
|
const enabledIds = new Set((await listEnabledMonitoringServers()).map(server => server.id))
|
|
const enabled = rows.filter((r) => r.enabled !== false && r.srcServerId !== r.dstServerId
|
|
&& enabledIds.has(r.srcServerId) && enabledIds.has(r.dstServerId))
|
|
const protocol = (p: (typeof enabled)[number]) => (p.protocol === "udp" ? "udp" : "tcp")
|
|
const direction = (p: (typeof enabled)[number]) =>
|
|
p.direction === "transmit" || p.direction === "receive" ? p.direction : "both"
|
|
const durationSec = (p: (typeof enabled)[number]) => Math.max(3, p.durationSec || 10)
|
|
|
|
for (const p of enabled) {
|
|
try {
|
|
const r = await runBandwidthSpeedTest({
|
|
probeId: p.id,
|
|
srcServerId: p.srcServerId,
|
|
dstServerId: p.dstServerId,
|
|
srcInterface: p.srcInterface || undefined,
|
|
dstInterface: p.dstInterface || undefined,
|
|
protocol: protocol(p) as "tcp" | "udp",
|
|
direction: direction(p) as "transmit" | "receive" | "both",
|
|
durationSec: durationSec(p),
|
|
})
|
|
runs.push({
|
|
probeId: p.id,
|
|
srcServerId: p.srcServerId,
|
|
srcServerName: await serverDisplayName(p.srcServerId),
|
|
dstServerId: p.dstServerId,
|
|
dstServerName: await serverDisplayName(p.dstServerId),
|
|
srcInterface: r.srcInterface ?? "",
|
|
dstInterface: r.dstInterface ?? "",
|
|
protocol: r.protocol,
|
|
direction: r.direction,
|
|
durationSec: r.durationSec,
|
|
ok: true,
|
|
txAvgMbps: r.txAvgMbps,
|
|
rxAvgMbps: r.rxAvgMbps,
|
|
pingRttMs: r.afterBtPing.rttMs,
|
|
pingLossPct: r.afterBtPing.lossPct,
|
|
pingError: r.afterBtPing.error,
|
|
})
|
|
} catch (e) {
|
|
const msg = e instanceof Error ? e.message : String(e)
|
|
if (!(e instanceof SpeedRunPersistenceUnknown)) await recordSpeedTestFailure({
|
|
probeId: p.id,
|
|
srcServerId: p.srcServerId,
|
|
dstServerId: p.dstServerId,
|
|
srcInterface: p.srcInterface,
|
|
dstInterface: p.dstInterface,
|
|
protocol: protocol(p),
|
|
direction: direction(p),
|
|
durationSec: p.durationSec,
|
|
message: msg,
|
|
})
|
|
runs.push({
|
|
probeId: p.id,
|
|
srcServerId: p.srcServerId,
|
|
srcServerName: await serverDisplayName(p.srcServerId),
|
|
dstServerId: p.dstServerId,
|
|
dstServerName: await serverDisplayName(p.dstServerId),
|
|
srcInterface: p.srcInterface || "",
|
|
dstInterface: p.dstInterface || "",
|
|
protocol: protocol(p),
|
|
direction: direction(p),
|
|
durationSec: durationSec(p),
|
|
ok: false,
|
|
error: msg,
|
|
})
|
|
}
|
|
}
|
|
return {
|
|
v: SCHEDULER_RUN_SNAPSHOT_VERSION,
|
|
job: "uptime_speed",
|
|
sampledAt,
|
|
runs,
|
|
}
|
|
}
|
|
import { assertMonitoringServersEnabled, listEnabledMonitoringServers } from "./monitoring-server-catalog.js"
|