Add Gateway traffic totals and dashboard chart

src/server/services/deviceInventoryService.js

@@ -5,7 +5,7 @@ import { HarborError } from '../../shared/errors.js';
 import { isDeviceInterface } from '../adapters/neighbors.js';
 import { fingerprintDirectDevices } from './devicePolicyService.js';
 
-export const DEVICE_INVENTORY_SCHEMA_VERSION = 2;
+export const DEVICE_INVENTORY_SCHEMA_VERSION = 3;
 const ONLINE_MS = 2 * 60 * 1000;
 const RECENT_MS = 24 * 60 * 60 * 1000;
 const RETENTION_MS = 30 * 24 * 60 * 60 * 1000;
@@ -47,6 +47,15 @@ const DEFAULT_PROXY_TRAFFIC = {
   rebaselineMacs: [],
 };
 
+const DEFAULT_GLOBAL_TRAFFIC_SOURCE = {
+  epoch: null,
+  lastObservedAt: null,
+  uploadBytes: '0',
+  downloadBytes: '0',
+  baselinesByMac: {},
+  rebaselineMacs: [],
+};
+
 const DEFAULT_STATE = {
   schemaVersion: DEVICE_INVENTORY_SCHEMA_VERSION,
   revision: 0,
@@ -62,6 +71,10 @@ const DEFAULT_STATE = {
     totalsByMac: {},
     rebaselineMacs: [],
     proxy: DEFAULT_PROXY_TRAFFIC,
+    global: {
+      gateway: DEFAULT_GLOBAL_TRAFFIC_SOURCE,
+      proxy: DEFAULT_GLOBAL_TRAFFIC_SOURCE,
+    },
   },
   devices: [],
 };
@@ -77,6 +90,57 @@ const parseStoredCounter = (value) => {
   return COUNTER_PATTERN.test(counter) ? BigInt(counter).toString() : null;
 };
 
+const sumStoredTotals = (totalsByMac, key) => recordEntries(totalsByMac)
+  .reduce((total, [, value]) => total + BigInt(value?.[key] || '0'), 0n)
+  .toString();
+
+function normalizeGlobalTrafficSource(value, fallback, version) {
+  const source = value && typeof value === 'object' && !Array.isArray(value) ? value : {};
+  const fallbackMacs = new Set([
+    ...Object.keys(fallback.baselinesByMac),
+    ...Object.keys(fallback.totalsByMac),
+    ...fallback.rebaselineMacs,
+  ]);
+  if (version < 3 || source !== value) {
+    return {
+      epoch: fallback.epoch,
+      lastObservedAt: fallback.lastObservedAt,
+      uploadBytes: sumStoredTotals(fallback.totalsByMac, 'uploadBytes'),
+      downloadBytes: sumStoredTotals(fallback.totalsByMac, 'downloadBytes'),
+      baselinesByMac: structuredClone(fallback.baselinesByMac),
+      rebaselineMacs: [...fallback.rebaselineMacs],
+    };
+  }
+  const rebaselineMacs = new Set((Array.isArray(source.rebaselineMacs) ? source.rebaselineMacs : [])
+    .map(normalizeMac).filter((mac) => MAC_PATTERN.test(mac)));
+  const baselinesByMac = {};
+  let recovered = !source.baselinesByMac || typeof source.baselinesByMac !== 'object'
+    || Array.isArray(source.baselinesByMac);
+  for (const [rawMac, baseline] of recordEntries(source.baselinesByMac)) {
+    const mac = normalizeMac(rawMac);
+    const uploadBytes = parseStoredCounter(baseline?.uploadBytes);
+    const downloadBytes = parseStoredCounter(baseline?.downloadBytes);
+    if (!MAC_PATTERN.test(mac) || typeof baseline?.epoch !== 'string' || !baseline.epoch
+      || uploadBytes == null || downloadBytes == null) {
+      if (MAC_PATTERN.test(mac)) rebaselineMacs.add(mac);
+      recovered = true;
+      continue;
+    }
+    baselinesByMac[mac] = { epoch: baseline.epoch, uploadBytes, downloadBytes };
+  }
+  if (recovered) for (const mac of fallbackMacs) rebaselineMacs.add(mac);
+  return {
+    epoch: typeof source.epoch === 'string' ? source.epoch : fallback.epoch,
+    lastObservedAt: typeof source.lastObservedAt === 'string' ? source.lastObservedAt : fallback.lastObservedAt,
+    uploadBytes: parseStoredCounter(source.uploadBytes)
+      ?? sumStoredTotals(fallback.totalsByMac, 'uploadBytes'),
+    downloadBytes: parseStoredCounter(source.downloadBytes)
+      ?? sumStoredTotals(fallback.totalsByMac, 'downloadBytes'),
+    baselinesByMac,
+    rebaselineMacs: [...rebaselineMacs],
+  };
+}
+
 function normalizeProxyTraffic(value, devices) {
   const proxy = value && typeof value === 'object' && !Array.isArray(value) ? value : {};
   if (Number.isSafeInteger(proxy.schemaVersion) && proxy.schemaVersion > 1) {
@@ -264,6 +328,22 @@ export function migrateDeviceInventoryState(value) {
       recoveredTraffic = true;
     }
   }
+  const global = {
+    gateway: normalizeGlobalTrafficSource(traffic.global?.gateway, {
+      epoch: typeof traffic.epoch === 'string' ? traffic.epoch : null,
+      lastObservedAt: typeof traffic.lastObservedAt === 'string' ? traffic.lastObservedAt : null,
+      baselinesByMac,
+      totalsByMac,
+      rebaselineMacs: [...rebaselineMacs],
+    }, version),
+    proxy: normalizeGlobalTrafficSource(traffic.global?.proxy, {
+      epoch: Object.values(proxyTraffic.baselinesByMac)[0]?.epoch || null,
+      lastObservedAt: proxyTraffic.lastObservedAt,
+      baselinesByMac: proxyTraffic.baselinesByMac,
+      totalsByMac: proxyTraffic.totalsByMac,
+      rebaselineMacs: proxyTraffic.rebaselineMacs,
+    }, version),
+  };
   return {
     ...DEFAULT_STATE,
     ...state,
@@ -280,6 +360,7 @@ export function migrateDeviceInventoryState(value) {
       totalsByMac,
       rebaselineMacs: [...rebaselineMacs].filter((mac) => MAC_PATTERN.test(mac)),
       proxy: proxyTraffic,
+      global,
     },
     devices,
   };
@@ -292,6 +373,43 @@ function deviceStatus(lastSeenAt, now) {
   return 'offline';
 }
 
+function accumulateGlobalTraffic(source, countersByMac, epoch, observedAt, label) {
+  const epochChanged = Boolean(source.epoch && source.epoch !== epoch);
+  const baselinesByMac = epochChanged ? {} : { ...source.baselinesByMac };
+  const rebaselineMacs = new Set(epochChanged ? [] : source.rebaselineMacs);
+  let uploadBytes = BigInt(source.uploadBytes);
+  let downloadBytes = BigInt(source.downloadBytes);
+  for (const [mac, processTotal] of countersByMac) {
+    const baseline = baselinesByMac[mac];
+    const recovering = rebaselineMacs.has(mac);
+    const sameEpoch = !recovering && baseline?.epoch === epoch;
+    const baselineUpload = sameEpoch ? BigInt(baseline.uploadBytes) : 0n;
+    const baselineDownload = sameEpoch ? BigInt(baseline.downloadBytes) : 0n;
+    if (processTotal.upload < baselineUpload || processTotal.download < baselineDownload) {
+      throw new Error(`Dataplane ${label} traffic counter уменьшился внутри одного epoch`);
+    }
+    if (!recovering) {
+      uploadBytes += processTotal.upload - baselineUpload;
+      downloadBytes += processTotal.download - baselineDownload;
+    }
+    baselinesByMac[mac] = {
+      epoch,
+      uploadBytes: processTotal.upload.toString(),
+      downloadBytes: processTotal.download.toString(),
+    };
+    rebaselineMacs.delete(mac);
+  }
+  // ponytail: per-MAC baselines live for one dataplane epoch; use a dataplane-wide counter if MAC churn becomes large.
+  return {
+    epoch,
+    lastObservedAt: observedAt || source.lastObservedAt,
+    uploadBytes: uploadBytes.toString(),
+    downloadBytes: downloadBytes.toString(),
+    baselinesByMac,
+    rebaselineMacs: [...rebaselineMacs],
+  };
+}
+
 export function createDeviceInventoryService({
   store,
   observe,
@@ -305,6 +423,8 @@ export function createDeviceInventoryService({
   let policyQueue = Promise.resolve();
   const trafficHistoryByMac = new Map();
   const trafficCursorByMac = new Map();
+  let globalTrafficHistory = [];
+  let globalTrafficCursor = null;
 
   function captureTrafficHistory(state) {
     const knownMacs = new Set(state.devices.map(({ mac }) => mac));
@@ -332,6 +452,20 @@ export function createDeviceInventoryService({
         trafficHistoryByMac.delete(mac);
       }
     }
+    const gatewaySource = state.traffic.global.gateway;
+    const proxySource = state.traffic.global.proxy;
+    const gateway = BigInt(gatewaySource.uploadBytes) + BigInt(gatewaySource.downloadBytes);
+    const proxy = BigInt(proxySource.uploadBytes) + BigInt(proxySource.downloadBytes);
+    const signature = `${gatewaySource.lastObservedAt || ''}|${proxySource.lastObservedAt || ''}`;
+    const previous = globalTrafficCursor;
+    globalTrafficCursor = { signature, gateway, proxy };
+    if (previous && previous.signature !== signature) {
+      globalTrafficHistory = [...globalTrafficHistory, {
+        observedAt: [gatewaySource.lastObservedAt, proxySource.lastObservedAt].filter(Boolean).sort().at(-1),
+        gatewayBytes: gateway > previous.gateway ? (gateway - previous.gateway).toString() : '0',
+        proxyBytes: proxy > previous.proxy ? (proxy - previous.proxy).toString() : '0',
+      }].slice(-TRAFFIC_HISTORY_LIMIT);
+    }
   }
 
   function serializePolicy(action) {
@@ -406,9 +540,29 @@ export function createDeviceInventoryService({
       || rank[left.status] - rank[right.status]
       || String(right.lastSeenAt).localeCompare(String(left.lastSeenAt))
     ));
+    const gatewayTraffic = state.traffic.global.gateway;
+    const proxyTraffic = state.traffic.global.proxy;
+    const gatewayBytes = BigInt(gatewayTraffic.uploadBytes) + BigInt(gatewayTraffic.downloadBytes);
+    const proxyBytes = BigInt(proxyTraffic.uploadBytes) + BigInt(proxyTraffic.downloadBytes);
+    const contributingTimes = [
+      gatewayBytes > 0n ? gatewayTraffic.lastObservedAt : null,
+      proxyBytes > 0n ? proxyTraffic.lastObservedAt : null,
+    ].filter(Boolean);
+    const observedTimes = contributingTimes.length
+      ? contributingTimes
+      : [gatewayTraffic.lastObservedAt, proxyTraffic.lastObservedAt].filter(Boolean);
     return {
       revision: state.revision,
       trafficHistoryCapacity: TRAFFIC_HISTORY_LIMIT,
+      traffic: {
+        gatewayBytes: gatewayBytes.toString(),
+        proxyBytes: proxyBytes.toString(),
+        totalBytes: (gatewayBytes + proxyBytes).toString(),
+        gatewayObservedAt: gatewayTraffic.lastObservedAt,
+        proxyObservedAt: proxyTraffic.lastObservedAt,
+        observedAt: observedTimes.sort()[0] || null,
+        history: globalTrafficHistory,
+      },
       source: {
         kind: 'neighbor',
         lastObservedAt: state.lastObservedAt,
@@ -678,6 +832,13 @@ export function createDeviceInventoryService({
               };
               rebaselineMacs.delete(mac);
             }
+            const globalGateway = accumulateGlobalTraffic(
+              traffic.global.gateway,
+              processByMac,
+              trafficResult.epoch,
+              trafficResult.observedAt || traffic.lastObservedAt,
+              'Gateway',
+            );
             for (const mac of Object.keys(totalsByMac)) {
               if (!knownMacs.has(mac)) {
                 delete totalsByMac[mac];
@@ -714,6 +875,7 @@ export function createDeviceInventoryService({
               totalsByMac: proxyTotals,
               rebaselineMacs: [...proxyRebaseline],
             };
+            let globalProxy = traffic.global.proxy;
             if (proxySampleError) {
               proxy = { ...proxy, lastError: proxySampleError };
             } else if (proxyRows) {
@@ -746,6 +908,13 @@ export function createDeviceInventoryService({
                   };
                   nextProxyRebaseline.delete(mac);
                 }
+                const nextGlobalProxy = accumulateGlobalTraffic(
+                  traffic.global.proxy,
+                  proxyByMac,
+                  trafficResult.epoch,
+                  trafficResult.observedAt || proxy.lastObservedAt,
+                  'proxy',
+                );
                 proxy = {
                   ...proxy,
                   lastObservedAt: trafficResult.observedAt || proxy.lastObservedAt,
@@ -755,6 +924,7 @@ export function createDeviceInventoryService({
                   totalsByMac: nextProxyTotals,
                   rebaselineMacs: [...nextProxyRebaseline],
                 };
+                globalProxy = nextGlobalProxy;
               } catch (error) {
                 proxy = { ...proxy, lastError: error.message || String(error) };
               }
@@ -770,6 +940,7 @@ export function createDeviceInventoryService({
               totalsByMac,
               rebaselineMacs: [...rebaselineMacs],
               proxy,
+              global: { gateway: globalGateway, proxy: globalProxy },
             };
           } catch (error) {
             traffic = { ...traffic, lastError: error.message || String(error) };