diff --git a/app/(main)/traffic/page.tsx b/app/(main)/traffic/page.tsx
index b4630ff4..d5c6cebb 100644
--- a/app/(main)/traffic/page.tsx
+++ b/app/(main)/traffic/page.tsx
@@ -2,6 +2,8 @@
import { useState, useMemo, useEffect, type ReactNode } from "react"
import { Alert, AlertTitle, AlertDescription } from "@/components/reui/alert"
+import { projectTrafficObservations, type TrafficObservationMeta } from "@/lib/traffic-observations"
+import { TrafficObservationStatus } from "@/components/traffic/traffic-observation-status"
import { TrafficDataTime } from "@/components/traffic/traffic-data-time"
import { PageHeader } from "@/components/page-header"
import { Frame, FramePanel } from "@/components/reui/frame"
@@ -84,6 +86,7 @@ function flowEmptyHint(stats: FlowStatsDto | null, collectorAlive?: boolean): st
// ─── data model ───────────────────────────────────────────────────────────────
interface BoundIfaceTraffic {
+ observations?: TrafficObservationMeta
id: string
bindingId: string
userId: string
@@ -98,75 +101,79 @@ interface BoundIfaceTraffic {
serverName: string
serverSite: string
serverCountry: string
- rxNow: number
- txNow: number
- rxPeak: number
- txPeak: number
- rxTotal: number
- txTotal: number
- rxSeries: number[]
- txSeries: number[]
+ rxNow: number | null
+ txNow: number | null
+ rxPeak: number | null
+ txPeak: number | null
+ rxTotal: number | null
+ txTotal: number | null
+ rxSeries: (number|null)[]
+ txSeries: (number|null)[]
status: "online" | "offline"
}
interface ServerTraffic {
+ observations?: TrafficObservationMeta
id: string
name: string
site: string
country: string
status: "online" | "offline" | "degraded"
- rxNow: number
- txNow: number
- rxPeak: number
- txPeak: number
- rxTotal: number
- txTotal: number
+ rxNow: number | null
+ txNow: number | null
+ rxPeak: number | null
+ txPeak: number | null
+ rxTotal: number | null
+ txTotal: number | null
sessions: number
- rxSeries: number[]
- txSeries: number[]
+ rxSeries: (number|null)[]
+ txSeries: (number|null)[]
boundIfaces: BoundIfaceTraffic[]
}
interface UserTraffic {
+ observations?: TrafficObservationMeta
id: string
login: string
displayName: string
role: string
active?: boolean
interfaces: BoundIfaceTraffic[]
- rxNow: number
- txNow: number
- rxPeak: number
- txPeak: number
- rxTotal: number
- txTotal: number
- rxSeries: number[]
- txSeries: number[]
+ rxNow: number | null
+ txNow: number | null
+ rxPeak: number | null
+ txPeak: number | null
+ rxTotal: number | null
+ txTotal: number | null
+ rxSeries: (number|null)[]
+ txSeries: (number|null)[]
}
interface LiveTrafficServer {
+ observations?: TrafficObservationMeta
id: string
name: string
site: string
country: string
status: "online" | "offline" | "degraded"
- rxNow: number
- txNow: number
- rxPeak: number
- txPeak: number
- rxTotal: number
- txTotal: number
+ rxNow: number | null
+ txNow: number | null
+ rxPeak: number | null
+ txPeak: number | null
+ rxTotal: number | null
+ txTotal: number | null
sessions: number
- rxSeries: number[]
- txSeries: number[]
+ rxSeries: (number|null)[]
+ txSeries: (number|null)[]
}
interface LiveTrafficInterface {
+ observations?: TrafficObservationMeta
name: string
running: boolean
disabled: boolean
- rxNow: number
- txNow: number
+ rxNow: number | null
+ txNow: number | null
}
function makeApiFetch(backendUrl: string) {
@@ -175,6 +182,13 @@ function makeApiFetch(backendUrl: string) {
}
}
+function compareRates(a: number|null, b: number|null, direction: "asc"|"desc") {
+ if (a === null && b === null) return 0
+ if (a === null) return direction === "desc" ? -1 : 1
+ if (b === null) return direction === "desc" ? 1 : -1
+ return a - b
+}
+
function hashSeed(s: string): number {
let h = 0
for (let i = 0; i < s.length; i++) h = (h * 31 + s.charCodeAt(i)) | 0
@@ -278,14 +292,14 @@ function buildUserTraffic(
clients: BoundIfaceTraffic[],
active = true,
): UserTraffic {
- const rxNow = clients.reduce((a, c) => a + c.rxNow, 0)
- const txNow = clients.reduce((a, c) => a + c.txNow, 0)
- const rxPeak = clients.reduce((a, c) => a + c.rxPeak, 0)
- const txPeak = clients.reduce((a, c) => a + c.txPeak, 0)
- const rxTotal = clients.reduce((a, c) => a + c.rxTotal, 0)
- const txTotal = clients.reduce((a, c) => a + c.txTotal, 0)
- const rxSeries = clients.reduce((a, c) => addSeries(a, c.rxSeries), Array(60).fill(0) as number[])
- const txSeries = clients.reduce((a, c) => addSeries(a, c.txSeries), Array(60).fill(0) as number[])
+ const rxNow = clients.reduce((a, c) => a + (c.rxNow ?? 0), 0)
+ const txNow = clients.reduce((a, c) => a + (c.txNow ?? 0), 0)
+ const rxPeak = clients.reduce((a, c) => a + (c.rxPeak ?? 0), 0)
+ const txPeak = clients.reduce((a, c) => a + (c.txPeak ?? 0), 0)
+ const rxTotal = clients.reduce((a, c) => a + (c.rxTotal ?? 0), 0)
+ const txTotal = clients.reduce((a, c) => a + (c.txTotal ?? 0), 0)
+ const rxSeries = clients.reduce((a, c) => addSeries(a, c.rxSeries.map(v => v ?? 0)), Array(60).fill(0) as number[])
+ const txSeries = clients.reduce((a, c) => addSeries(a, c.txSeries.map(v => v ?? 0)), Array(60).fill(0) as number[])
return { id, login, displayName, role, active, interfaces: clients, rxNow, txNow, rxPeak, txPeak, rxTotal, txTotal, rxSeries, txSeries }
}
@@ -538,21 +552,21 @@ function OfflinePlaceholder({ text = "Нет данных — объект не
}
function TotalsRow({ rxTotal, txTotal, rxSeries, txSeries }: {
- rxTotal: number; txTotal: number; rxSeries: number[]; txSeries: number[]
+ rxTotal: number|null; txTotal: number|null; rxSeries: (number|null)[]; txSeries: (number|null)[]
}) {
return (
-
Получено за период
+
Учтено RX по замерам
- {rxTotal.toFixed(1)} ГБ
+ {fmtGB(rxTotal)}
-
Отправлено за период
+
Учтено TX по замерам
- {txTotal.toFixed(1)} ГБ
+ {fmtGB(txTotal)}
@@ -583,7 +597,8 @@ function ServerDetail({
{sel.site}
- {sel.status === "offline" ?
:
}
+
+
{sel.serverSite}
- {sel.status === "offline" ?
:
}
+
} label="RX" value={fmtRate(sel.rxNow)} />
} label="TX" value={fmtRate(sel.txNow)} />
@@ -738,24 +753,16 @@ const toLiveServer = (s: LiveTrafficServer): ServerTraffic => {
site: s.site || "—",
country: s.country || "UN",
status: s.status,
- rxNow: s.rxNow,
- txNow: s.txNow,
- rxPeak: s.rxPeak,
- txPeak: s.txPeak,
- rxTotal: s.rxTotal,
- txTotal: s.txTotal,
sessions: s.sessions,
- rxSeries: s.rxSeries,
- txSeries: s.txSeries,
boundIfaces: [],
+ ...projectTrafficObservations(s),
}
}
const toLiveBound = (c: BoundIfaceTraffic): BoundIfaceTraffic => ({
...c,
interfaceType: (["ether", "gre", "wg", "other"].includes(c.interfaceType) ? c.interfaceType : "other") as InterfaceType,
- rxSeries: c.rxSeries,
- txSeries: c.txSeries,
+ ...projectTrafficObservations(c),
})
const toLiveUser = (u: UserTraffic & { interfaces?: BoundIfaceTraffic[] }): UserTraffic => {
@@ -764,14 +771,19 @@ const toLiveUser = (u: UserTraffic & { interfaces?: BoundIfaceTraffic[] }): User
...u,
role: ROLE_LABEL[u.role as keyof typeof ROLE_LABEL] ?? u.role,
interfaces: ifaces,
- rxSeries: u.rxSeries,
- txSeries: u.txSeries,
+ ...projectTrafficObservations(u),
}
}
export default function TrafficPage() {
const { mode, backendUrl } = useDataSource()
const isLive = mode === "live"
+ const [trafficClock, setTrafficClock] = useState(() => Date.now())
+ useEffect(() => {
+ if (!isLive) return
+ const timer = setInterval(() => setTrafficClock(Date.now()), 10_000)
+ return () => clearInterval(timer)
+ }, [isLive])
const apiFetch = useMemo(() => makeApiFetch(backendUrl), [backendUrl])
const [groupMode, setGroupMode] = useState
("servers")
@@ -945,9 +957,9 @@ export default function TrafficPage() {
void listServers(backendUrl).then(setCatalogServers).catch(() => setCatalogServers([]))
}, [isLive, backendUrl])
- const activeServerTraffic = useMemo(() => isLive ? (trafficReady ? liveServers : []) : serverTraffic, [isLive, trafficReady, liveServers])
- const activeUserTraffic = useMemo(() => isLive ? (trafficReady ? liveUsers : []) : userTraffic, [isLive, trafficReady, liveUsers])
- const activeBoundIfaces = useMemo(() => isLive ? (trafficReady ? liveBoundIfaces : []) : boundIfaces, [isLive, trafficReady, liveBoundIfaces])
+ const activeServerTraffic = useMemo(() => isLive ? (trafficReady ? liveServers.map(s => ({ ...s, ...projectTrafficObservations(s, trafficClock) })) : []) : serverTraffic, [isLive, trafficReady, liveServers, trafficClock])
+ const activeUserTraffic = useMemo(() => isLive ? (trafficReady ? liveUsers.map(u => ({ ...u, ...projectTrafficObservations(u, trafficClock), interfaces: u.interfaces.map(toLiveBound) })) : []) : userTraffic, [isLive, trafficReady, liveUsers, trafficClock])
+ const activeBoundIfaces = useMemo(() => isLive ? (trafficReady ? liveBoundIfaces.map(c => ({ ...c, ...projectTrafficObservations(c, trafficClock) })) : []) : boundIfaces, [isLive, trafficReady, liveBoundIfaces, trafficClock])
const visibleIfaces = useMemo(
() => hideDisabledIfaces ? serverIfaces.filter((i) => !i.disabled && i.running) : serverIfaces,
[serverIfaces, hideDisabledIfaces],
@@ -1022,8 +1034,8 @@ export default function TrafficPage() {
.filter(s => !q || s.name.toLowerCase().includes(q) || s.site.toLowerCase().includes(q))
.sort((a, b) => {
let v = 0
- if (sortField === "rx") v = a.rxNow - b.rxNow
- else if (sortField === "tx") v = a.txNow - b.txNow
+ if (sortField === "rx") v = compareRates(a.rxNow, b.rxNow, sortDir)
+ else if (sortField === "tx") v = compareRates(a.txNow, b.txNow, sortDir)
else if (sortField === "name") v = a.name.localeCompare(b.name)
else if (sortField === "sessions") v = a.sessions - b.sessions
return sortDir === "desc" ? -v : v
@@ -1035,8 +1047,8 @@ export default function TrafficPage() {
.filter(u => !q || u.login.toLowerCase().includes(q) || u.displayName.toLowerCase().includes(q))
.sort((a, b) => {
let v = 0
- if (sortField === "rx") v = a.rxNow - b.rxNow
- else if (sortField === "tx") v = a.txNow - b.txNow
+ if (sortField === "rx") v = compareRates(a.rxNow, b.rxNow, sortDir)
+ else if (sortField === "tx") v = compareRates(a.txNow, b.txNow, sortDir)
else if (sortField === "name") v = a.login.localeCompare(b.login)
else if (sortField === "sessions") v = a.interfaces.length - b.interfaces.length
return sortDir === "desc" ? -v : v
@@ -1052,8 +1064,8 @@ export default function TrafficPage() {
|| c.userLogin.toLowerCase().includes(q))
.sort((a, b) => {
let v = 0
- if (sortField === "rx") v = a.rxNow - b.rxNow
- else if (sortField === "tx") v = a.txNow - b.txNow
+ if (sortField === "rx") v = compareRates(a.rxNow, b.rxNow, sortDir)
+ else if (sortField === "tx") v = compareRates(a.txNow, b.txNow, sortDir)
else if (sortField === "name") v = a.interfaceName.localeCompare(b.interfaceName)
return sortDir === "desc" ? -v : v
})
@@ -1065,8 +1077,8 @@ export default function TrafficPage() {
.filter((c) => !q || c.name.toLowerCase().includes(q) || c.subtitle.toLowerCase().includes(q) || c.site.toLowerCase().includes(q))
.sort((a, b) => {
let v = 0
- if (sortField === "rx") v = a.rxNow - b.rxNow
- else if (sortField === "tx") v = a.txNow - b.txNow
+ if (sortField === "rx") v = compareRates(a.rxNow, b.rxNow, sortDir)
+ else if (sortField === "tx") v = compareRates(a.txNow, b.txNow, sortDir)
else if (sortField === "name") v = a.name.localeCompare(b.name)
else if (sortField === "sessions") v = a.sessions - b.sessions
return sortDir === "desc" ? -v : v
@@ -1081,7 +1093,7 @@ export default function TrafficPage() {
const displayedFlow = currentFlows && !liveError ? (flowLiveSample ?? currentAnalytics?.data ?? null) : null
const selServer = useMemo(() => activeServerTraffic.find(s => s.id === selectedId) ?? activeServerTraffic[0], [selectedId, activeServerTraffic])
- const detailServer = isLive ? currentDetail?.server : selServer
+ const detailServer = isLive ? (currentDetail?.server ? { ...currentDetail.server, ...projectTrafficObservations(currentDetail.server, trafficClock) } : null) : selServer
const selUser = useMemo(() => activeUserTraffic.find(u => u.id === selectedId) ?? activeUserTraffic[0], [selectedId, activeUserTraffic])
const selIface = useMemo(() => activeBoundIfaces.find(c => c.id === selectedId) ?? activeBoundIfaces[0], [selectedId, activeBoundIfaces])
@@ -1090,10 +1102,10 @@ export default function TrafficPage() {
: effectiveMode === "ifaces"
? activeBoundIfaces
: activeServerTraffic
- const totalRx = kpiSource.reduce((a, s) => a + s.rxNow, 0)
- const totalTx = kpiSource.reduce((a, s) => a + s.txNow, 0)
- const peakRx = kpiSource.reduce((a, s) => Math.max(a, s.rxPeak), 0)
- const peakTx = kpiSource.reduce((a, s) => Math.max(a, s.txPeak), 0)
+ const totalRx = kpiSource.every(s => s.rxNow !== null) ? kpiSource.reduce((a, s) => a + (s.rxNow ?? 0), 0) : null
+ const totalTx = kpiSource.every(s => s.txNow !== null) ? kpiSource.reduce((a, s) => a + (s.txNow ?? 0), 0) : null
+ const peakRx = kpiSource.every(s => s.rxPeak !== null) ? kpiSource.reduce((a, s) => Math.max(a, s.rxPeak ?? 0), 0) : null
+ const peakTx = kpiSource.every(s => s.txPeak !== null) ? kpiSource.reduce((a, s) => Math.max(a, s.txPeak ?? 0), 0) : null
const visibleSortFields = SORT_FIELDS.filter(s => !s.modesOnly || s.modesOnly.includes(effectiveMode))
const ingestLine = flowIngestLine(flowStats)
diff --git a/backend/drizzle/0011_traffic_collection_state.sql b/backend/drizzle/0011_traffic_collection_state.sql
new file mode 100644
index 00000000..f5c5245f
--- /dev/null
+++ b/backend/drizzle/0011_traffic_collection_state.sql
@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS traffic_collection_state (
+ server_id BIGINT PRIMARY KEY REFERENCES servers(id) ON DELETE CASCADE,
+ server_updated_at TIMESTAMPTZ NOT NULL,
+ last_attempt_at TIMESTAMPTZ NOT NULL,
+ last_succeeded_at TIMESTAMPTZ,
+ status TEXT NOT NULL CHECK(status IN ('pending', 'ok', 'error')),
+ error TEXT
+);
diff --git a/backend/package.json b/backend/package.json
index e1268137..8f50dff9 100644
--- a/backend/package.json
+++ b/backend/package.json
@@ -19,7 +19,7 @@
"test:exec": "tsx src/services/mikrotik-logs.test.ts && tsx src/routes/exec.test.ts && tsx src/plugins/auth.exec.test.ts",
"test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts && tsx src/plugins/auth.ospf-optimizer.test.ts",
"test:wireguard": "npx tsx src/services/wireguard-config.test.ts && npx tsx src/services/wireguard-ipv6.test.ts && npx tsx --test src/services/wireguard-import.test.ts src/services/wireguard-import-apply.test.ts src/services/wireguard-import-revision.test.ts src/services/wireguard-delete.test.ts && npx tsx src/routes/wireguard-import.test.ts",
- "test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts && tsx src/services/traffic-observation-sql.test.ts",
+ "test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts && tsx src/services/traffic-freshness.test.ts && tsx src/services/traffic-observation-sql.test.ts && tsx src/services/traffic-history-sql.test.ts",
"test:traffic-flow": "tsx src/routes/traffic-flow-query.test.ts && tsx src/services/traffic-flow-month.test.ts && tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-ipv6-pipeline.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-ifindex.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-dest.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-purge.test.ts && tsx src/services/traffic-flow-geoip.test.ts && tsx src/services/traffic-flow-facts.test.ts && tsx src/services/traffic-flow-facts-filter.test.ts && tsx src/services/traffic-flow-facts-rebuild.test.ts && tsx src/services/statistics-aggregate.test.ts",
"test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts && tsx src/modules/users/interface-inventory.test.ts && tsx src/routes/users-catalog.test.ts",
"test:pg": "tsx src/db/pg-pool.test.ts && tsx src/db/sql-bind.test.ts && tsx src/db/sqlite-json.test.ts && tsx src/db/traffic-flags.test.ts && tsx src/db/pg-schema.test.ts && tsx src/services/config-revisions.test.ts",
diff --git a/backend/src/db/schema.ts b/backend/src/db/schema.ts
index 22807bd7..ab78f791 100644
--- a/backend/src/db/schema.ts
+++ b/backend/src/db/schema.ts
@@ -341,6 +341,15 @@ export const uptimeSettings = pgTable("uptime_settings", {
updatedAt: ts("updated_at").notNull().defaultNow(),
})
+export const trafficCollectionState = pgTable("traffic_collection_state", {
+ serverId: intPkRef().primaryKey().references(() => servers.id, { onDelete: "cascade" }),
+ serverUpdatedAt: ts("server_updated_at").notNull(),
+ lastAttemptAt: ts("last_attempt_at").notNull(),
+ lastSucceededAt: ts("last_succeeded_at"),
+ status: text("status").notNull(),
+ error: text("error"),
+})
+
export const uptimeProbes = pgTable("uptime_probes", {
id: text("id").primaryKey(),
srcServerId: bigint("src_server_id", { mode: "number" }).notNull()
diff --git a/backend/src/db/traffic-flags.test.ts b/backend/src/db/traffic-flags.test.ts
index 1fd05e48..06db8ea3 100644
--- a/backend/src/db/traffic-flags.test.ts
+++ b/backend/src/db/traffic-flags.test.ts
@@ -12,8 +12,10 @@ assert.equal(encodeTrafficFlags(true, true), 3)
assert.equal(encodeTrafficFlags(false, false), 0)
assert.equal(trafficFlagRunning(1), true)
assert.equal(trafficFlagDisabled(1), false)
-assert.deepEqual(decodeTrafficSampleFlags(1), { running: true, disabled: false })
-assert.deepEqual(decodeTrafficSampleFlags(0), { running: false, disabled: false })
-assert.deepEqual(decodeTrafficSampleFlags(undefined), { running: false, disabled: false })
+assert.deepEqual(decodeTrafficSampleFlags(1), { running: true, disabled: false, countersKnown: false })
+assert.deepEqual(decodeTrafficSampleFlags(0), { running: false, disabled: false, countersKnown: false })
+assert.deepEqual(decodeTrafficSampleFlags(undefined), { running: false, disabled: false, countersKnown: false })
+assert.equal(encodeTrafficFlags(true,false,true),5)
+assert.deepEqual(decodeTrafficSampleFlags(5), {running:true,disabled:false,countersKnown:true})
console.log("traffic-flags.test.ts: ok")
diff --git a/backend/src/db/traffic-flags.ts b/backend/src/db/traffic-flags.ts
index a0a813bc..4c1da1e5 100644
--- a/backend/src/db/traffic-flags.ts
+++ b/backend/src/db/traffic-flags.ts
@@ -1,8 +1,10 @@
export const TRAFFIC_FLAG_RUNNING = 1
export const TRAFFIC_FLAG_DISABLED = 2
+export const TRAFFIC_FLAG_COUNTERS_KNOWN = 4
-export function encodeTrafficFlags(running: boolean, disabled: boolean): number {
+export function encodeTrafficFlags(running: boolean, disabled: boolean, countersKnown = false): number {
return (running ? TRAFFIC_FLAG_RUNNING : 0) | (disabled ? TRAFFIC_FLAG_DISABLED : 0)
+ | (countersKnown ? TRAFFIC_FLAG_COUNTERS_KNOWN : 0)
}
export function trafficFlagRunning(flags: number | null | undefined): boolean {
@@ -16,9 +18,11 @@ export function trafficFlagDisabled(flags: number | null | undefined): boolean {
export function decodeTrafficSampleFlags(flags: number | null | undefined): {
running: boolean
disabled: boolean
+ countersKnown: boolean
} {
return {
running: trafficFlagRunning(flags),
disabled: trafficFlagDisabled(flags),
+ countersKnown: ((Number(flags) || 0) & TRAFFIC_FLAG_COUNTERS_KNOWN) !== 0,
}
}
diff --git a/backend/src/routes/traffic.ts b/backend/src/routes/traffic.ts
index a8c71995..72ed2a74 100644
--- a/backend/src/routes/traffic.ts
+++ b/backend/src/routes/traffic.ts
@@ -5,32 +5,28 @@ import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
import { db } from "../db/index.js"
import { serverSnapshots, servers } from "../db/schema.js"
import {
- collectTrafficOnce,
getTrafficCollectorState,
getTrafficSettings,
readServerSamplesInRange,
updateTrafficSettings,
} from "../services/traffic-collector.js"
-import { scheduleAlertEngineAfterDataCollectors } from "../services/alert-collector-hooks.js"
-import { refreshScheduler } from "../services/scheduler.js"
+import { refreshScheduler, runSchedulerJobNow } from "../services/scheduler.js"
import { appendEvent } from "../modules/events/service/events-service.js"
import { MikrotikClient } from "../services/mikrotik.js"
import { getEnabledServerById } from "../services/wireguard-live.js"
+import { readTrafficCollectionContext } from "../services/traffic-collection-state.js"
+import { applyTrafficFreshness } from "../services/traffic-freshness.js"
import {
- bpsToMbps,
buildTrafficFromSamples,
type TrafficObservations,
isLoopbackName,
parseMonitorTraffic,
- rateBpsFromDelta,
} from "../services/traffic-rate.js"
import {
buildBoundInterfaceTraffic,
buildUserTrafficList,
} from "../services/traffic-users.js"
-type SnapshotRow = typeof serverSnapshots.$inferSelect
-
interface TrafficServerDto {
observations?:TrafficObservations
id: string
@@ -53,8 +49,8 @@ interface TrafficInterfaceDto {
name: string
running: boolean
disabled: boolean
- rxNow: number
- txNow: number
+ rxNow: number | null
+ txNow: number | null
}
const LIVE_TICK_MS = 1500
@@ -96,6 +92,7 @@ function buildServerTraffic(
rangeStartMs: number,
rangeEndMs: number,
onlyInterface?: string,
+ context?: Awaited>,
): TrafficServerDto {
const built = buildTrafficFromSamples(rows, rangeStartMs, rangeEndMs, onlyInterface,undefined,true)
return {
@@ -104,7 +101,7 @@ function buildServerTraffic(
site: s.site || "—",
country: s.country || "UN",
status,
- observations:built.observations,
+ observations:built.observations && applyTrafficFreshness(built.observations, s, context?.settings, context?.states.get(s.id), rangeEndMs),
rxNow: built.rxNow,
txNow: built.txNow,
rxPeak: built.rxPeak,
@@ -122,23 +119,6 @@ async function snapshotStatus(serverId: number): Promise {
app.post("/traffic/collect-now", async (_req, reply) => {
try {
- await collectTrafficOnce()
- scheduleAlertEngineAfterDataCollectors()
+ const receipt = await runSchedulerJobNow("traffic")
+ if (!receipt.result || receipt.result.job !== "traffic") return reply.status(502).send({ ok: false, error: receipt.error ?? "Нет подтверждённого результата сбора" })
+ const result = receipt.result
+ if (result.skipped) return reply.status(409).send({ error: "Сбор трафика уже выполняется" })
+ const failures = result.servers.filter(s => !s.ok)
+ if (result.fatalError || failures.length) {
+ const error = result.fatalError ?? failures.map(s => `${s.name}: ${s.error ?? "Ошибка опроса"}`).join("; ")
+ await appendEvent({ level: "warning", eventType: "traffic.collect.manual.failed", sourceModule: "traffic",
+ title: "Сбор трафика завершён с ошибками", message: error })
+ return reply.status(502).send({ ok: false, error, failures, succeeded: result.servers.filter(s => s.ok).map(s => s.serverId) })
+ }
const updated = await getTrafficSettings()
await appendEvent({
level: "info",
@@ -287,6 +276,8 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
lastError: updated.lastError || null,
})
} catch (error) {
+ if (typeof error === "object" && error && "statusCode" in error && error.statusCode === 409)
+ return reply.status(409).send({ error: error instanceof Error ? error.message : "Сбор трафика уже выполняется" })
const msg = error instanceof Error ? error.message : String(error)
await appendEvent({
level: "critical",
@@ -306,9 +297,10 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
const rangeStartMs = rangeEndMs - minutes * 60_000
const sinceIso = new Date(rangeStartMs).toISOString()
const allServers = await db.select().from(servers).where(eq(servers.enabled, true))
+ const context = await readTrafficCollectionContext()
const data = await Promise.all(allServers.map(async (s): Promise => {
const rows = await readServerSamplesInRange(s.id, sinceIso)
- return buildServerTraffic(s, await snapshotStatus(s.id), rows, rangeStartMs, rangeEndMs)
+ return buildServerTraffic(s, await snapshotStatus(s.id), rows, rangeStartMs, rangeEndMs, undefined, context)
}))
return reply.send({ servers: data })
@@ -385,9 +377,10 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
const sinceIso = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString()
const rows = await readServerSamplesInRange(serverId, sinceIso)
+ const context = await readTrafficCollectionContext(), end = Date.now()
const byIface = new Map()
for (const r of rows) {
- if (isLoopbackName(r.interfaceName)) continue
+ if (isLoopbackName(r.interfaceName) || r.peerPublicKey) continue
const arr = byIface.get(r.interfaceName) ?? []
arr.push(r)
byIface.set(r.interfaceName, arr)
@@ -395,19 +388,19 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
const interfaces: TrafficInterfaceDto[] = [...byIface.entries()].map(([name, arr]) => {
const sorted = arr.sort((a, b) => a.sampledAt.localeCompare(b.sampledAt))
const last = sorted[sorted.length - 1]
- const prev = sorted[sorted.length - 2]
if (!last) {
return { name, running: false, disabled: true, rxNow: 0, txNow: 0 }
}
- const now = ifaceNowMbps(prev, last)
+ const built = buildTrafficFromSamples(sorted, Date.parse(sinceIso), end, name, undefined, true)
+ const observed = applyTrafficFreshness(built.observations!, server, context.settings, context.states.get(server.id), end)
return {
name,
running: Boolean(last.running),
disabled: Boolean(last.disabled),
- rxNow: now.rxNow,
- txNow: now.txNow,
+ rxNow: observed.rxNow ? built.rxNow : null,
+ txNow: observed.txNow ? built.txNow : null,
}
- }).sort((a, b) => (b.rxNow + b.txNow) - (a.rxNow + a.txNow))
+ }).sort((a, b) => ((b.rxNow ?? -1) + (b.txNow ?? -1)) - ((a.rxNow ?? -1) + (a.txNow ?? -1)))
return reply.send({ interfaces })
})
@@ -432,6 +425,7 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
rangeStartMs,
rangeEndMs,
q.iface && q.iface !== "__all__" ? q.iface : undefined,
+ await readTrafficCollectionContext(),
)
return reply.send({ server: data })
})
diff --git a/backend/src/services/alert-engine/signals-from-collectors.ts b/backend/src/services/alert-engine/signals-from-collectors.ts
index 12ecf3f2..b0353908 100644
--- a/backend/src/services/alert-engine/signals-from-collectors.ts
+++ b/backend/src/services/alert-engine/signals-from-collectors.ts
@@ -4,6 +4,8 @@ import { collectGreSignalsFromRuns } from "./gre-signals.js"
import { readGreBgpRunSnapshots } from "../gre-bgp-snapshot-history.js"
import { collectSourceSignals, probePeriodMs, type CollectorRunRow, type CollectorSourceConfig } from "./collector-source-signals.js"
import type { SignalSnapshot } from "./types.js"
+import type { TrafficAttempt } from "../traffic-freshness.js"
+import { applyTrafficFreshness } from "../traffic-freshness.js"
export type CollectorRead = (query: string, params?: unknown[]) => Promise
export type CollectorSourceSnapshot = Pick
@@ -18,7 +20,7 @@ async function readRuns(read: CollectorRead, job: string, limit: number, since?:
/** Read-only: do not seed defaults or load credentials while building alert facts. */
export async function readCollectorSourceSnapshot(clock: () => number = Date.now, read: CollectorRead = dbAll): Promise {
const now = clock()
- const [servers, probes, uptime, rest, traffic] = await Promise.all([
+ const [servers, probes, uptime, rest, traffic, attempts] = await Promise.all([
read("SELECT id, name, host, enabled, updated_at AS \"updatedAt\" FROM servers"),
read(`SELECT id, name, target, src_server_id AS "srcServerId", src_interface AS "srcInterface",
enabled, interval_sec AS "intervalSec", updated_at AS "updatedAt" FROM uptime_probes`),
@@ -26,6 +28,8 @@ export async function readCollectorSourceSnapshot(clock: () => number = Date.now
interval_sec AS "intervalSec", probe_interval_sec AS "probeIntervalSec" FROM uptime_settings WHERE id = 1`),
read>('SELECT enabled, interval_sec AS "intervalSec" FROM servers_api_ping_settings WHERE id = 1'),
read>('SELECT enabled, interval_sec AS "intervalSec" FROM traffic_settings WHERE id = 1'),
+ read(`SELECT server_id AS "serverId",server_updated_at AS "serverUpdatedAt",
+ last_attempt_at AS "lastAttemptAt",last_succeeded_at AS "lastSucceededAt",status,error FROM traffic_collection_state`),
])
const config: CollectorSourceConfig = { servers, probes, uptime: uptime[0], rest: rest[0], traffic: traffic[0] }
const periods = probes.filter(p => p.enabled).map(p => probePeriodMs(config, p)).filter(Number.isFinite)
@@ -36,9 +40,17 @@ export async function readCollectorSourceSnapshot(clock: () => number = Date.now
readRuns(read, "uptime_ping", 4096, new Date(now - probeWindowMs).toISOString()),
readRuns(read, "traffic", 8),
])
+ const facts = collectSourceSignals(config, { resources: resourceRuns, rest: restRuns, ping: pingRuns, traffic: trafficRuns }, now)
+ facts.trafficByServer = facts.trafficByServer.filter(signal => {
+ const server = servers.find(s => (s.name || s.host) === signal.serverName)
+ const attempt = server && attempts.find(a => a.serverId === Number(server.id))
+ if (!attempt) return true // Existing scheduler provenance remains readable until this device's first new attempt.
+ return !!server && applyTrafficFreshness({ rxNow: true, txNow: true, rxSeries: [], txSeries: [], sampledAt: signal.sampledAt },
+ { enabled: server.enabled === true, updatedAt: server.updatedAt ?? "" }, traffic[0], attempt, now).currentState === "current"
+ })
return {
sampledAt: new Date(now).toISOString(),
- ...collectSourceSignals(config, { resources: resourceRuns, rest: restRuns, ping: pingRuns, traffic: trafficRuns }, now),
+ ...facts,
}
}
diff --git a/backend/src/services/scheduler-recorded-run.test.ts b/backend/src/services/scheduler-recorded-run.test.ts
index e08e16cc..6fa20b2c 100644
--- a/backend/src/services/scheduler-recorded-run.test.ts
+++ b/backend/src/services/scheduler-recorded-run.test.ts
@@ -55,9 +55,15 @@ test("partial backups, renew and alert errors are explicit while confirmed resul
assert.match(schedulerSnapshotError({ v: 1, job: "alert_engine", sampledAt: "now", rulesChecked: 1, standaloneFires: 0, groupFires: 0, skippedNoTelegram: false, errors: ["send failed"] })!, /send failed/)
})
test("valid offline/down observations are not collector errors", () => {
- assert.equal(schedulerSnapshotError({ v: 1, job: "traffic", sampledAt: "now", servers: [{ serverId: 1, name: "router", host: "host", ok: false, error: "offline" }] }), null)
+ assert.equal(schedulerSnapshotError({ v: 1, job: "traffic", sampledAt: "now", servers: [{ serverId: 1, name: "router", host: "host", ok: true, sumRxMbps: 0, sumTxMbps: 0 }] }), null)
assert.equal(schedulerSnapshotError({ v: 1, job: "uptime_ping", sampledAt: "now", probes: [{ probeId: "probe", name: "probe", target: "host", srcServerId: 1, srcServerName: "router", srcInterface: "ether1", ok: false, rttMs: null, lossPct: 100, status: "down" }] }), null)
})
+test("traffic command errors are limitations; partial measurements remain in the receipt",async()=>{
+ const snapshot={v:1 as const,job:"traffic" as const,sampledAt:"now",servers:[{serverId:1,name:"router",host:"host",ok:false,error:"401"},{serverId:2,name:"second",host:"second",ok:true,sumRxMbps:0,sumTxMbps:0}]}
+ const h=harness(snapshot as never),receipt=await runRecordedSchedulerJob(h.deps)
+ assert.equal(receipt.status,"error");assert.match(receipt.error!,/router: 401/)
+ assert.equal(receipt.result,snapshot);assert.equal(h.records.length,1);assert.equal(h.events[0].level,"warning")
+})
test("skipped/not-due snapshots preserve their reason and do not announce completed work", async () => {
for (const snapshot of [backup({ due: false }), backup({ skipped: true })]) {
const h = harness(snapshot); const receipt = await runRecordedSchedulerJob(h.deps)
diff --git a/backend/src/services/scheduler-recorded-run.ts b/backend/src/services/scheduler-recorded-run.ts
index 68554200..4c9a1a13 100644
--- a/backend/src/services/scheduler-recorded-run.ts
+++ b/backend/src/services/scheduler-recorded-run.ts
@@ -20,6 +20,9 @@ export function schedulerSnapshotError(snapshot: SchedulerRunSnapshot | null): s
if (!snapshot) return null
if ("fatalError" in snapshot && snapshot.fatalError) return snapshot.fatalError
const errors = "errors" in snapshot ? [...(snapshot.errors ?? [])] : []
+ if (snapshot.job === "traffic") for (const server of snapshot.servers) {
+ if (!server.ok) errors.push(`${server.name}: ${server.error ?? "Ошибка опроса трафика"}`)
+ }
if (snapshot.job === "uptime_speed") {
for (const run of snapshot.runs) if (!run.ok && run.error) errors.push(run.error)
}
diff --git a/backend/src/services/traffic-collection-state.ts b/backend/src/services/traffic-collection-state.ts
new file mode 100644
index 00000000..59758a6a
--- /dev/null
+++ b/backend/src/services/traffic-collection-state.ts
@@ -0,0 +1,27 @@
+import { and, eq, lte } from "drizzle-orm"
+import { db } from "../db/index.js"
+import { trafficCollectionState, trafficSettings, type Server } from "../db/schema.js"
+
+export async function recordTrafficCollectionAttempt(server: Pick,
+ status: "pending" | "ok" | "error", sampledAt: string, error?: string) {
+ if (status === "pending") {
+ const next = { serverId: server.id, serverUpdatedAt: server.updatedAt,
+ lastAttemptAt: sampledAt, status, error: null }
+ await db.insert(trafficCollectionState).values(next).onConflictDoUpdate({
+ target: trafficCollectionState.serverId, set: next,
+ setWhere: lte(trafficCollectionState.lastAttemptAt, sampledAt),
+ })
+ } else {
+ await db.update(trafficCollectionState).set({ status, error: error?.slice(0, 1024) ?? null,
+ ...(status === "ok" ? { lastSucceededAt: sampledAt } : {}) }).where(and(
+ eq(trafficCollectionState.serverId, server.id), eq(trafficCollectionState.lastAttemptAt, sampledAt),
+ ))
+ }
+}
+
+/** Read-only: does not create default settings or contact a device. */
+export async function readTrafficCollectionContext() {
+ const [states, settings] = await Promise.all([db.select().from(trafficCollectionState),
+ db.select().from(trafficSettings).where(eq(trafficSettings.id, 1)).limit(1)])
+ return { states: new Map(states.map(s => [s.serverId, s])), settings: settings[0] }
+}
diff --git a/backend/src/services/traffic-collector.ts b/backend/src/services/traffic-collector.ts
index 11e5f77e..19af9f12 100644
--- a/backend/src/services/traffic-collector.ts
+++ b/backend/src/services/traffic-collector.ts
@@ -8,6 +8,7 @@ import { MikrotikClient } from "./mikrotik.js"
import { bpsToMbps, shouldIncludeIface } from "./traffic-rate.js"
import { observedTrafficRate, parseTrafficCounter } from "./traffic-observation.js"
import { rememberServerIfaces } from "./traffic-flow-ifindex.js"
+import { recordTrafficCollectionAttempt } from "./traffic-collection-state.js"
interface RosIfaceTraffic {
".id"?: string
@@ -86,6 +87,7 @@ async function readPreviousWave(serverId: number): Promise