Add domain traffic metrics and Grafana dashboard

src/server/config.js

@@ -15,6 +15,7 @@ export const settings = {
   port: parsePort(process.env.PORT, 3456),
   proxyPort,
   diagnosticsProxyPort: parsePort(process.env.DIAGNOSTICS_PROXY_PORT, 18080),
+  singboxApiPort: parsePort(process.env.SING_BOX_API_PORT, 19090),
   tproxyPort: parsePort(process.env.TPROXY_PORT, 7895),
   tproxyMark: process.env.TPROXY_MARK || "1",
   tproxyChain: process.env.TPROXY_CHAIN || "VPN_PROXY_TPROXY",

src/server/dataplane.js

@@ -8,6 +8,10 @@ import { readNeighborSnapshot } from './adapters/neighbors.js';
 import { createDeviceTrafficService } from './services/deviceTrafficService.js';
 import { createDevicePolicyService } from './services/devicePolicyService.js';
 import { createConnectivityDiagnosticsService } from './services/connectivityDiagnosticsService.js';
+import {
+  createDomainTrafficService,
+  readSingboxConnections,
+} from './services/domainTrafficService.js';
 
 const socketPath = settings.dataplaneSocket;
 const runtime = createSingboxRuntime({
@@ -31,8 +35,13 @@ const devicePolicy = createDevicePolicyService({
 const connectivityDiagnostics = createConnectivityDiagnosticsService({
   proxyPort: settings.diagnosticsProxyPort,
 });
+const domainTraffic = createDomainTrafficService({
+  observe: () => readSingboxConnections(settings.singboxApiPort),
+  devices: () => traffic.snapshot().devices,
+});
 let ready = false;
 let trafficTimer = null;
+let domainTrafficTimer = null;
 const MAX_POLICY_BODY_BYTES = 256 * 1024;
 
 function readJson(req) {
@@ -83,6 +92,9 @@ const server = http.createServer(async (req, res) => {
     if (req.method === 'GET' && req.url === '/device-traffic') {
       return sendJson(res, 200, traffic.snapshot());
     }
+    if (req.method === 'GET' && req.url === '/domain-traffic') {
+      return sendJson(res, 200, domainTraffic.snapshot());
+    }
     if (req.method === 'GET' && req.url === '/device-policy') {
       return sendJson(res, 200, devicePolicy.snapshot());
     }
@@ -131,6 +143,16 @@ server.listen(socketPath, async () => {
       traffic.refresh().catch((error) => console.warn(`[dataplane] traffic counters не обновлены: ${error.message}`));
     }, 15_000);
     trafficTimer.unref();
+    setImmediate(() => {
+      domainTraffic.refresh()
+        .catch((error) => console.warn(`[dataplane] domain traffic не запущен: ${error.message}`));
+    });
+    // ponytail: snapshots can miss connections shorter than 2s; switch to an upstream close-event API if sing-box adds one.
+    domainTrafficTimer = setInterval(() => {
+      domainTraffic.refresh()
+        .catch((error) => console.warn(`[dataplane] domain traffic не обновлён: ${error.message}`));
+    }, 2_000);
+    domainTrafficTimer.unref();
     console.log(`[dataplane] control socket: ${socketPath}`);
   }
 });
@@ -141,6 +163,7 @@ async function shutdown() {
   shuttingDown = true;
   ready = false;
   if (trafficTimer) clearInterval(trafficTimer);
+  if (domainTrafficTimer) clearInterval(domainTrafficTimer);
   await runtime.shutdown();
   server.close(() => {
     fs.rmSync(socketPath, { force: true });

src/server/dataplaneClient.js

@@ -54,6 +54,7 @@ export function createDataplaneClient(socketPath, send = request) {
     refresh: () => update('/status', 'GET'),
     observeDevices: () => send(socketPath, '/devices', 'GET'),
     observeTraffic: () => send(socketPath, '/device-traffic', 'GET'),
+    observeDomainTraffic: () => send(socketPath, '/domain-traffic', 'GET'),
     observeDevicePolicy: () => send(socketPath, '/device-policy', 'GET'),
     applyDevicePolicies: (devices) => send(socketPath, '/device-policy', 'PUT', { devices }),
     runConnectivityDiagnostics: async (services = [], target = null) => {

src/server/index.js

@@ -127,6 +127,9 @@ const deviceInventory = settings.appMode === 'gateway'
       observeTraffic: remoteDataplane
         ? () => singboxRuntime.observeTraffic()
         : null,
+      observeDomainTraffic: remoteDataplane
+        ? () => singboxRuntime.observeDomainTraffic()
+        : null,
       observePolicy: remoteDataplane
         ? () => singboxRuntime.observeDevicePolicy()
         : () => localDevicePolicy.snapshot(),
@@ -864,7 +867,7 @@ const server = http.createServer(async (req, res) => {
     const requestUrl = new URL(req.url, `http://localhost:${settings.port}`);
     if (requestUrl.pathname === '/metrics') {
       if (!deviceInventory || req.method !== 'GET') throw new HarborError('ENDPOINT_NOT_FOUND');
-      return sendPrometheusMetrics(res, deviceInventory.snapshot());
+      return sendPrometheusMetrics(res, deviceInventory.metricsSnapshot());
     }
     return requestUrl.pathname.startsWith('/api/')
       ? await handleApi(req, res)

src/server/prometheusMetrics.js

@@ -99,6 +99,44 @@ export function renderPrometheusMetrics(snapshot) {
     }
   }
 
+  const domainTraffic = snapshot?.domainTraffic;
+  const domainSeries = Array.isArray(domainTraffic?.series) ? domainTraffic.series : [];
+  if (domainSeries.length) {
+    lines.push(
+      '# HELP harbor_device_domain_traffic_bytes_total Traffic observed by sing-box for a device and domain.',
+      '# TYPE harbor_device_domain_traffic_bytes_total counter',
+    );
+    for (const series of domainSeries) {
+      for (const [direction, value] of [
+        ['download', series.downloadBytes],
+        ['upload', series.uploadBytes],
+      ]) {
+        metric(lines, 'harbor_device_domain_traffic_bytes_total', {
+          device_id: series.deviceId,
+          domain: series.domain,
+          service: series.service,
+          source: series.source,
+          direction,
+        }, counter(value));
+      }
+    }
+  }
+  const domainObservedAt = timestamp(domainTraffic?.observedAt);
+  if (domainObservedAt) {
+    lines.push(
+      '# HELP harbor_domain_traffic_last_observed_timestamp_seconds Unix timestamp of the last successful sing-box connection observation.',
+      '# TYPE harbor_domain_traffic_last_observed_timestamp_seconds gauge',
+    );
+    lines.push(`harbor_domain_traffic_last_observed_timestamp_seconds ${domainObservedAt}`);
+  }
+  if (domainTraffic?.overflowConnections != null) {
+    lines.push(
+      '# HELP harbor_domain_traffic_overflow_connections_total Connections aggregated after the domain series limit was reached.',
+      '# TYPE harbor_domain_traffic_overflow_connections_total counter',
+    );
+    lines.push(`harbor_domain_traffic_overflow_connections_total ${counter(domainTraffic.overflowConnections)}`);
+  }
+
   return `${lines.join('\n')}\n`;
 }
 

src/server/services/deviceInventoryService.js

@@ -80,7 +80,7 @@ const DEFAULT_STATE = {
 };
 
 const normalizeMac = (value) => String(value || '').trim().toLowerCase();
-const deviceId = (mac) => `dev_${crypto.createHash('sha256').update(mac).digest('hex').slice(0, 16)}`;
+export const deviceId = (mac) => `dev_${crypto.createHash('sha256').update(mac).digest('hex').slice(0, 16)}`;
 const isPrivateMac = (mac) => (Number.parseInt(mac.slice(0, 2), 16) & 2) !== 0;
 const recordEntries = (value) => value && typeof value === 'object' && !Array.isArray(value)
   ? Object.entries(value)
@@ -414,6 +414,7 @@ export function createDeviceInventoryService({
   store,
   observe,
   observeTraffic = null,
+  observeDomainTraffic = null,
   observePolicy = null,
   applyPolicies = null,
   vendor = () => null,
@@ -425,6 +426,13 @@ export function createDeviceInventoryService({
   const trafficCursorByMac = new Map();
   let globalTrafficHistory = [];
   let globalTrafficCursor = null;
+  let domainTrafficSnapshot = {
+    epoch: null,
+    observedAt: null,
+    source: { error: null },
+    overflowConnections: '0',
+    series: [],
+  };
 
   function captureTrafficHistory(state) {
     const knownMacs = new Set(state.devices.map(({ mac }) => mac));
@@ -584,6 +592,10 @@ export function createDeviceInventoryService({
     };
   }
 
+  function metricsSnapshot() {
+    return { ...snapshot(), domainTraffic: domainTrafficSnapshot };
+  }
+
   function markPolicyEpoch(observed) {
     if (typeof observed?.epoch !== 'string' || !observed.epoch || !Array.isArray(observed.appliedIds)) return;
     store.update((stored) => {
@@ -695,7 +707,7 @@ export function createDeviceInventoryService({
   }
 
   async function performRefresh() {
-    const [result, trafficResult, policyResult] = await Promise.all([
+    const [result, trafficResult, policyResult, domainTrafficResult] = await Promise.all([
       Promise.resolve().then(() => observe()).catch((error) => ({
         observedAt: now().toISOString(),
         observations: [],
@@ -709,6 +721,10 @@ export function createDeviceInventoryService({
         ? Promise.resolve().then(() => observePolicy())
           .catch((error) => ({ transportError: error.message || String(error) }))
         : null,
+      observeDomainTraffic
+        ? Promise.resolve().then(() => observeDomainTraffic())
+          .catch((error) => ({ transportError: error.message || String(error) }))
+        : null,
     ]);
     const observedAt = result?.observedAt || now().toISOString();
     const observations = (Array.isArray(result?.observations) ? result.observations : [])
@@ -721,6 +737,14 @@ export function createDeviceInventoryService({
       identitiesByMac.get(mac).add(`${observation.ip}|${observation.interface || ''}`);
     }
     return serializePolicy(async () => {
+      if (domainTrafficResult?.transportError) {
+        domainTrafficSnapshot = {
+          ...domainTrafficSnapshot,
+          source: { error: domainTrafficResult.transportError },
+        };
+      } else if (domainTrafficResult) {
+        domainTrafficSnapshot = domainTrafficResult;
+      }
       const nextState = store.update((stored) => {
         const state = migrateDeviceInventoryState(stored);
         const byMac = new Map(state.devices.map((device) => [device.mac, device]));
@@ -1039,5 +1063,5 @@ export function createDeviceInventoryService({
     return serializePolicy(() => reconcileLocked(observed, true));
   }
 
-  return { snapshot, refresh, update, setPolicy, reconcilePolicies };
+  return { snapshot, metricsSnapshot, refresh, update, setPolicy, reconcilePolicies };
 }

src/server/services/domainTrafficService.js

@@ -0,0 +1,213 @@
+import crypto from 'node:crypto';
+import http from 'node:http';
+import net from 'node:net';
+import { domainToASCII } from 'node:url';
+import { deviceId } from './deviceInventoryService.js';
+
+const MAX_RESPONSE_BYTES = 4 * 1024 * 1024;
+const DEFAULT_MAX_SERIES = 4096;
+const SERVICE_DOMAINS = [
+  ['YouTube', ['youtube.com', 'youtube-nocookie.com', 'youtu.be', 'googlevideo.com', 'ytimg.com']],
+  ['OpenAI / ChatGPT', ['chatgpt.com', 'openai.com', 'oaistatic.com', 'oaiusercontent.com']],
+];
+
+const matchesDomain = (domain, suffix) => domain === suffix || domain.endsWith(`.${suffix}`);
+
+export function classifyDomain(value) {
+  let domain = domainToASCII(String(value || '').trim().replace(/\.$/, '')).toLowerCase();
+  if (domain.startsWith('www.')) domain = domain.slice(4);
+  const labels = domain.split('.');
+  if (!domain || domain.length > 253 || net.isIP(domain) || labels.length < 2
+    || labels.some((label) => !/^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$/.test(label))) return null;
+  for (const [service, suffixes] of SERVICE_DOMAINS) {
+    const suffix = suffixes.find((candidate) => matchesDomain(domain, candidate));
+    if (suffix) return { domain: suffix, service };
+  }
+  return { domain, service: domain };
+}
+
+function sourceFor(type) {
+  if (type === 'tproxy/tproxy-in') return 'gateway';
+  if (type === 'mixed/mixed-in') return 'proxy';
+  return null;
+}
+
+function parseConnection(connection, devicesByIp) {
+  const id = String(connection?.id || '');
+  const metadata = connection?.metadata;
+  const source = sourceFor(String(metadata?.type || ''));
+  const domain = classifyDomain(metadata?.host);
+  const currentDeviceId = devicesByIp.get(String(metadata?.sourceIP || ''));
+  const upload = connection?.upload;
+  const download = connection?.download;
+  if (!id || !Number.isSafeInteger(upload) || upload < 0
+    || !Number.isSafeInteger(download) || download < 0) {
+    throw new Error('Sing-box вернул невалидный domain traffic counter');
+  }
+  if (!source || !domain || !currentDeviceId) return null;
+  return {
+    id,
+    deviceId: currentDeviceId,
+    ...domain,
+    source,
+    upload: BigInt(upload),
+    download: BigInt(download),
+  };
+}
+
+export function readSingboxConnections(port, timeoutMs = 1500) {
+  return new Promise((resolve, reject) => {
+    const request = http.get({ host: '127.0.0.1', port, path: '/connections' }, (response) => {
+      const chunks = [];
+      let size = 0;
+      let tooLarge = false;
+      response.on('data', (chunk) => {
+        if (tooLarge) return;
+        size += chunk.length;
+        if (size > MAX_RESPONSE_BYTES) {
+          tooLarge = true;
+          request.destroy(new Error('Sing-box connections response слишком большой'));
+          return;
+        }
+        chunks.push(chunk);
+      });
+      response.on('end', () => {
+        if (tooLarge) return;
+        if ((response.statusCode || 500) >= 400) {
+          reject(new Error(`Sing-box connections HTTP ${response.statusCode}`));
+          return;
+        }
+        try {
+          resolve(JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'));
+        } catch (cause) {
+          reject(new Error('Sing-box вернул невалидный connections JSON', { cause }));
+        }
+      });
+    });
+    request.setTimeout(timeoutMs, () => request.destroy(new Error('Sing-box connections timeout')));
+    request.on('error', reject);
+  });
+}
+
+export function createDomainTrafficService({
+  observe,
+  devices,
+  now = () => new Date(),
+  maxSeries = DEFAULT_MAX_SERIES,
+}) {
+  if (!Number.isInteger(maxSeries) || maxSeries < 2) throw new Error('Domain traffic series limit должен быть не меньше 2');
+  const epoch = crypto.randomUUID();
+  const totals = new Map();
+  const normalSeriesLimit = maxSeries - 2;
+  let normalSeries = 0;
+  let previousConnections = new Map();
+  let overflowConnections = 0n;
+  let refreshPromise = null;
+  let current = {
+    epoch,
+    observedAt: null,
+    source: { error: null },
+    overflowConnections: '0',
+    series: [],
+  };
+
+  function buildSnapshot(error = null) {
+    return {
+      epoch,
+      observedAt: current.observedAt,
+      source: { error },
+      overflowConnections: overflowConnections.toString(),
+      series: [...totals.values()]
+        .map((entry) => ({
+          ...entry,
+          uploadBytes: entry.uploadBytes.toString(),
+          downloadBytes: entry.downloadBytes.toString(),
+        }))
+        .sort((left, right) => (
+          left.deviceId.localeCompare(right.deviceId)
+          || left.service.localeCompare(right.service)
+          || left.domain.localeCompare(right.domain)
+          || left.source.localeCompare(right.source)
+        )),
+    };
+  }
+
+  async function performRefresh() {
+    try {
+      const response = await observe();
+      if (!Array.isArray(response?.connections)) throw new Error('Sing-box не вернул connections array');
+      const devicesByIp = new Map();
+      const observedDevices = devices();
+      for (const device of Array.isArray(observedDevices) ? observedDevices : []) {
+        const ip = String(device?.ip || '');
+        const id = typeof device?.mac === 'string' ? deviceId(device.mac.toLowerCase()) : null;
+        if (!net.isIPv4(ip) || !id) continue;
+        devicesByIp.set(ip, devicesByIp.has(ip) ? null : id);
+      }
+      const parsed = response.connections
+        .map((connection) => parseConnection(connection, devicesByIp))
+        .filter(Boolean);
+      const activeConnections = new Map();
+      for (const connection of parsed) {
+        const requestedKey = `${connection.deviceId}\0${connection.domain}\0${connection.source}`;
+        const previous = previousConnections.get(connection.id);
+        let key = previous?.requestedKey === requestedKey ? previous.key : requestedKey;
+        let domain = connection.domain;
+        let service = connection.service;
+        if (key !== requestedKey) {
+          domain = '_other';
+          service = 'Другие домены';
+        } else if (!totals.has(key) && normalSeries >= normalSeriesLimit) {
+          overflowConnections += 1n;
+          domain = '_other';
+          service = 'Другие домены';
+          key = `_other\0${domain}\0${connection.source}`;
+        } else if (!totals.has(key)) {
+          normalSeries += 1;
+        }
+        const sameSeries = previous?.key === key;
+        const uploadDelta = sameSeries && connection.upload >= previous.upload
+          ? connection.upload - previous.upload
+          : connection.upload;
+        const downloadDelta = sameSeries && connection.download >= previous.download
+          ? connection.download - previous.download
+          : connection.download;
+        const total = totals.get(key) || {
+          deviceId: key === requestedKey ? connection.deviceId : '_other',
+          domain,
+          service,
+          source: connection.source,
+          uploadBytes: 0n,
+          downloadBytes: 0n,
+        };
+        total.uploadBytes += uploadDelta;
+        total.downloadBytes += downloadDelta;
+        totals.set(key, total);
+        activeConnections.set(connection.id, {
+          key,
+          requestedKey,
+          upload: connection.upload,
+          download: connection.download,
+        });
+      }
+      previousConnections = activeConnections;
+      current = { ...current, observedAt: now().toISOString() };
+      current = buildSnapshot();
+      return current;
+    } catch (error) {
+      current = buildSnapshot(error.message || String(error));
+      throw error;
+    }
+  }
+
+  function refresh() {
+    if (!refreshPromise) {
+      refreshPromise = performRefresh().finally(() => {
+        refreshPromise = null;
+      });
+    }
+    return refreshPromise;
+  }
+
+  return { snapshot: () => current, refresh };
+}

src/server/singbox.js

@@ -79,6 +79,9 @@ export function buildGatewayConfig(subscriptionConfig, selectedTag, {
     log: { level: settings.logLevel, timestamp: true },
     experimental: {
       cache_file: { enabled: true, path: settings.cachePath },
+      ...(!clientMode ? {
+        clash_api: { external_controller: `127.0.0.1:${settings.singboxApiPort}` },
+      } : {}),
     },
     dns: { independent_cache: true },
     inbounds,