diff --git a/app.js b/app.js index 9990bfd..93c5509 100644 --- a/app.js +++ b/app.js @@ -125,6 +125,8 @@ manager.on('disconnect', (conn) => { } if (!manager.active?.connected) { updateHealthBadge(null); + lastStatsInterestSent = null; + lastStatsInterestPeerId = null; } refreshFleetIfVisible(); if (isBootRestoring) return; @@ -133,6 +135,8 @@ manager.on('disconnect', (conn) => { if (!hasActiveConnection()) { resetContainerList(); stopStatsInterval(); + lastStatsInterestSent = null; + lastStatsInterestPeerId = null; if (typeof manager.isReconnecting === 'function' && manager.isReconnecting()) { // reconnecting banner is driven by 'reconnecting' events return; @@ -160,6 +164,26 @@ manager.on('connect', (conn) => { { latency: conn.latency, docker: conn.dockerHealth, status: conn.healthStatus }, conn ); + // Re-declare view interest after (re)connect so stats streams resume if needed + lastStatsInterestSent = null; + lastStatsInterestPeerId = null; + try { + syncStatsInterest({ force: true }); + } catch { + // ignore + } + } +}); +manager.on('active', (conn) => { + // Switching fleet peer: re-send interest to the new active host + lastStatsInterestSent = null; + lastStatsInterestPeerId = null; + if (conn?.connected) { + try { + syncStatsInterest({ force: true }); + } catch { + // ignore + } } }); /** Throttle reconnect notices (standard mode only) */ @@ -419,6 +443,66 @@ function startStatsInterval() { stopStatsInterval(); } +/** + * Views that show live per-container CPU/memory (table or dashboard KPIs). + * Server only opens Docker stats streams while ≥1 peer reports interest. + */ +const STATS_INTEREST_VIEWS = new Set(['containers', 'dashboard', 'container-details']); + +/** @type {boolean|null} last interest sent for active peer (avoid spam) */ +let lastStatsInterestSent = null; +/** @type {string|null} peer id we last told */ +let lastStatsInterestPeerId = null; + +/** + * Tell the active peer whether this UI needs live fleet stats. + * Call on view change and after connect. Leaving Containers/Dashboard tears + * down server-side Docker stats streams for this client. + * @param {{ force?: boolean }} [opts] + */ +function syncStatsInterest(opts = {}) { + const force = opts.force === true; + const conn = manager.active; + if (!conn?.connected) { + lastStatsInterestSent = null; + lastStatsInterestPeerId = null; + return; + } + const view = + typeof currentView === 'string' + ? currentView + : typeof window !== 'undefined' && typeof window.currentView === 'string' + ? window.currentView + : ''; + const want = STATS_INTEREST_VIEWS.has(view); + if ( + !force && + lastStatsInterestSent === want && + lastStatsInterestPeerId === conn.id + ) { + return; + } + lastStatsInterestSent = want; + lastStatsInterestPeerId = conn.id; + try { + conn + .request(Methods.setStatsInterest || 'setStatsInterest', { + active: want, + view: view || undefined, + }) + .catch(() => { + // older servers may lack the method — ignore + lastStatsInterestSent = null; + }); + } catch { + lastStatsInterestSent = null; + } +} + +if (typeof window !== 'undefined') { + window.syncStatsInterest = syncStatsInterest; +} + // Utility functions are now imported from uiUtils.js @@ -1384,6 +1468,13 @@ function navigateToView(viewName, opts = {}) { // Expose for auto-refresh poller if (typeof window !== 'undefined') window.currentView = viewName; + // Demand-driven fleet stats: only collect while on Containers / Dashboard / details + try { + syncStatsInterest(); + } catch { + // ignore + } + // Drop live streams / PTYs when leaving container details if (leavingDetails) { if (typeof stopDetailsLogs === 'function') stopDetailsLogs(); diff --git a/client/api.js b/client/api.js index 65db1af..97406d5 100644 --- a/client/api.js +++ b/client/api.js @@ -385,6 +385,13 @@ export const api = { return connOrActive(connection).request(Methods.updateStatsConfig, args) }, + setStatsInterest(active, view, connection) { + return connOrActive(connection).request(Methods.setStatsInterest, { + active: Boolean(active), + view: view || undefined, + }) + }, + getHostSnapshot(connection) { return connOrActive(connection).request(Methods.getHostSnapshot, {}) }, diff --git a/config.js b/config.js index 46c6732..20dccf5 100644 --- a/config.js +++ b/config.js @@ -4,17 +4,17 @@ export const CONFIG = { // Stats collection (server/services/stats.js) - // Round-robin one-shot samples — not full-fleet stream:true (Dozzle-style streams - // keep dockerd busy ~1Hz × N containers). See Settings → Performance. + // Demand-driven: live stream:true only while a client is on Containers / + // Dashboard / container-details. Fully off otherwise. See Settings → Performance. STATS: { - INTERVAL_MS: 5000, // Min gap between allStats broadcasts - ACTIVE_INTERVAL_MS: 5000, // Idle gap after each sample batch completes - IDLE_INTERVAL_MS: 5000, // Reserved (peer-idle uses full pause) - CACHE_TTL_MS: 2000, // Per-container broadcast cache TTL - CONCURRENCY: 2, // Max parallel one-shot stats to dockerd - SAMPLES_PER_TICK: 4, // Max containers sampled per tick (round-robin) - LIST_INTERVAL_MS: 15000, // How often to re-list containers - SMOOTHING_FACTOR: 0.2, // Client-side smoothing (if used) + INTERVAL_MS: 1000, // allStats push cadence while watching + ACTIVE_INTERVAL_MS: 1000, // alias for broadcast when watching + IDLE_INTERVAL_MS: 5000, // reserved + CACHE_TTL_MS: 1000, + CONCURRENCY: 2, // legacy (stream mode) + SAMPLES_PER_TICK: 4, // legacy (stream mode) + LIST_INTERVAL_MS: 10000, // re-list running containers while watching + SMOOTHING_FACTOR: 0.2, }, // UI updates diff --git a/docs/SERVER.md b/docs/SERVER.md index 9db24ef..a7c6dc0 100644 --- a/docs/SERVER.md +++ b/docs/SERVER.md @@ -105,22 +105,14 @@ Subscribes to Engine events; fans out `push:dockerEvent` and may trigger list re Periodic container stats → `push:allStats`; history buffer for charts. -**Important:** peardock does **not** open perpetual per-container `stream: true` stats (Dozzle does, and [documents elevated dockerd CPU](https://dozzle.dev/guide/faq) as expected). Instead: - -1. **Round-robin one-shot** samples (`stream: false`) — at most `samplesPerTick` containers per batch -2. **Post-batch idle gap** — next batch waits `collectIntervalMs` after the previous finishes (no overlapping sweeps) -3. **Roster throttle** — `listContainers` only every `listIntervalMs` -4. **Peer-idle pause** — no sampling when `peers.size === 0` +**Demand-driven real-time stats:** clients call `setStatsInterest({ active, view })` while on **Containers**, **Dashboard**, or **container details**. Only then does the server open per-container Docker `stats({ stream: true })` attachments and push `push:allStats` to **watchers** (not every connected peer). Leaving those views (or disconnecting) tears streams down immediately — dockerd is not sampled in the background. | Env / UI | Default | Meaning | |----------|---------|---------| -| Collect gap / `PEARDOCK_STATS_INTERVAL_MS` | 5000 | Idle after each sample batch | -| Samples/tick / `PEARDOCK_STATS_SAMPLES_PER_TICK` | 4 | Max containers sampled per batch (main load knob) | -| Concurrency / `PEARDOCK_STATS_CONCURRENCY` | 2 | Parallel one-shots within a batch | -| List interval / `PEARDOCK_STATS_LIST_INTERVAL_MS` | 15000 | Roster refresh | -| Broadcast / `PEARDOCK_STATS_BROADCAST_MS` | 5000 | Min gap between `push:allStats` | +| UI update interval / `PEARDOCK_STATS_BROADCAST_MS` | 1000 | `push:allStats` cadence while watching | +| Roster refresh / `PEARDOCK_STATS_LIST_INTERVAL_MS` | 10000 | Re-list running containers + attach streams | -Runtime values persist in `peardock-stats.json` (cwd or `PEARDOCK_STATS_PATH`). RPC: `getStatsConfig` (viewer), `updateStatsConfig` (admin). Settings → **Performance** applies live. On hot hosts: raise collect gap, keep samples/tick at 2–4 and concurrency at 1–2. +RPC: `setStatsInterest` (viewer), `getStatsConfig` (viewer), `updateStatsConfig` (admin). Settings → **Performance**. ### Metrics (`services/metrics.js`) diff --git a/index.html b/index.html index 35c1677..93f6528 100644 --- a/index.html +++ b/index.html @@ -2888,9 +2888,10 @@ services:

Docker stats collection Server

- Controls how the connected peer samples container CPU/memory for the fleet UI. - Uses short one-shot polls (not perpetual streams) so dockerd load stays bounded. - Requires an admin session to change. Values persist on the server as peardock-stats.json. + Live CPU/memory for the Containers table uses real-time Docker stats streams, but only while a client is on + Containers, Dashboard, or container details. + Leaving those views stops collection immediately so dockerd is not sampled in the background. + Requires admin to change timings. Persists as peardock-stats.json.

@@ -2903,44 +2904,22 @@ services: -

Changes apply live on the server — no restart and no Save button.

+

Changes apply live — no restart. Collection is still off when no client is watching stats views.

- - -
Wait this long after a sample batch finishes before the next. Higher = less dockerd CPU.
+ + +
How often push:allStats refreshes the table while watching (1 = near real-time).
- - -
Containers sampled each tick (round-robin). Main load control — do not set to fleet size.
-
-
- - -
Max parallel one-shot stats to dockerd within a tick. Keep at 1–2 on busy hosts.
-
-
- - -
Min gap between push:allStats to clients (can reuse last samples).
-
-
-
-
- - -
Roster refresh rate (start/stop detection for the sampler).
-
-
- - -
Reuse last sample for idle containers when broadcasting.
+ + +
How often to re-list running containers and attach/detach streams.
-
One-time server log when this many containers are running.
+
One-time server log when this many streams would open.
+ + + + +
diff --git a/server/handlers/system.js b/server/handlers/system.js index c08d7f8..944bb73 100644 --- a/server/handlers/system.js +++ b/server/handlers/system.js @@ -18,6 +18,7 @@ import { import { getStatsConfig, updateStatsConfig, + setPeerStatsInterest, getStatsFilePath, STATS_LIMITS, } from '../services/stats.js' @@ -210,6 +211,21 @@ export function registerSystemHandlers(session) { return { success: true, config } }) + /** + * Client declares whether it needs live fleet stats for the current UI view. + * Server only opens Docker stats streams while ≥1 peer has active interest. + */ + session.respond( + 'setStatsInterest', + async (args = {}) => { + const active = args.active === true || args.active === 1 || args.active === '1' + const view = args.view != null ? String(args.view) : undefined + const config = setPeerStatsInterest(session.id, active, { view }) + return { success: true, active, config } + }, + { hot: true } + ) + session.respond('listSchedules', async () => { return { success: true, schedules: listSchedules() } }) diff --git a/server/rpc/register.js b/server/rpc/register.js index c9c5f2f..dc181d6 100644 --- a/server/rpc/register.js +++ b/server/rpc/register.js @@ -21,6 +21,7 @@ import { registerBinaryStreamHandlers } from './binary-stream.js' import { registerSuggestionHandlers } from '../handlers/suggestions.js' import { registerTunnelHandlers } from '../handlers/tunnels.js' import { registerAlertHandlers } from '../handlers/alerts.js' +import { clearPeerStatsInterest } from '../services/stats.js' /** * @param {import('./session.js').PeerSession} session @@ -54,6 +55,11 @@ export function registerAllHandlers(session) { export function cleanupSession(session) { cleanupTerminalOnClose(session) cleanupLogsOnClose(session) + try { + clearPeerStatsInterest(session.id) + } catch { + // ignore + } for (const [key, value] of session.state.entries()) { if (key.startsWith('exec:') && value?.stream) { try { diff --git a/server/services/stats.js b/server/services/stats.js index 1c3ef3c..4c346c9 100644 --- a/server/services/stats.js +++ b/server/services/stats.js @@ -1,18 +1,17 @@ /** * Container stats collection and broadcast. * - * Strategy (tuned against dockerd load, informed by Dozzle's approach): + * Demand-driven real-time fleet stats: * - * - **One-shot** `stats({ stream: false })` — no perpetual stream:true attachments. - * Dozzle uses long-lived streams for all running containers and documents - * elevated dockerd CPU as expected; we intentionally avoid that. - * - **Round-robin batches** — each tick samples at most `samplesPerTick` - * containers (not the entire fleet). Full-fleet one-shot sweeps are often - * *worse* than streams when sweep time > interval (dockerd never idles). - * - **Post-sweep spacing** — next tick is scheduled only after the previous - * finishes + collectIntervalMs (not a fixed setInterval that overlaps work). - * - **Roster cache** — listContainers is not called every tick. - * - **Peer-idle pause** — no collection when peers.size === 0. + * - Clients call `setStatsInterest({ active: true })` while on Containers / + * Dashboard / container-details (views that show CPU/mem). + * - Only then do we open per-container Docker stats streams and push + * `push:allStats` to interested peers. + * - When no peer wants stats, all streams are torn down immediately — + * dockerd is not sampled in the background. + * + * Streams (not round-robin one-shots) so the containers table updates live. + * Config knobs (Settings → Performance) still tune broadcast/list intervals. */ import fs from 'fs' import path from 'path' @@ -26,15 +25,18 @@ import logger from '../utils/logger.js' const STATS_FILE = process.env.PEARDOCK_STATS_PATH || path.join(process.cwd(), 'peardock-stats.json') -/** Clamp helpers for RPC / env / disk */ export const STATS_LIMITS = Object.freeze({ - collectIntervalMs: { min: 1000, max: 60_000 }, - broadcastIntervalMs: { min: 1000, max: 60_000 }, + /** How often we fan-out push:allStats while someone is watching */ + broadcastIntervalMs: { min: 500, max: 30_000 }, + /** How often to re-list running containers while watching */ + listIntervalMs: { min: 2000, max: 120_000 }, + /** Soft warn when many concurrent streams */ + warnContainers: { min: 10, max: 10_000 }, + /** Legacy keys kept for config normalize / UI compatibility */ + collectIntervalMs: { min: 500, max: 60_000 }, concurrency: { min: 1, max: 32 }, samplesPerTick: { min: 1, max: 100 }, - listIntervalMs: { min: 2000, max: 120_000 }, cacheTtlMs: { min: 0, max: 30_000 }, - warnContainers: { min: 10, max: 10_000 }, }) function envInt(name, fallback) { @@ -49,57 +51,24 @@ function clampInt(value, min, max, fallback) { } /** - * Built-in defaults, then env overrides (bootstrap before disk / UI). - * Conservative defaults keep dockerd idle between batches on busy hosts. - * @returns {{ - * collectIntervalMs: number, - * broadcastIntervalMs: number, - * concurrency: number, - * samplesPerTick: number, - * listIntervalMs: number, - * cacheTtlMs: number, - * warnContainers: number, - * }} + * Defaults tuned for real-time table updates while a peer is watching. + * Collection is fully off when no peer has stats interest. */ export function defaultStatsConfig() { - const collectIntervalMs = clampInt( - envInt( - 'PEARDOCK_STATS_INTERVAL_MS', - CONFIG.STATS?.ACTIVE_INTERVAL_MS ?? CONFIG.STATS?.INTERVAL_MS ?? 5000 - ), - STATS_LIMITS.collectIntervalMs.min, - STATS_LIMITS.collectIntervalMs.max, - 5000 - ) const broadcastIntervalMs = clampInt( - envInt('PEARDOCK_STATS_BROADCAST_MS', CONFIG.STATS?.INTERVAL_MS ?? 5000), + envInt( + 'PEARDOCK_STATS_BROADCAST_MS', + CONFIG.STATS?.INTERVAL_MS ?? CONFIG.STATS?.ACTIVE_INTERVAL_MS ?? 1000 + ), STATS_LIMITS.broadcastIntervalMs.min, STATS_LIMITS.broadcastIntervalMs.max, - 5000 - ) - const concurrency = clampInt( - envInt('PEARDOCK_STATS_CONCURRENCY', CONFIG.STATS?.CONCURRENCY ?? 2), - STATS_LIMITS.concurrency.min, - STATS_LIMITS.concurrency.max, - 2 - ) - const samplesPerTick = clampInt( - envInt('PEARDOCK_STATS_SAMPLES_PER_TICK', CONFIG.STATS?.SAMPLES_PER_TICK ?? 4), - STATS_LIMITS.samplesPerTick.min, - STATS_LIMITS.samplesPerTick.max, - 4 + 1000 ) const listIntervalMs = clampInt( - envInt('PEARDOCK_STATS_LIST_INTERVAL_MS', CONFIG.STATS?.LIST_INTERVAL_MS ?? 15_000), + envInt('PEARDOCK_STATS_LIST_INTERVAL_MS', CONFIG.STATS?.LIST_INTERVAL_MS ?? 10_000), STATS_LIMITS.listIntervalMs.min, STATS_LIMITS.listIntervalMs.max, - 15_000 - ) - const cacheTtlMs = clampInt( - envInt('PEARDOCK_STATS_CACHE_TTL_MS', CONFIG.STATS?.CACHE_TTL_MS ?? 2000), - STATS_LIMITS.cacheTtlMs.min, - STATS_LIMITS.cacheTtlMs.max, - 2000 + 10_000 ) const warnContainers = clampInt( envInt('PEARDOCK_STATS_WARN_CONTAINERS', 80), @@ -107,14 +76,21 @@ export function defaultStatsConfig() { STATS_LIMITS.warnContainers.max, 80 ) + // Legacy fields (UI / older peardock-stats.json) — unused in stream mode + const collectIntervalMs = clampInt( + envInt('PEARDOCK_STATS_INTERVAL_MS', broadcastIntervalMs), + STATS_LIMITS.collectIntervalMs.min, + STATS_LIMITS.collectIntervalMs.max, + broadcastIntervalMs + ) return { - collectIntervalMs, broadcastIntervalMs, - concurrency, - samplesPerTick, listIntervalMs, - cacheTtlMs, warnContainers, + collectIntervalMs, + concurrency: 2, + samplesPerTick: 4, + cacheTtlMs: 1000, } } @@ -137,15 +113,17 @@ function loadStatsDisk() { function saveStatsDisk() { try { const payload = { - version: 1, + version: 2, + mode: 'interest-stream', updatedAt: new Date().toISOString(), - collectIntervalMs: runtime.collectIntervalMs, broadcastIntervalMs: runtime.broadcastIntervalMs, + listIntervalMs: runtime.listIntervalMs, + warnContainers: runtime.warnContainers, + // Keep legacy keys so older UIs still round-trip + collectIntervalMs: runtime.collectIntervalMs, concurrency: runtime.concurrency, samplesPerTick: runtime.samplesPerTick, - listIntervalMs: runtime.listIntervalMs, cacheTtlMs: runtime.cacheTtlMs, - warnContainers: runtime.warnContainers, } const tmp = `${STATS_FILE}.${process.pid}.tmp` fs.writeFileSync(tmp, JSON.stringify(payload, null, 2), { mode: 0o600 }) @@ -156,37 +134,51 @@ function saveStatsDisk() { } /** - * Normalize a partial config object with clamps. * @param {object} partial * @param {ReturnType} [base] */ export function normalizeStatsConfig(partial = {}, base = defaultStatsConfig()) { const src = partial && typeof partial === 'object' ? partial : {} - // Accept seconds from UI as collectIntervalSec / broadcastIntervalSec / listIntervalSec - let collectMs = src.collectIntervalMs - if (collectMs == null && src.collectIntervalSec != null) { - collectMs = Number(src.collectIntervalSec) * 1000 - } let broadcastMs = src.broadcastIntervalMs if (broadcastMs == null && src.broadcastIntervalSec != null) { broadcastMs = Number(src.broadcastIntervalSec) * 1000 } + // Older UI "collect interval" maps to broadcast cadence in stream mode + if (broadcastMs == null && src.collectIntervalMs != null) { + broadcastMs = src.collectIntervalMs + } + if (broadcastMs == null && src.collectIntervalSec != null) { + broadcastMs = Number(src.collectIntervalSec) * 1000 + } let listMs = src.listIntervalMs if (listMs == null && src.listIntervalSec != null) { listMs = Number(src.listIntervalSec) * 1000 } + const broadcastIntervalMs = clampInt( + broadcastMs ?? base.broadcastIntervalMs, + STATS_LIMITS.broadcastIntervalMs.min, + STATS_LIMITS.broadcastIntervalMs.max, + base.broadcastIntervalMs + ) return { + broadcastIntervalMs, + listIntervalMs: clampInt( + listMs ?? base.listIntervalMs, + STATS_LIMITS.listIntervalMs.min, + STATS_LIMITS.listIntervalMs.max, + base.listIntervalMs + ), + warnContainers: clampInt( + src.warnContainers ?? base.warnContainers, + STATS_LIMITS.warnContainers.min, + STATS_LIMITS.warnContainers.max, + base.warnContainers + ), collectIntervalMs: clampInt( - collectMs ?? base.collectIntervalMs, + src.collectIntervalMs ?? broadcastIntervalMs, STATS_LIMITS.collectIntervalMs.min, STATS_LIMITS.collectIntervalMs.max, - base.collectIntervalMs - ), - broadcastIntervalMs: clampInt( - broadcastMs ?? base.broadcastIntervalMs, - STATS_LIMITS.broadcastIntervalMs.min, - STATS_LIMITS.broadcastIntervalMs.max, - base.broadcastIntervalMs + broadcastIntervalMs ), concurrency: clampInt( src.concurrency ?? base.concurrency, @@ -200,30 +192,15 @@ export function normalizeStatsConfig(partial = {}, base = defaultStatsConfig()) STATS_LIMITS.samplesPerTick.max, base.samplesPerTick ), - listIntervalMs: clampInt( - listMs ?? base.listIntervalMs, - STATS_LIMITS.listIntervalMs.min, - STATS_LIMITS.listIntervalMs.max, - base.listIntervalMs - ), cacheTtlMs: clampInt( src.cacheTtlMs ?? base.cacheTtlMs, STATS_LIMITS.cacheTtlMs.min, STATS_LIMITS.cacheTtlMs.max, base.cacheTtlMs ), - warnContainers: clampInt( - src.warnContainers ?? base.warnContainers, - STATS_LIMITS.warnContainers.min, - STATS_LIMITS.warnContainers.max, - base.warnContainers - ), } } -/** - * Ensure disk overlay is applied once (env defaults first, then peardock-stats.json). - */ export function ensureStatsConfigLoaded() { if (diskLoaded) return runtime diskLoaded = true @@ -233,59 +210,123 @@ export function ensureStatsConfigLoaded() { runtime = normalizeStatsConfig(disk, runtime) logger.debug('Stats config loaded from disk', { path: STATS_FILE, - collectIntervalMs: runtime.collectIntervalMs, - concurrency: runtime.concurrency, + broadcastIntervalMs: runtime.broadcastIntervalMs, }) } return runtime } +// ─── Interest (view-driven) ─────────────────────────────────────────────────── + +/** Peer ids that currently want live fleet stats */ +const interestedPeers = new Set() +/** @type {Map} peerId → view name (debug) */ +const peerViews = new Map() + /** - * Public config snapshot for RPC / Settings UI. - * @returns {object} + * Client reports whether it needs live container CPU/mem (Containers list, etc.). + * @param {string} peerId + * @param {boolean} active + * @param {{ view?: string }} [meta] */ +export function setPeerStatsInterest(peerId, active, meta = {}) { + if (!peerId) return getStatsConfig() + const was = interestedPeers.size + if (active) { + interestedPeers.add(peerId) + if (meta.view) peerViews.set(peerId, String(meta.view)) + } else { + interestedPeers.delete(peerId) + peerViews.delete(peerId) + } + const now = interestedPeers.size + if (was === 0 && now > 0) { + resumeStatsCollection() + } else if (was > 0 && now === 0) { + pauseStatsCollection() + } + logger.debug('Stats interest', { + peerId: peerId.slice(0, 12), + active: Boolean(active), + view: meta.view || null, + watchers: now, + }) + return getStatsConfig() +} + +/** + * Drop interest for a peer (session close). + * @param {string} peerId + */ +export function clearPeerStatsInterest(peerId) { + if (!peerId || !interestedPeers.has(peerId)) return + setPeerStatsInterest(peerId, false) +} + +export function getStatsInterestCount() { + return interestedPeers.size +} + +// ─── Live state ─────────────────────────────────────────────────────────────── + +/** @type {Record} */ +const containerStats = {} +const statsCache = new Map() +const containerActivity = new Map() + +/** @type {ReturnType|null} */ +let broadcastTimer = null +/** @type {ReturnType|null} */ +let rosterTimer = null +let collectionActive = false +let unsubPeers = null +let serviceArmed = false +let lastBroadcast = 0 +let dockerDownLogged = false +let dockerBackoffUntil = 0 +let rosterInFlight = false +let largeFleetWarned = false + export function getStatsConfig() { ensureStatsConfigLoaded() return { - collectIntervalMs: runtime.collectIntervalMs, + mode: 'interest-stream', broadcastIntervalMs: runtime.broadcastIntervalMs, + listIntervalMs: runtime.listIntervalMs, + warnContainers: runtime.warnContainers, + collectIntervalMs: runtime.collectIntervalMs, concurrency: runtime.concurrency, samplesPerTick: runtime.samplesPerTick, - listIntervalMs: runtime.listIntervalMs, cacheTtlMs: runtime.cacheTtlMs, - warnContainers: runtime.warnContainers, - // Convenience for UI collectIntervalSec: Math.round(runtime.collectIntervalMs / 1000), broadcastIntervalSec: Math.round(runtime.broadcastIntervalMs / 1000), listIntervalSec: Math.round(runtime.listIntervalMs / 1000), - active: collectionLoopActive, + active: collectionActive, peers: peers.size, - runningTracked: rosterRunning.length, + watchers: interestedPeers.size, + streams: Object.values(containerStats).filter((s) => s.stream).length, + runningTracked: Object.keys(containerStats).length, limits: { ...STATS_LIMITS }, path: STATS_FILE, } } -/** - * Apply partial config live (no process restart). Optionally persist to disk. - * @param {object} partial - * @param {{ persist?: boolean, replace?: boolean }} [opts] - * @returns {object} getStatsConfig() - */ export function updateStatsConfig(partial = {}, opts = {}) { ensureStatsConfigLoaded() const base = opts.replace ? defaultStatsConfig() : { ...runtime } const next = normalizeStatsConfig(partial, base) + const broadcastChanged = next.broadcastIntervalMs !== runtime.broadcastIntervalMs + const listChanged = next.listIntervalMs !== runtime.listIntervalMs runtime = next if (opts.persist !== false) saveStatsDisk() - // Interval is applied on the next schedule after a tick completes + if (collectionActive) { + if (broadcastChanged) armBroadcastLoop() + if (listChanged) armRosterLoop() + } logger.info('Stats config updated', { - collectIntervalMs: runtime.collectIntervalMs, broadcastIntervalMs: runtime.broadcastIntervalMs, - concurrency: runtime.concurrency, - samplesPerTick: runtime.samplesPerTick, listIntervalMs: runtime.listIntervalMs, - cacheTtlMs: runtime.cacheTtlMs, + watchers: interestedPeers.size, persist: opts.persist !== false, }) return getStatsConfig() @@ -295,38 +336,8 @@ export function getStatsFilePath() { return STATS_FILE } -/** @type {Record} */ -const containerStats = {} -const statsCache = new Map() -const containerActivity = new Map() +// ─── CPU / mem helpers ──────────────────────────────────────────────────────── -/** @type {ReturnType|null} */ -let loopTimer = null -/** True while peer-connected collection loop is armed */ -let collectionLoopActive = false -/** Peer-registry unsubscribe; set while stats service is armed */ -let unsubPeers = null -/** True after startStatsBroadcast(); false after stopStatsBroadcast() */ -let serviceArmed = false -let lastBroadcast = 0 -let dockerDownLogged = false -let dockerBackoffUntil = 0 -/** Serialize ticks so overlapping async collects cannot pile up */ -let tickInFlight = false -let largeFleetWarned = false - -/** Cached running containers from last listContainers */ -let rosterRunning = [] -/** @type {Set} */ -let rosterAllIds = new Set() -let rosterFetchedAt = 0 -/** Round-robin cursor into rosterRunning */ -let rrIndex = 0 - -/** - * Docker Engine CPU % (same idea as `docker stats`). - * One-shot stats populate precpu_stats after ~1s wait inside the engine. - */ function calculateCPUPercent(stats) { try { const cpu = stats?.cpu_stats @@ -344,7 +355,6 @@ function calculateCPUPercent(stats) { if (systemDelta > 0 && cpuDelta >= 0) { const pct = (cpuDelta / systemDelta) * cpuCount * 100.0 - // Clamp absurd spikes from clock glitches if (!Number.isFinite(pct) || pct < 0) return 0 return Math.min(pct, cpuCount * 100) } @@ -354,9 +364,6 @@ function calculateCPUPercent(stats) { } } -/** - * Working-set style memory (closer to `docker stats` MEM USAGE). - */ function calculateMemoryUsage(stats) { try { const mem = stats?.memory_stats @@ -366,12 +373,10 @@ function calculateMemoryUsage(stats) { const s = mem.stats || {} let working = usage if (s.inactive_file != null) { - // cgroup v2 working = Math.max(0, usage - Number(s.inactive_file)) } else if (s.total_inactive_file != null) { working = Math.max(0, usage - Number(s.total_inactive_file)) } else if (s.cache != null) { - // cgroup v1 working = Math.max(0, usage - Number(s.cache)) } return { usage: working, limit } @@ -409,13 +414,6 @@ function sumBlkioBytes(stats) { return { read, write } } -/** - * Bytes/sec rates from consecutive cumulative counters. - * @param {object} statsData - * @param {{ rx: number, tx: number }} net - * @param {{ read: number, write: number }} blk - * @param {number} now - */ function updateIoRates(statsData, net, blk, now) { const prevT = statsData._ioTs || 0 const dt = prevT ? Math.max(0.001, (now - prevT) / 1000) : 0 @@ -445,11 +443,6 @@ function isContainerActive(statsData) { return statsData.cpu > 0.5 || statsData.memory > 512 * 1024 } -/** - * IP from listContainers payload (avoids N× inspect). - * @param {object} containerInfo - * @returns {string} - */ function ipFromListContainer(containerInfo) { const nets = containerInfo?.NetworkSettings?.Networks if (nets && typeof nets === 'object') { @@ -463,6 +456,91 @@ function ipFromListContainer(containerInfo) { return 'No IP Assigned' } +function applyDockerStatsSample(statsData, sample) { + const now = Date.now() + statsData.cpu = calculateCPUPercent(sample) + const mem = calculateMemoryUsage(sample) + statsData.memory = mem.usage + statsData.memoryLimit = mem.limit + updateIoRates(statsData, sumNetworkBytes(sample), sumBlkioBytes(sample), now) + statsData.updatedAt = now +} + +// ─── Streams ────────────────────────────────────────────────────────────────── + +/** + * Attach live NDJSON stats stream for one running container. + * @param {object} statsData + * @param {import('dockerode').Container} container + */ +function attachStatsStream(statsData, container) { + if (statsData.stream || statsData._attaching) return + statsData._attaching = true + let buf = '' + + const onChunk = (chunk) => { + try { + buf += chunk.toString('utf8') + let nl + while ((nl = buf.indexOf('\n')) >= 0) { + const line = buf.slice(0, nl).trim() + buf = buf.slice(nl + 1) + if (!line) continue + try { + applyDockerStatsSample(statsData, JSON.parse(line)) + } catch { + // incomplete line + } + } + if (buf.length > 2 && buf.startsWith('{')) { + try { + const sample = JSON.parse(buf) + buf = '' + applyDockerStatsSample(statsData, sample) + } catch { + if (buf.length > 2_000_000) buf = '' + } + } + } catch (err) { + logger.debug('stats chunk parse failed', { id: statsData.id, error: err.message }) + } + } + + container + .stats({ stream: true }) + .then((statsStream) => { + statsData._attaching = false + if (!collectionActive || interestedPeers.size === 0) { + try { + statsStream.destroy() + } catch { + // ignore + } + return + } + statsData.stream = statsStream + statsStream.on('data', onChunk) + statsStream.on('error', (err) => { + logger.debug('Stats stream error', { id: statsData.id?.slice?.(0, 12), error: err.message }) + statsData.stream = null + }) + statsStream.on('close', () => { + statsData.stream = null + }) + statsStream.on('end', () => { + statsData.stream = null + }) + }) + .catch((err) => { + statsData._attaching = false + statsData.stream = null + logger.debug('Failed to start stats stream', { + id: statsData.id?.slice?.(0, 12), + error: err.message, + }) + }) +} + function ensureStatsEntry(containerInfo) { const id = containerInfo.Id let statsData = containerStats[id] @@ -482,11 +560,11 @@ function ensureStatsEntry(containerInfo) { blkReadTotal: 0, blkWriteTotal: 0, ip: ipFromListContainer(containerInfo), + stream: null, updatedAt: 0, } containerStats[id] = statsData } else { - // Refresh name / IP from list (no inspect) const name = containerInfo.Names?.[0]?.replace(/^\//, '') if (name) statsData.name = name const ip = ipFromListContainer(containerInfo) @@ -495,114 +573,23 @@ function ensureStatsEntry(containerInfo) { return statsData } -function applyDockerStatsSample(statsData, sample) { - const now = Date.now() - statsData.cpu = calculateCPUPercent(sample) - const mem = calculateMemoryUsage(sample) - statsData.memory = mem.usage - statsData.memoryLimit = mem.limit - updateIoRates(statsData, sumNetworkBytes(sample), sumBlkioBytes(sample), now) - statsData.updatedAt = now -} - -/** - * One-shot container stats. dockerode may return a Buffer/string or a brief stream; - * normalize to a single JSON object. - * @param {string} id - * @returns {Promise} - */ -async function fetchOneShotStats(id) { - const container = docker.getContainer(id) - const result = await container.stats({ stream: false }) - - // Most dockerode versions return a Promise resolving to the stats object or Buffer - if (result && typeof result === 'object' && !Buffer.isBuffer(result) && result.cpu_stats) { - return result - } - - // Stream-like (Node Readable) — rare with stream:false, but handle safely - if (result && typeof result.on === 'function') { - return await new Promise((resolve, reject) => { - let buf = '' - const timeout = setTimeout(() => { - try { - result.destroy?.() - } catch { - // ignore - } - reject(new Error('one-shot stats timeout')) - }, 15_000) - result.on('data', (chunk) => { - buf += chunk.toString('utf8') - // First complete JSON object is enough - const nl = buf.indexOf('\n') - const slice = nl >= 0 ? buf.slice(0, nl).trim() : buf.trim() - if (!slice.startsWith('{')) return - try { - const sample = JSON.parse(slice) - clearTimeout(timeout) - try { - result.destroy?.() - } catch { - // ignore - } - resolve(sample) - } catch { - // wait for more - } - }) - result.on('end', () => { - clearTimeout(timeout) - try { - resolve(JSON.parse(buf.trim())) - } catch (err) { - reject(err) - } - }) - result.on('error', (err) => { - clearTimeout(timeout) - reject(err) - }) - }) - } - - if (Buffer.isBuffer(result) || typeof result === 'string') { - const text = String(result).trim().split('\n').find(Boolean) - if (text) return JSON.parse(text) - } - - return result && typeof result === 'object' ? result : null -} - -/** - * Run async work over items with a fixed worker pool. - * @template T - * @param {T[]} items - * @param {number} concurrency - * @param {(item: T) => Promise} worker - */ -async function mapPool(items, concurrency, worker) { - if (!items.length) return - let i = 0 - const n = Math.min(concurrency, items.length) - async function run() { - while (i < items.length) { - const idx = i++ - await worker(items[idx]) - } - } - await Promise.all(Array.from({ length: n }, () => run())) -} - function destroyStatsEntry(id) { const statsData = containerStats[id] if (!statsData) return + if (statsData.stream) { + try { + statsData.stream.destroy() + } catch { + // ignore + } + } + statsData.stream = null delete containerStats[id] statsCache.delete(id) } /** - * Drop in-memory stats for a container before stop/remove. + * Drop live stats for a container before stop/remove. * @param {string} id */ export function destroyStatsForContainer(id) { @@ -621,220 +608,41 @@ export function destroyStatsForContainer(id) { } /** - * Refresh container roster from Docker (throttled). - * @param {boolean} [force] + * Sync streams with running containers (list + attach/detach). */ -async function refreshRoster(force = false) { - const now = Date.now() - if (!force && rosterFetchedAt && now - rosterFetchedAt < runtime.listIntervalMs) { - return - } - - const all = await docker.listContainers({ all: true }) - const running = all - .filter((c) => String(c.State || '').toLowerCase() === 'running') - .sort((a, b) => String(a.Id).localeCompare(String(b.Id))) - const allIds = new Set(all.map((c) => c.Id)) - const runningIds = new Set(running.map((c) => c.Id)) - - rosterRunning = running - rosterAllIds = allIds - rosterFetchedAt = now - if (rrIndex >= running.length) rrIndex = 0 - - if (running.length >= runtime.warnContainers && !largeFleetWarned) { - largeFleetWarned = true - logger.warn( - 'Large running fleet for stats collection — dockerd load scales with samples/tick', - { - running: running.length, - samplesPerTick: runtime.samplesPerTick, - concurrency: runtime.concurrency, - intervalMs: runtime.collectIntervalMs, - hint: 'Settings → Performance: raise interval, lower samples/tick & concurrency', - } - ) - } - - for (const containerInfo of running) { - ensureStatsEntry(containerInfo) - } - - for (const id of Object.keys(containerStats)) { - if (!allIds.has(id)) { - destroyStatsEntry(id) - continue - } - if (!runningIds.has(id)) { - const s = containerStats[id] - if (s) { - s.cpu = 0 - s.memory = 0 - s.netRxRate = 0 - s.netTxRate = 0 - s.blkReadRate = 0 - s.blkWriteRate = 0 - } - } - } -} - -/** - * Pick next round-robin batch of running containers (at most samplesPerTick). - * @param {object[]} running - * @returns {object[]} - */ -function nextSampleBatch(running) { - if (!running.length) return [] - const n = Math.min(runtime.samplesPerTick, running.length) - const batch = [] - for (let i = 0; i < n; i++) { - batch.push(running[(rrIndex + i) % running.length]) - } - rrIndex = (rrIndex + n) % running.length - return batch -} - -/** - * Throttled roster + round-robin one-shot samples. - * Dockerd cost ≈ samplesPerTick × ~1s / concurrency per tick, then idle for interval. - */ -async function collectContainerStats() { - await refreshRoster(false) - - const batch = nextSampleBatch(rosterRunning) - if (!batch.length) return - - await mapPool(batch, runtime.concurrency, async (containerInfo) => { - if (peers.size === 0) return - const statsData = containerStats[containerInfo.Id] - if (!statsData) return - try { - const sample = await fetchOneShotStats(containerInfo.Id) - if (sample) applyDockerStatsSample(statsData, sample) - } catch (err) { - logger.debug('one-shot stats failed', { - id: containerInfo.Id?.slice?.(0, 12), - error: err.message, - }) - } - }) -} - -/** - * Tear down in-memory collection state. - * Used when the last peer disconnects so Docker is not polled idle. - */ -function destroyAllStatsEntries() { - for (const id of Object.keys(containerStats)) { - destroyStatsEntry(id) - } - statsCache.clear() - containerActivity.clear() - largeFleetWarned = false - rosterRunning = [] - rosterAllIds = new Set() - rosterFetchedAt = 0 - rrIndex = 0 -} - -function clearLoopTimer() { - if (loopTimer) { - clearTimeout(loopTimer) - loopTimer = null - } -} - -/** - * Pause collection: stop loop and drop in-memory state. - * Safe to call when already paused. - */ -export function pauseStatsCollection() { - collectionLoopActive = false - clearLoopTimer() - destroyAllStatsEntries() - tickInFlight = false - logger.debug('Stats collection paused (no peers)') -} - -async function statsTick() { - if (tickInFlight) return - if (peers.size === 0) { - pauseStatsCollection() - return - } - tickInFlight = true +async function refreshRosterAndStreams() { + if (!collectionActive || interestedPeers.size === 0) return + if (rosterInFlight) return + if (Date.now() < dockerBackoffUntil) return + rosterInFlight = true try { - const now = Date.now() - if (now < dockerBackoffUntil) return - - await collectContainerStats() + const running = await docker.listContainers({ all: false }) dockerDownLogged = false + const runningIds = new Set(running.map((c) => c.Id)) - // Peer may have left while we were collecting - if (peers.size === 0) { - pauseStatsCollection() - return + if (running.length >= runtime.warnContainers && !largeFleetWarned) { + largeFleetWarned = true + logger.warn( + 'Many running containers with live stats streams — dockerd load scales with fleet size', + { + running: running.length, + watchers: interestedPeers.size, + hint: 'Stats only run while a client is on Containers/Dashboard; leave the tab to free dockerd', + } + ) } - if (now - lastBroadcast < runtime.broadcastIntervalMs) return - - const aggregatedStats = [] - for (const [containerId, statsData] of Object.entries(containerStats)) { - const cached = statsCache.get(containerId) - if ( - cached && - now - cached.timestamp < runtime.cacheTtlMs && - !isContainerActive(statsData) - ) { - aggregatedStats.push(cached.data) - continue + for (const info of running) { + const entry = ensureStatsEntry(info) + if (!entry.stream && !entry._attaching) { + attachStatsStream(entry, docker.getContainer(info.Id)) } - if (isContainerActive(statsData)) { - containerActivity.set(containerId, now) - } - const statsObj = { - id: statsData.id, - name: statsData.name, - cpu: Number(statsData.cpu) || 0, - memory: Number(statsData.memory) || 0, - memoryLimit: Number(statsData.memoryLimit) || 0, - netRxRate: Number(statsData.netRxRate) || 0, - netTxRate: Number(statsData.netTxRate) || 0, - blkReadRate: Number(statsData.blkReadRate) || 0, - blkWriteRate: Number(statsData.blkWriteRate) || 0, - netRxTotal: Number(statsData.netRxTotal) || 0, - netTxTotal: Number(statsData.netTxTotal) || 0, - blkReadTotal: Number(statsData.blkReadTotal) || 0, - blkWriteTotal: Number(statsData.blkWriteTotal) || 0, - ip: statsData.ip, - } - statsCache.set(containerId, { data: statsObj, timestamp: now }) - aggregatedStats.push(statsObj) - recordSample(containerId, { - cpu: statsObj.cpu, - memory: statsObj.memory, - memoryLimit: statsObj.memoryLimit, - netRxRate: statsObj.netRxRate, - netTxRate: statsObj.netTxRate, - blkReadRate: statsObj.blkReadRate, - blkWriteRate: statsObj.blkWriteRate, - }) - } - pruneMissing(Object.keys(containerStats)) - - // Always broadcast when we have containers — even zeros so UI can show 0.00% not "—" - if (aggregatedStats.length > 0) { - peers.broadcast(Pushes.allStats, { type: 'allStats', data: aggregatedStats }) - lastBroadcast = now } - const cacheTtl = Math.max(runtime.cacheTtlMs, 100) - for (const [id, cached] of statsCache.entries()) { - if (now - cached.timestamp > cacheTtl * 10) statsCache.delete(id) - } - for (const [id, ts] of containerActivity.entries()) { - if (now - ts > 60000) containerActivity.delete(id) + for (const id of Object.keys(containerStats)) { + if (!runningIds.has(id)) { + destroyStatsEntry(id) + } } } catch (err) { const msg = err.message || '' @@ -843,80 +651,204 @@ async function statsTick() { msg.includes('ECONNREFUSED') || msg.includes('docker.sock') if (dockerDown) { - dockerBackoffUntil = Date.now() + 15000 + dockerBackoffUntil = Date.now() + 15_000 if (!dockerDownLogged) { - logger.error('Docker unavailable; stats paused', { error: msg }) + logger.error('Docker unavailable; stats roster paused', { error: msg }) dockerDownLogged = true } } else { - logger.error('Stats broadcast failed', { error: msg }) + logger.debug('Stats roster refresh failed', { error: msg }) } } finally { - tickInFlight = false + rosterInFlight = false } } -/** - * Run one stats tick, then schedule the next after collectIntervalMs. - * Spacing is measured from tick *completion* so long sweeps cannot overlap. - */ -function runLoopTick() { - if (!collectionLoopActive || !serviceArmed || peers.size === 0) { - collectionLoopActive = false - clearLoopTimer() - return - } - statsTick() - .catch(() => {}) - .finally(() => { - if (!collectionLoopActive || !serviceArmed || peers.size === 0) { - collectionLoopActive = false - clearLoopTimer() - return - } - clearLoopTimer() - loopTimer = setTimeout(() => { - loopTimer = null - runLoopTick() - }, runtime.collectIntervalMs) - if (typeof loopTimer.unref === 'function') loopTimer.unref() +function buildAggregatedStats() { + const now = Date.now() + const aggregatedStats = [] + for (const [containerId, statsData] of Object.entries(containerStats)) { + if (isContainerActive(statsData)) { + containerActivity.set(containerId, now) + } + const statsObj = { + id: statsData.id, + name: statsData.name, + cpu: Number(statsData.cpu) || 0, + memory: Number(statsData.memory) || 0, + memoryLimit: Number(statsData.memoryLimit) || 0, + netRxRate: Number(statsData.netRxRate) || 0, + netTxRate: Number(statsData.netTxRate) || 0, + blkReadRate: Number(statsData.blkReadRate) || 0, + blkWriteRate: Number(statsData.blkWriteRate) || 0, + netRxTotal: Number(statsData.netRxTotal) || 0, + netTxTotal: Number(statsData.netTxTotal) || 0, + blkReadTotal: Number(statsData.blkReadTotal) || 0, + blkWriteTotal: Number(statsData.blkWriteTotal) || 0, + ip: statsData.ip, + } + statsCache.set(containerId, { data: statsObj, timestamp: now }) + aggregatedStats.push(statsObj) + recordSample(containerId, { + cpu: statsObj.cpu, + memory: statsObj.memory, + memoryLimit: statsObj.memoryLimit, + netRxRate: statsObj.netRxRate, + netTxRate: statsObj.netTxRate, + blkReadRate: statsObj.blkReadRate, + blkWriteRate: statsObj.blkWriteRate, }) + } + pruneMissing(Object.keys(containerStats)) + for (const [id, ts] of containerActivity.entries()) { + if (now - ts > 60_000) containerActivity.delete(id) + } + return aggregatedStats } /** - * Resume collection when at least one peer is connected. - * Idempotent; kicks an immediate tick so first client is not waiting a full interval. + * Push allStats only to peers that requested interest (not every connected peer). + * @param {object[]} aggregatedStats + */ +function broadcastToWatchers(aggregatedStats) { + if (!aggregatedStats.length || interestedPeers.size === 0) return + const payload = { type: 'allStats', data: aggregatedStats } + for (const peerId of interestedPeers) { + const session = peers.get(peerId) + if (!session) { + interestedPeers.delete(peerId) + peerViews.delete(peerId) + continue + } + try { + session.push(Pushes.allStats, payload) + } catch (err) { + logger.debug('stats push failed', { + peerId: peerId.slice(0, 12), + error: err.message, + }) + } + } + lastBroadcast = Date.now() +} + +function destroyAllStatsEntries() { + for (const id of Object.keys(containerStats)) { + destroyStatsEntry(id) + } + statsCache.clear() + containerActivity.clear() + largeFleetWarned = false +} + +function clearBroadcastLoop() { + if (broadcastTimer) { + clearTimeout(broadcastTimer) + broadcastTimer = null + } +} + +function clearRosterLoop() { + if (rosterTimer) { + clearTimeout(rosterTimer) + rosterTimer = null + } +} + +function armBroadcastLoop() { + clearBroadcastLoop() + if (!collectionActive) return + const tick = () => { + broadcastTimer = null + if (!collectionActive || interestedPeers.size === 0) return + try { + const data = buildAggregatedStats() + if (data.length > 0) broadcastToWatchers(data) + } catch (err) { + logger.debug('stats broadcast tick failed', { error: err.message }) + } + if (collectionActive && interestedPeers.size > 0) { + broadcastTimer = setTimeout(tick, runtime.broadcastIntervalMs) + if (typeof broadcastTimer.unref === 'function') broadcastTimer.unref() + } + } + // First push soon so UI is not blank for a full interval + broadcastTimer = setTimeout(tick, 200) + if (typeof broadcastTimer.unref === 'function') broadcastTimer.unref() +} + +function armRosterLoop() { + clearRosterLoop() + if (!collectionActive) return + const tick = () => { + rosterTimer = null + if (!collectionActive || interestedPeers.size === 0) return + refreshRosterAndStreams() + .catch(() => {}) + .finally(() => { + if (collectionActive && interestedPeers.size > 0) { + rosterTimer = setTimeout(tick, runtime.listIntervalMs) + if (typeof rosterTimer.unref === 'function') rosterTimer.unref() + } + }) + } + // Immediate roster + streams + tick() +} + +/** + * Stop all streams and timers (no watchers / shutdown). + */ +export function pauseStatsCollection() { + collectionActive = false + clearBroadcastLoop() + clearRosterLoop() + destroyAllStatsEntries() + logger.debug('Stats collection paused (no watchers)') +} + +/** + * Start live streams while at least one peer has interest. */ export function resumeStatsCollection() { if (!serviceArmed) return - if (peers.size === 0) return - if (collectionLoopActive) return - collectionLoopActive = true + if (interestedPeers.size === 0) return + if (collectionActive) return + collectionActive = true lastBroadcast = 0 - logger.debug('Stats collection resumed', { - peers: peers.size, - intervalMs: runtime.collectIntervalMs, - concurrency: runtime.concurrency, - samplesPerTick: runtime.samplesPerTick, - listIntervalMs: runtime.listIntervalMs, + logger.info('Stats collection resumed (view interest)', { + watchers: interestedPeers.size, + broadcastMs: runtime.broadcastIntervalMs, + listMs: runtime.listIntervalMs, }) - // Force fresh roster on first tick after resume - rosterFetchedAt = 0 - runLoopTick() + armRosterLoop() + armBroadcastLoop() } function onPeerCountChange(size, prevSize) { if (!serviceArmed) return - if (size > 0 && prevSize === 0) { - resumeStatsCollection() - } else if (size === 0 && prevSize > 0) { + // Drop interest for peers that no longer exist + if (size < prevSize) { + for (const id of [...interestedPeers]) { + if (!peers.get(id)) { + interestedPeers.delete(id) + peerViews.delete(id) + } + } + if (interestedPeers.size === 0) { + pauseStatsCollection() + } + } + // Connecting alone does NOT start stats — client must set interest for a view + if (size === 0) { + interestedPeers.clear() + peerViews.clear() pauseStatsCollection() } } /** - * Arm the stats service. Collection only runs while peers.size > 0. - * Call once at server boot (safe if already started). + * Arm the stats service at boot. Idle until a peer sets stats interest. */ export function startStatsBroadcast() { if (serviceArmed) return @@ -930,18 +862,14 @@ export function startStatsBroadcast() { unsubPeers = peers.onChange(onPeerCountChange) } - if (peers.size > 0) { + if (interestedPeers.size > 0) { resumeStatsCollection() } else { - // Explicit idle: no interval, no Docker stats requests pauseStatsCollection() - logger.debug('Stats service armed (idle until first peer)') + logger.debug('Stats service armed (idle until a client opens Containers/Dashboard)') } } -/** - * Fully stop the stats service (process shutdown). - */ export function stopStatsBroadcast() { serviceArmed = false if (unsubPeers) { @@ -952,34 +880,35 @@ export function stopStatsBroadcast() { } unsubPeers = null } + interestedPeers.clear() + peerViews.clear() pauseStatsCollection() } -/** @returns {boolean} whether the collect loop is currently armed */ +/** @returns {boolean} whether live streams are currently active */ export function isStatsCollectionActive() { - return collectionLoopActive + return collectionActive } -/** Test/ops helpers — same shape as getStatsConfig subset */ export function getStatsRuntimeConfig() { ensureStatsConfigLoaded() return { - collectIntervalMs: runtime.collectIntervalMs, broadcastIntervalMs: runtime.broadcastIntervalMs, + listIntervalMs: runtime.listIntervalMs, + collectIntervalMs: runtime.collectIntervalMs, concurrency: runtime.concurrency, samplesPerTick: runtime.samplesPerTick, - listIntervalMs: runtime.listIntervalMs, cacheTtlMs: runtime.cacheTtlMs, warnContainers: runtime.warnContainers, + watchers: interestedPeers.size, } } -/** - * Test helper: reset runtime to defaults and clear disk-loaded flag. - * Does not delete the on-disk file. - */ export function resetStatsConfigForTests() { diskLoaded = false runtime = defaultStatsConfig() + interestedPeers.clear() + peerViews.clear() largeFleetWarned = false + pauseStatsCollection() } diff --git a/shared/protocol.js b/shared/protocol.js index 8c06a81..cab62a7 100644 --- a/shared/protocol.js +++ b/shared/protocol.js @@ -43,6 +43,8 @@ export const MethodRoles = Object.freeze({ getMetrics: Roles.viewer, listSchedules: Roles.viewer, getStatsConfig: Roles.viewer, + /** Client view interest for live fleet stats (Containers / Dashboard) */ + setStatsInterest: Roles.viewer, browseVolume: Roles.viewer, getAlertsConfig: Roles.viewer, getAlertsStatus: Roles.viewer, @@ -317,6 +319,7 @@ export const Methods = Object.freeze({ // Docker stats collection (server-side, Settings → Performance) getStatsConfig: 'getStatsConfig', updateStatsConfig: 'updateStatsConfig', + setStatsInterest: 'setStatsInterest', // Alerts / webhooks (server-side) getAlertsConfig: 'getAlertsConfig', getAlertsStatus: 'getAlertsStatus', diff --git a/test/stats-config.test.js b/test/stats-config.test.js index d92689c..87d2e9a 100644 --- a/test/stats-config.test.js +++ b/test/stats-config.test.js @@ -33,17 +33,23 @@ test('normalizeStatsConfig clamps out-of-range values', (t) => { test('normalizeStatsConfig accepts seconds aliases from UI', (t) => { const n = normalizeStatsConfig({ - collectIntervalSec: 5, - broadcastIntervalSec: 3, + collectIntervalSec: 5, // legacy alias → broadcast when broadcast unset listIntervalSec: 20, concurrency: 4, samplesPerTick: 8, }) + t.is(n.broadcastIntervalMs, 5000) t.is(n.collectIntervalMs, 5000) - t.is(n.broadcastIntervalMs, 3000) t.is(n.listIntervalMs, 20000) t.is(n.concurrency, 4) t.is(n.samplesPerTick, 8) + + const n2 = normalizeStatsConfig({ + broadcastIntervalSec: 3, + collectIntervalSec: 9, + }) + // explicit broadcast wins over legacy collect alias + t.is(n2.broadcastIntervalMs, 3000) }) test('defaultStatsConfig is within limits', (t) => { diff --git a/test/stats-idle.test.js b/test/stats-idle.test.js index 5699ca4..9f61d16 100644 --- a/test/stats-idle.test.js +++ b/test/stats-idle.test.js @@ -1,5 +1,5 @@ /** - * Peer-demand stats: no Docker collection while zero peers. + * Peer-demand stats: no Docker collection until a peer sets view interest. */ import test from 'brittle' import { PeerRegistry } from '../server/core/peer-registry.js' @@ -29,7 +29,6 @@ test('PeerRegistry onChange fires on add/remove and size transitions', (t) => { t.is(reg.size, 0) t.alike(events[events.length - 1], [0, 1]) - // remove missing id is no-op (no event) const n = events.length reg.remove('nope') t.is(events.length, n) @@ -58,17 +57,16 @@ test('PeerRegistry.broadcast is a no-op with zero peers', (t) => { reg.clear() }) -test('stats service: idle when no peers; active when peer present', async (t) => { - // Isolate module state by dynamic import after resetting — use public API only. +test('stats service: idle until view interest; pauses when interest cleared', async (t) => { const stats = await import('../server/services/stats.js') const { peers } = await import('../server/core/peer-registry.js') - // Ensure clean slate from any prior test importing peers peers.clear() stats.stopStatsBroadcast() + stats.resetStatsConfigForTests?.() stats.startStatsBroadcast() - t.is(stats.isStatsCollectionActive(), false, 'no collect interval with 0 peers') + t.is(stats.isStatsCollectionActive(), false, 'idle with no interest') const fake = { id: 'test-peer-stats-idle', @@ -77,13 +75,47 @@ test('stats service: idle when no peers; active when peer present', async (t) => destroy() {}, } peers.add(fake) - // resume is async-kicked; give microtask + timer a tick - await new Promise((r) => setTimeout(r, 50)) - t.is(stats.isStatsCollectionActive(), true, 'collect interval after first peer') + await new Promise((r) => setTimeout(r, 30)) + // Connected alone is NOT enough — must declare interest + t.is(stats.isStatsCollectionActive(), false, 'peer connected but no view interest') - peers.remove(fake.id) + stats.setPeerStatsInterest(fake.id, true, { view: 'containers' }) + await new Promise((r) => setTimeout(r, 30)) + t.is(stats.isStatsCollectionActive(), true, 'active after setStatsInterest') + t.is(stats.getStatsInterestCount(), 1) + + stats.setPeerStatsInterest(fake.id, false) await new Promise((r) => setTimeout(r, 20)) - t.is(stats.isStatsCollectionActive(), false, 'collect interval stopped after last peer') + t.is(stats.isStatsCollectionActive(), false, 'paused after interest cleared') + t.is(stats.getStatsInterestCount(), 0) + + stats.stopStatsBroadcast() + peers.clear() +}) + +test('stats interest cleared when peer removed from registry', async (t) => { + const stats = await import('../server/services/stats.js') + const { peers } = await import('../server/core/peer-registry.js') + + peers.clear() + stats.stopStatsBroadcast() + stats.resetStatsConfigForTests?.() + stats.startStatsBroadcast() + + const fake = { + id: 'test-peer-stats-drop', + role: 'viewer', + push() {}, + destroy() {}, + } + peers.add(fake) + stats.setPeerStatsInterest(fake.id, true, { view: 'dashboard' }) + t.is(stats.isStatsCollectionActive(), true) + + peers.remove(fake.id) + await new Promise((r) => setTimeout(r, 20)) + t.is(stats.isStatsCollectionActive(), false, 'paused after peer disconnect') + t.is(stats.getStatsInterestCount(), 0) stats.stopStatsBroadcast() peers.clear() diff --git a/ui/stats-settings.js b/ui/stats-settings.js index 52f611e..ea972b4 100644 --- a/ui/stats-settings.js +++ b/ui/stats-settings.js @@ -1,6 +1,6 @@ /** * Settings → Performance - * Server-side Docker stats collection (interval, concurrency). + * Server-side Docker stats (view-interest streams + push cadence). */ import { Methods } from '../shared/protocol.js' import { manager } from '../client/manager.js' @@ -8,7 +8,6 @@ import { presentError } from '../client/errors.js' /** @type {object|null} */ let statsState = null -/** Skip auto-apply while hydrating form from server */ let suppressLiveApply = false /** @type {ReturnType|null} */ let applyTimer = null @@ -72,47 +71,42 @@ function renderStatsForm(config) { if (!config) return suppressLiveApply = true try { - const collect = document.getElementById('stats-collect-sec') - if (collect) { - collect.value = String( - config.collectIntervalSec ?? Math.round((config.collectIntervalMs || 5000) / 1000) - ) - } const broadcast = document.getElementById('stats-broadcast-sec') if (broadcast) { broadcast.value = String( - config.broadcastIntervalSec ?? Math.round((config.broadcastIntervalMs || 5000) / 1000) + config.broadcastIntervalSec ?? Math.round((config.broadcastIntervalMs || 1000) / 1000) ) } - const samples = document.getElementById('stats-samples-per-tick') - if (samples) samples.value = String(config.samplesPerTick ?? 4) - const conc = document.getElementById('stats-concurrency') - if (conc) conc.value = String(config.concurrency ?? 2) const listSec = document.getElementById('stats-list-sec') if (listSec) { listSec.value = String( - config.listIntervalSec ?? Math.round((config.listIntervalMs || 15000) / 1000) + config.listIntervalSec ?? Math.round((config.listIntervalMs || 10000) / 1000) ) } - const cache = document.getElementById('stats-cache-ttl-ms') - if (cache) cache.value = String(config.cacheTtlMs ?? 2000) const warn = document.getElementById('stats-warn-containers') if (warn) warn.value = String(config.warnContainers ?? 80) - const active = config.active ? 'collecting' : 'idle (no peers or paused)' - setEngineStatus( - `${active} · gap ${config.collectIntervalSec ?? '?'}s · ${config.samplesPerTick ?? '?'} samples · ×${config.concurrency ?? '?'}` - ) + // Keep hidden legacy inputs in sync if present + const collect = document.getElementById('stats-collect-sec') + if (collect) { + collect.value = String( + config.broadcastIntervalSec ?? Math.round((config.broadcastIntervalMs || 1000) / 1000) + ) + } + + const active = config.active + ? `streaming · ${config.streams ?? '?'} streams · ${config.watchers ?? 0} watcher(s)` + : 'idle (no client on Containers/Dashboard)' + setEngineStatus(active) const meta = document.getElementById('stats-runtime-meta') if (meta) { const parts = [ + `mode=${config.mode || 'interest-stream'}`, `peers=${config.peers ?? 0}`, - `running=${config.runningTracked ?? '?'}`, - `gap=${config.collectIntervalMs}ms`, - `samples/tick=${config.samplesPerTick}`, - `concurrency=${config.concurrency}`, - `list=${config.listIntervalMs}ms`, + `watchers=${config.watchers ?? 0}`, + `streams=${config.streams ?? 0}`, `broadcast=${config.broadcastIntervalMs}ms`, + `list=${config.listIntervalMs}ms`, ] if (config.path) parts.push(`file=${config.path}`) meta.textContent = parts.join(' · ') @@ -125,20 +119,12 @@ function renderStatsForm(config) { } function readFormPartial() { - const collectSec = Number(document.getElementById('stats-collect-sec')?.value) const broadcastSec = Number(document.getElementById('stats-broadcast-sec')?.value) - const samplesPerTick = Number(document.getElementById('stats-samples-per-tick')?.value) - const concurrency = Number(document.getElementById('stats-concurrency')?.value) const listSec = Number(document.getElementById('stats-list-sec')?.value) - const cacheTtlMs = Number(document.getElementById('stats-cache-ttl-ms')?.value) const warnContainers = Number(document.getElementById('stats-warn-containers')?.value) return { - collectIntervalSec: Number.isFinite(collectSec) ? collectSec : undefined, broadcastIntervalSec: Number.isFinite(broadcastSec) ? broadcastSec : undefined, - samplesPerTick: Number.isFinite(samplesPerTick) ? samplesPerTick : undefined, - concurrency: Number.isFinite(concurrency) ? concurrency : undefined, listIntervalSec: Number.isFinite(listSec) ? listSec : undefined, - cacheTtlMs: Number.isFinite(cacheTtlMs) ? cacheTtlMs : undefined, warnContainers: Number.isFinite(warnContainers) ? warnContainers : undefined, } } @@ -209,9 +195,10 @@ export function initStatsSettings() { panel.dataset.statsWired = '1' panel.querySelectorAll('.stats-live-field').forEach((el) => { + // Skip hidden legacy fields from auto-apply noise + if (el.type === 'hidden') return el.addEventListener('change', () => scheduleApply()) el.addEventListener('input', () => { - // Debounce number typing if (el.tagName === 'INPUT') scheduleApply() }) })