diff --git a/backend/src/services/traffic-flow-analytics.test.ts b/backend/src/services/traffic-flow-analytics.test.ts index 1868afff..3c7f6670 100644 --- a/backend/src/services/traffic-flow-analytics.test.ts +++ b/backend/src/services/traffic-flow-analytics.test.ts @@ -4,7 +4,7 @@ import { ingestParsedFlowsForServerForTests, resetFlowRingsForTests, } from "./traffic-flow-ingest.js" -import { buildFlowAnalytics, formatLiveSseFromBuilder, getFlowMonthly, listFlowClients, listFlowExporters } from "./traffic-flow-analytics.js" +import { buildFlowAnalytics, formatLiveSseFromBuilder, getFlowMonthly, listFlowClients, listFlowExporters, resetFlowAnalyticsCacheForTests } from "./traffic-flow-analytics.js" import { withPgOrSkip } from "../test/pg.js" import { dbQuery } from "../db/index.js" import { seedFlowTopologyForTests, type FlowTopology } from "./traffic-flow-topology.js" @@ -556,6 +556,22 @@ try { } } +// Historical 5m/15m charts survive an empty live ring (e.g. immediately after restart). +resetFlowRingsForTests() +await dbQuery("DELETE FROM flow_buckets WHERE server_id=$1", [serverId]) +for (const minutesAgo of [2, 3, 4, 10]) { + await dbQuery(`INSERT INTO flow_buckets (server_id,bucket_at,src,dst,in_iface,bytes,packets) + VALUES ($1,date_trunc('minute',now())-($2::int*interval '1 minute'),'10.1.1.8','8.8.8.8','2',60000,10)`, [serverId, minutesAgo]) +} +for (const minutes of [5, 15]) for (const dedup of [true, false]) { + resetFlowAnalyticsCacheForTests() + const history = await buildFlowAnalytics({ minutes, serverId, dedup, excludeMesh:false, excludeOverlay:false }) + assert.equal(history.rxSeries.length, minutes) + assert.equal(history.rxSeries.filter(v => v > 0).length, minutes === 5 ? 3 : 4) + const missing = await buildFlowAnalytics({ minutes, serverId, iface:"not-an-interface" }) + assert(missing.rxSeries.every(v => v === 0), "Historical chart respects interface filter") +} + } finally { for (const table of ["flow_daily_dims", "flow_hour_facts", "flow_daily_facts", "flow_buckets", "traffic_samples", "user_interface_bindings", "server_snapshots"]) { await dbQuery("DELETE FROM " + table + " WHERE server_id = $1", [serverId]) diff --git a/backend/src/services/traffic-flow-analytics.ts b/backend/src/services/traffic-flow-analytics.ts index b5fd6497..b18907d4 100644 --- a/backend/src/services/traffic-flow-analytics.ts +++ b/backend/src/services/traffic-flow-analytics.ts @@ -423,9 +423,10 @@ async function buildFlowAnalyticsUncached(q: FlowAnalyticsQuery): Promise b[1].bytes - a[1].bytes) diff --git a/backend/src/services/traffic-flow-ingest.ts b/backend/src/services/traffic-flow-ingest.ts index 093a2bd3..8919d44d 100644 --- a/backend/src/services/traffic-flow-ingest.ts +++ b/backend/src/services/traffic-flow-ingest.ts @@ -439,7 +439,7 @@ export async function listStoredFlowRows(sinceIso: string): Promise { const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString() - if (minutes <= 15 && !worker) return await listLiveFlowRows(sinceIso) + // History remains available while the collector starts or is stopped; pending rows are merged below. return await listStoredFlowRows(sinceIso) } diff --git a/backend/src/services/traffic-rate.test.ts b/backend/src/services/traffic-rate.test.ts index 4d516465..d5943e63 100644 --- a/backend/src/services/traffic-rate.test.ts +++ b/backend/src/services/traffic-rate.test.ts @@ -132,7 +132,7 @@ const observed=(rows:ReturnType[])=>buildTrafficFromSa assert.ok(observed([]).observations?.rxSeries.every(v=>!v));assert.equal(observed([]).observations?.rxNow,false) assert.equal(observed([observationSample(0,10,10)]).observations?.rxNow,false) const observedZero=observed([observationSample(0,10,10),observationSample(30_000,10,10)]) -assert.equal(observedZero.rxNow,0);assert.equal(observedZero.observations?.rxNow,true);assert.equal(observedZero.observations?.rxSeries.filter(Boolean).length,1) +assert.equal(observedZero.rxNow,0);assert.equal(observedZero.observations?.rxNow,true);assert.equal(observedZero.observations?.rxSeries.filter(Boolean).length,30) const partialReset=observed([observationSample(0,10,10),observationSample(30_000,1,20)]) assert.equal(partialReset.observations?.rxNow,false);assert.equal(partialReset.observations?.txNow,true) const fullReset=observed([observationSample(0,10,10),observationSample(20_000,20,20),observationSample(40_000,1,1)]) @@ -147,3 +147,28 @@ console.log("traffic observation metadata tests ok") const complete=observed([observationSample(0,10,10),observationSample(30_000,20,20),observationSample(0,10,10,"ether2"),observationSample(30_000,20,20,"ether2")]);assert.equal(complete.observations?.rxNow,true) const badTime=observed([{...observationSample(0,10,10),sampledAt:"invalid"},observationSample(30_000,20,20)]);assert.equal(badTime.observations?.rxNow,false) assert.equal(rateBpsFromDelta(0,Number.MAX_VALUE,0,1000),null) + +// A 30-second collector must produce a line across measured intervals on short ranges. +for (const minutes of [5, 15]) { + const span = minutes * 60_000 + const rows = Array.from({ length: span / 30_000 + 1 }, (_, i) => ({ + ...observationSample(i * 30_000, i * 3_750_000, i * 1_875_000), + continuityKnown: true, nativeCounterIdentity: "ether1-v1", + })) + const history = buildTrafficFromSamples(rows, observationStart, observationStart + span, undefined, undefined, true) + assert(history.observations!.rxSeries.every(Boolean), `${minutes}m continuous measured RX`) + assert(history.observations!.txSeries.every(Boolean), `${minutes}m continuous measured TX`) + assert(history.rxSeries.every(v => v === 1)); assert(history.txSeries.every(v => v === 0.5)) + assert.equal(history.rxTotalGiB, (span / 30_000) * 3_750_000 / 1024 ** 3) + // Explicit loss of continuity stays unknown; it must not be bridged by chart rendering. + rows[2].continuityKnown = false + const gap = buildTrafficFromSamples(rows, observationStart, observationStart + span, undefined, undefined, true) + assert.equal(gap.observations!.rxSeries.filter(v => !v).length, 30_000 / (span / 60)) + assert(gap.observations!.rxSeries.slice(0, 30_000 / (span / 60)).every(Boolean)) + const partial = buildTrafficFromSamples(rows.slice(4), observationStart, observationStart + span, undefined, undefined, true) + assert(partial.observations!.rxSeries.slice(0, 120_000 / (span / 60)).every(v => !v), "No history invented before first sample") +} +// Unequal poll intervals contribute by covered duration when sharing a chart bucket. +const weighted = observed([observationSample(0,0,0),observationSample(500,62_500,62_500),observationSample(1500,312_500,312_500)]) +assert.equal(weighted.rxSeries[0], 1.5) +assert.equal(weighted.rxSeries[1], 2) diff --git a/backend/src/services/traffic-rate.ts b/backend/src/services/traffic-rate.ts index 73a8478d..a2aa6c6d 100644 --- a/backend/src/services/traffic-rate.ts +++ b/backend/src/services/traffic-rate.ts @@ -114,6 +114,34 @@ function parseIsoMs(iso: string): number { return Date.parse(iso) } +/** A counter delta measures its entire interval, not just the next poll's timestamp. */ +function intervalSeries( + waves: Map, start: number, end: number, side: "rx" | "tx", +): { values: number[]; observed: boolean[] } { + const sums = Array(SERIES_POINTS).fill(0) as number[] + const durations = Array(SERIES_POINTS).fill(0) as number[] + const unknown = Array(SERIES_POINTS).fill(false) as boolean[] + const width = (end - start) / SERIES_POINTS + if (!(width > 0)) return { values: sums, observed: unknown } + for (const [at, wave] of waves) { + const from = Math.max(start, wave.start), to = Math.min(end, at) + if (!(to > from)) continue + const first = Math.max(0, Math.floor((from - start) / width)) + const last = Math.min(SERIES_POINTS - 1, Math.ceil((to - start) / width) - 1) + for (let i = first; i <= last; i++) { + const overlap = Math.min(to, start + (i + 1) * width) - Math.max(from, start + i * width) + if (!(overlap > 0)) continue + sums[i] += wave[side] * overlap + durations[i] += overlap + unknown[i] ||= !wave[side === "rx" ? "rxKnown" : "txKnown"] + } + } + return { + values: sums.map((sum, i) => durations[i] > 0 ? sum / durations[i] : 0), + observed: durations.map((duration, i) => duration > 0 && !unknown[i]), + } +} + export function sampleSeriesKey(interfaceName: string, peerPublicKey = ""): string { return `${interfaceName}\0${peerPublicKey}` } @@ -232,8 +260,6 @@ export function buildTrafficFromSamples( const latestWave = membership.get(latestMembershipAt) if (latestWave?.declared) sessions = latestWave.active - const rxSeries = bucketAvg(rxPoints, rangeStartMs, rangeEndMs) - const txSeries = bucketAvg(txPoints, rangeStartMs, rangeEndMs) const lastTs = [...byTs.keys()].sort((a, b) => a - b).at(-1) const last = lastTs != null ? byTs.get(lastTs) : undefined const peak = (points: Array<{t:number;v:number}>, side: "rxObserved"|"txObserved") => points.reduce((m,p) => { @@ -249,6 +275,8 @@ export function buildTrafficFromSamples( const complete = wave.sameStart && wave.count === expectedAt(t) waves.set(t,{start:wave.start,rx:wave.rx,tx:wave.tx,rxKnown:complete&&wave.rxObserved,txKnown:complete&&wave.txObserved}) } + const rxHistory = intervalSeries(waves, rangeStartMs, rangeEndMs, "rx") + const txHistory = intervalSeries(waves, rangeStartMs, rangeEndMs, "tx") return retainPeakWaves({ rxNow: last?.rx ?? 0, txNow: last?.tx ?? 0, @@ -257,13 +285,13 @@ export function buildTrafficFromSamples( rxTotalGiB: rxBytesDelta / (1024 ** 3), txTotalGiB: txBytesDelta / (1024 ** 3), sessions, - rxSeries, - txSeries, + rxSeries: rxHistory.values, + txSeries: txHistory.values, ...(includeObservations?{observations:{ rxNow:lastAttemptAt!==null&&lastTs===lastAttemptAt&&!!last?.rxObserved&&last.count===expectedAt(lastTs!), txNow:lastAttemptAt!==null&&lastTs===lastAttemptAt&&!!last?.txObserved&&last.count===expectedAt(lastTs!), - rxSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.rxObserved&&v.count===expectedAt(t)&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1), - txSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.txObserved&&v.count===expectedAt(t)&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1), + rxSeries:rxHistory.observed, + txSeries:txHistory.observed, sampledAt:lastAttemptAt===null?null:new Date(lastAttemptAt).toISOString(), measurementIntervalMs:lastTs===lastAttemptAt?last?.intervalMs??null:null, rxPeak:[...waves.values()].some(w=>w.rxKnown), diff --git a/tests/e2e/traffic-short-history.spec.ts b/tests/e2e/traffic-short-history.spec.ts new file mode 100644 index 00000000..2244ba04 --- /dev/null +++ b/tests/e2e/traffic-short-history.spec.ts @@ -0,0 +1,38 @@ +import {test, expect} from '@playwright/test' +import {buildTrafficFromSamples} from '../../backend/src/services/traffic-rate' + +for (const mode of ['servers','flows'] as const) for (const minutes of [5,15]) { + test(`${mode}: ${minutes} минут рисуются линией`, async ({page}) => { + const end=Date.now(), name='Короткая история' + const history=(span:number)=>buildTrafficFromSamples(Array.from({length:span*2+1},(_,i)=>({ + interfaceName:'ether1',sampledAt:new Date(end-span*60000+i*30000).toISOString(), + rxBytes:i*3750000,txBytes:i*1875000,rxBps:0,txBps:0,running:true,disabled:false, + })),end-span*60000,end,undefined,undefined,true) + await page.route('**/health',r=>r.fulfill({json:{ok:true}})) + await page.route('**/api/**',r=>{ + const url=new URL(r.request().url()),path=url.pathname,span=url.searchParams.get('range')==='5m'?5:url.searchParams.get('range')==='15m'?15:60 + const built=history(span),server={id:'1',name,site:'TEST',country:'RU',status:'online',...built,rxTotal:1,txTotal:1,volumeUnit:'GiB'} + const card={id:'1',name,subtitle:'',site:'TEST',country:'RU',status:'online',rxNow:1,txNow:0,sessions:1,bytes:60000,rxSeries:[1,2,1],txSeries:[0,0,0]} + const analytics={bpsNow:1000000,bytes:60000,packets:10,conversations:1,uniqueSrc:1,uniqueDst:1,topProto:'TCP',rxSeries:Array.from({length:span},(_,i)=>1+i%3),txSeries:Array(span).fill(0),applications:[],protocols:[],sources:[],destinations:[],interfaces:[],conversationsList:[],ifaces:[],live:false} + if(path.endsWith('/live'))return r.fulfill({status:503,json:{error:'Live выключен в проверке истории'}}) + const json=path==='/api/auth/config'?{required:false}:path==='/api/traffic/servers'?{servers:[server]}: + path==='/api/traffic/servers/1'?{server}:path==='/api/traffic/flow/exporters'?{exporters:[card]}: + path==='/api/traffic/flow/analytics'?analytics:path==='/api/traffic/flow/clients'?{clients:[]}: + path==='/api/traffic/flows'?{exportersOnline:1,bytesPerMin:1,uniqueSrc:1,uniqueDst:1,topProto:'TCP',talkers:[],listenerBound:true}: + path==='/api/traffic/users'?{users:[]}:path.endsWith('/interfaces')||path.endsWith('/bound-interfaces')?{interfaces:[]}: + path==='/api/servers'?[]:{} + return r.fulfill({json}) + }) + await page.goto('/traffic') + if(mode==='flows')await page.getByRole('group',{name:'Разделы трафика'}).getByRole('button',{name:'Потоки',exact:true}).click() + await page.getByRole('button',{name:`${minutes}м`,exact:true}).click() + const chart=page.getByRole('region',{name:'История RX/TX'}) + await expect(chart).toBeVisible() + const curves=chart.locator('path.recharts-line-curve') + await expect(curves).toHaveCount(2) + await expect.poll(async()=>((await curves.first().getAttribute('d'))?.match(/[CL]/g)||[]).length).toBeGreaterThan(2) + await expect.poll(()=>curves.first().evaluate(el=>(el as SVGGraphicsElement).getBBox().width/el.closest('[role="region"]')!.clientWidth)).toBeGreaterThan(0.7) + await expect(chart.locator('[data-isolated-direction]')).toHaveCount(0) + await page.screenshot({path:`tmp/netflow-agent-stand/traffic-short-${mode}-${minutes}.png`,fullPage:true}) + }) +}