Track proxy traffic separately in device inventory

2026-08-07 17:08:24 +03:00
parent 9f31eaf396
commit 53e6cf2146
13 changed files with 714 additions and 146 deletions
@@ -136,6 +136,8 @@ test('device traffic totals persist exact deltas across polls and process epochs
      interface: 'eth0',
      uploadBytes: '9007199254740993',
      downloadBytes: '100',
+      proxyUploadBytes: '1000',
+      proxyDownloadBytes: '2000',
    }],
  };
  let trafficError = null;
@@ -153,7 +155,13 @@ test('device traffic totals persist exact deltas across polls and process epochs
  assert.equal(snapshot.devices[0].uploadBytes, '9007199254740993');
  assert.equal(snapshot.devices[0].downloadBytes, '100');
  assert.equal(snapshot.devices[0].trafficObservedAt, observedAt);
-  assert.deepEqual(snapshot.source.traffic, { lastObservedAt: observedAt, error: null });
+  assert.equal(snapshot.devices[0].proxyUploadBytes, '1000');
+  assert.equal(snapshot.devices[0].proxyDownloadBytes, '2000');
+  assert.deepEqual(snapshot.source.traffic, {
+    lastObservedAt: observedAt,
+    error: null,
+    proxy: { lastObservedAt: observedAt, error: null },
+  });

  snapshot = await service.refresh();
  assert.equal(snapshot.devices[0].uploadBytes, '9007199254740993');
@@ -161,11 +169,19 @@ test('device traffic totals persist exact deltas across polls and process epochs

  traffic = {
    ...traffic,
-    devices: [{ ...traffic.devices[0], uploadBytes: '9007199254740995', downloadBytes: '150' }],
+    devices: [{
+      ...traffic.devices[0],
+      uploadBytes: '9007199254740995',
+      downloadBytes: '150',
+      proxyUploadBytes: '1010',
+      proxyDownloadBytes: '2050',
+    }],
  };
  snapshot = await service.refresh();
  assert.equal(snapshot.devices[0].uploadBytes, '9007199254740995');
  assert.equal(snapshot.devices[0].downloadBytes, '150');
+  assert.equal(snapshot.devices[0].proxyUploadBytes, '1010');
+  assert.equal(snapshot.devices[0].proxyDownloadBytes, '2050');

  service = createService();
  snapshot = await service.refresh();
@@ -176,11 +192,19 @@ test('device traffic totals persist exact deltas across polls and process epochs
    ...traffic,
    epoch: 'epoch-b',
    generation: 'rules-b',
-    devices: [{ ...traffic.devices[0], uploadBytes: '10', downloadBytes: '20' }],
+    devices: [{
+      ...traffic.devices[0],
+      uploadBytes: '10',
+      downloadBytes: '20',
+      proxyUploadBytes: '3',
+      proxyDownloadBytes: '4',
+    }],
  };
  snapshot = await service.refresh();
  assert.equal(snapshot.devices[0].uploadBytes, '9007199254741005');
  assert.equal(snapshot.devices[0].downloadBytes, '170');
+  assert.equal(snapshot.devices[0].proxyUploadBytes, '1013');
+  assert.equal(snapshot.devices[0].proxyDownloadBytes, '2054');

  traffic = {
    ...traffic,
@@ -190,13 +214,114 @@ test('device traffic totals persist exact deltas across polls and process epochs
  assert.equal(snapshot.devices[0].uploadBytes, '9007199254741005');
  assert.match(snapshot.source.traffic.error, /уменьшился/);

+  traffic = {
+    ...traffic,
+    devices: [{
+      ...traffic.devices[0],
+      uploadBytes: '12',
+      downloadBytes: '25',
+      proxyUploadBytes: '2',
+      proxyDownloadBytes: '4',
+    }],
+  };
+  snapshot = await service.refresh();
+  assert.equal(snapshot.devices[0].uploadBytes, '9007199254741007');
+  assert.equal(snapshot.devices[0].downloadBytes, '175');
+  assert.equal(snapshot.devices[0].proxyUploadBytes, '1013');
+  assert.equal(snapshot.devices[0].proxyDownloadBytes, '2054');
+  assert.equal(snapshot.source.traffic.error, null);
+  assert.match(snapshot.source.traffic.proxy.error, /уменьшился/);
+
  trafficError = new Error('traffic unavailable');
  snapshot = await service.refresh();
-  assert.equal(snapshot.devices[0].uploadBytes, '9007199254741005');
+  assert.equal(snapshot.devices[0].uploadBytes, '9007199254741007');
  assert.equal(snapshot.source.traffic.lastObservedAt, observedAt);
  assert.equal(snapshot.source.traffic.error, 'traffic unavailable');
 });

+test('legacy dataplane samples preserve saved proxy totals while Gateway totals keep advancing', async (t) => {
+  const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-device-proxy-legacy-'));
+  t.after(() => fs.rmSync(directory, { recursive: true, force: true }));
+  const store = createJsonStore({
+    filePath: path.join(directory, 'devices.json'),
+    defaultValue: {},
+    migrate: migrateDeviceInventoryState,
+  });
+  const observedAt = '2026-08-07T12:00:00.000Z';
+  const mac = '00:11:22:33:44:55';
+  const neighbor = {
+    observedAt,
+    error: null,
+    observations: [{ ip: '192.168.50.7', mac, interface: 'eth0', observedAt, active: true }],
+  };
+  let row = {
+    mac,
+    uploadBytes: '10',
+    downloadBytes: '20',
+    proxyUploadBytes: '30',
+    proxyDownloadBytes: '40',
+  };
+  const service = createDeviceInventoryService({
+    store,
+    observe: () => neighbor,
+    observeTraffic: () => ({
+      epoch: 'epoch-a', generation: 'rules-a', observedAt, source: { error: null }, devices: [row],
+    }),
+  });
+
+  let snapshot = await service.refresh();
+  assert.equal(snapshot.devices[0].proxyUploadBytes, '30');
+  row = { mac, uploadBytes: '15', downloadBytes: '27' };
+  snapshot = await service.refresh();
+  assert.equal(snapshot.devices[0].uploadBytes, '15');
+  assert.equal(snapshot.devices[0].downloadBytes, '27');
+  assert.equal(snapshot.devices[0].proxyUploadBytes, '30');
+  assert.equal(snapshot.devices[0].proxyDownloadBytes, '40');
+  assert.equal(snapshot.source.traffic.proxy.error, null);
+});
+
+test('a proxy regression rejects every device in that proxy sample atomically', async (t) => {
+  const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-device-proxy-atomic-'));
+  t.after(() => fs.rmSync(directory, { recursive: true, force: true }));
+  const store = createJsonStore({
+    filePath: path.join(directory, 'devices.json'),
+    defaultValue: {},
+    migrate: migrateDeviceInventoryState,
+  });
+  const observedAt = '2026-08-07T12:00:00.000Z';
+  const macs = ['00:11:22:33:44:55', '00:11:22:33:44:66'];
+  const neighbor = {
+    observedAt,
+    error: null,
+    observations: macs.map((mac, index) => ({
+      ip: `192.168.50.${index + 7}`, mac, interface: 'eth0', observedAt, active: true,
+    })),
+  };
+  let rows = [
+    { mac: macs[0], uploadBytes: '10', downloadBytes: '20', proxyUploadBytes: '30', proxyDownloadBytes: '40' },
+    { mac: macs[1], uploadBytes: '50', downloadBytes: '60', proxyUploadBytes: '70', proxyDownloadBytes: '80' },
+  ];
+  const service = createDeviceInventoryService({
+    store,
+    observe: () => neighbor,
+    observeTraffic: () => ({
+      epoch: 'epoch-a', generation: 'rules-a', observedAt, source: { error: null }, devices: rows,
+    }),
+  });
+  await service.refresh();
+  const before = structuredClone(store.read().traffic.proxy);
+  rows = [
+    { mac: macs[0], uploadBytes: '11', downloadBytes: '22', proxyUploadBytes: '35', proxyDownloadBytes: '45' },
+    { mac: macs[1], uploadBytes: '52', downloadBytes: '63', proxyUploadBytes: '69', proxyDownloadBytes: '80' },
+  ];
+  const snapshot = await service.refresh();
+  const after = store.read().traffic.proxy;
+  assert.deepEqual(after.baselinesByMac, before.baselinesByMac);
+  assert.deepEqual(after.totalsByMac, before.totalsByMac);
+  assert.match(snapshot.source.traffic.proxy.error, /уменьшился/);
+  assert.equal(snapshot.devices.find(({ mac }) => mac === macs[0]).uploadBytes, '11');
+});
+
 test('pinned device policy persists, reconciles the full set, and keeps the last applied mode on failure', async (t) => {
  const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-device-policy-'));
  t.after(() => fs.rmSync(directory, { recursive: true, force: true }));
@@ -517,3 +642,57 @@ test('device inventory backs up malformed v2 traffic and re-baselines without do
  assert.equal(byMac.get(missingTotalMac).uploadBytes, '20');
  assert.equal(byMac.get(missingTotalMac).downloadBytes, '20');
 });
+
+test('malformed proxy totals are backed up and an expired recovery marker is cleared', async (t) => {
+  const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-device-corrupt-proxy-'));
+  t.after(() => fs.rmSync(directory, { recursive: true, force: true }));
+  const filePath = path.join(directory, 'devices.json');
+  const mac = '00:11:22:33:44:55';
+  fs.writeFileSync(filePath, JSON.stringify({
+    schemaVersion: 2,
+    revision: 1,
+    devices: [{
+      id: 'dev_0123456789abcdef',
+      alias: '',
+      pinned: false,
+      mac,
+      ip: '192.168.50.7',
+      interface: 'eth0',
+      firstSeenAt: '2026-06-01T12:00:00.000Z',
+      lastSeenAt: '2026-06-01T12:00:00.000Z',
+      source: 'neighbor',
+      confidence: 'high',
+    }],
+    traffic: {
+      baselinesByMac: {}, totalsByMac: {}, rebaselineMacs: [],
+      proxy: {
+        schemaVersion: 1,
+        baselinesByMac: { [mac]: { epoch: 'epoch-a', uploadBytes: '1', downloadBytes: '2' } },
+        totalsByMac: { [mac]: { uploadBytes: 'broken', downloadBytes: '4' } },
+        rebaselineMacs: [],
+      },
+    },
+  }));
+  const store = createJsonStore({
+    filePath,
+    defaultValue: {},
+    migrate: migrateDeviceInventoryState,
+    backupWhen: () => true,
+  });
+  assert.match(store.read().traffic.proxy.lastError, /proxy traffic checkpoint/);
+  assert.ok(fs.existsSync(store.migration.backupPath));
+
+  const service = createDeviceInventoryService({
+    store,
+    now: () => new Date('2026-08-07T12:00:00.000Z'),
+    observe: () => ({ observedAt: '2026-08-07T12:00:00.000Z', observations: [], error: null }),
+    observeTraffic: () => ({
+      epoch: 'epoch-a', generation: 'rules-a', observedAt: '2026-08-07T12:00:00.000Z',
+      source: { error: null }, devices: [],
+    }),
+  });
+  const snapshot = await service.refresh();
+  assert.equal(snapshot.devices.length, 0);
+  assert.deepEqual(store.read().traffic.proxy.rebaselineMacs, []);
+  assert.equal(snapshot.source.traffic.proxy.error, null);
+});