Raven Scott
c5015a66e8
Release rolling / release (push) Successful in 8m4s
1. Removed docker.df() entirely 2. Poll only what rules need • docker_daemon → cheap ping • container_health / stack_health → listContainers • resource → host statfs only (no Docker) • Event-only rules → no poll timer 3. Poll floor 60s, default 120s (was 15s min / 60s default) 4. No overlapping polls (pollInFlight) 5. Event path: drop noisy actions (exec_*, attach, …), match only relevant rules, serialize work (bounded queue) 6. UI status shows whether poll is active
986 lines
29 KiB
JavaScript
986 lines
29 KiB
JavaScript
/**
|
||
* Container stats collection and broadcast.
|
||
*
|
||
* Strategy (tuned against dockerd load, informed by Dozzle's approach):
|
||
*
|
||
* - **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.
|
||
*/
|
||
import fs from 'fs'
|
||
import path from 'path'
|
||
import { docker } from './docker.js'
|
||
import { peers } from '../core/peer-registry.js'
|
||
import { Pushes } from '../../shared/protocol.js'
|
||
import { recordSample, pruneMissing } from './stats-history.js'
|
||
import { CONFIG } from '../../config.js'
|
||
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 },
|
||
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) {
|
||
const n = Number(process.env[name])
|
||
return Number.isFinite(n) && n > 0 ? Math.floor(n) : fallback
|
||
}
|
||
|
||
function clampInt(value, min, max, fallback) {
|
||
const n = Number(value)
|
||
if (!Number.isFinite(n)) return fallback
|
||
return Math.min(max, Math.max(min, Math.floor(n)))
|
||
}
|
||
|
||
/**
|
||
* 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,
|
||
* }}
|
||
*/
|
||
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),
|
||
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
|
||
)
|
||
const listIntervalMs = clampInt(
|
||
envInt('PEARDOCK_STATS_LIST_INTERVAL_MS', CONFIG.STATS?.LIST_INTERVAL_MS ?? 15_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
|
||
)
|
||
const warnContainers = clampInt(
|
||
envInt('PEARDOCK_STATS_WARN_CONTAINERS', 80),
|
||
STATS_LIMITS.warnContainers.min,
|
||
STATS_LIMITS.warnContainers.max,
|
||
80
|
||
)
|
||
return {
|
||
collectIntervalMs,
|
||
broadcastIntervalMs,
|
||
concurrency,
|
||
samplesPerTick,
|
||
listIntervalMs,
|
||
cacheTtlMs,
|
||
warnContainers,
|
||
}
|
||
}
|
||
|
||
/** @type {ReturnType<typeof defaultStatsConfig>} */
|
||
let runtime = defaultStatsConfig()
|
||
let diskLoaded = false
|
||
|
||
function loadStatsDisk() {
|
||
try {
|
||
if (!fs.existsSync(STATS_FILE)) return null
|
||
const raw = JSON.parse(fs.readFileSync(STATS_FILE, 'utf8'))
|
||
if (!raw || typeof raw !== 'object') return null
|
||
return raw
|
||
} catch (err) {
|
||
logger.warn('Failed to load stats config', { error: err.message, path: STATS_FILE })
|
||
return null
|
||
}
|
||
}
|
||
|
||
function saveStatsDisk() {
|
||
try {
|
||
const payload = {
|
||
version: 1,
|
||
updatedAt: new Date().toISOString(),
|
||
collectIntervalMs: runtime.collectIntervalMs,
|
||
broadcastIntervalMs: runtime.broadcastIntervalMs,
|
||
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 })
|
||
fs.renameSync(tmp, STATS_FILE)
|
||
} catch (err) {
|
||
logger.warn('Failed to save stats config', { error: err.message, path: STATS_FILE })
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Normalize a partial config object with clamps.
|
||
* @param {object} partial
|
||
* @param {ReturnType<typeof defaultStatsConfig>} [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
|
||
}
|
||
let listMs = src.listIntervalMs
|
||
if (listMs == null && src.listIntervalSec != null) {
|
||
listMs = Number(src.listIntervalSec) * 1000
|
||
}
|
||
return {
|
||
collectIntervalMs: clampInt(
|
||
collectMs ?? base.collectIntervalMs,
|
||
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
|
||
),
|
||
concurrency: clampInt(
|
||
src.concurrency ?? base.concurrency,
|
||
STATS_LIMITS.concurrency.min,
|
||
STATS_LIMITS.concurrency.max,
|
||
base.concurrency
|
||
),
|
||
samplesPerTick: clampInt(
|
||
src.samplesPerTick ?? base.samplesPerTick,
|
||
STATS_LIMITS.samplesPerTick.min,
|
||
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
|
||
runtime = defaultStatsConfig()
|
||
const disk = loadStatsDisk()
|
||
if (disk) {
|
||
runtime = normalizeStatsConfig(disk, runtime)
|
||
logger.debug('Stats config loaded from disk', {
|
||
path: STATS_FILE,
|
||
collectIntervalMs: runtime.collectIntervalMs,
|
||
concurrency: runtime.concurrency,
|
||
})
|
||
}
|
||
return runtime
|
||
}
|
||
|
||
/**
|
||
* Public config snapshot for RPC / Settings UI.
|
||
* @returns {object}
|
||
*/
|
||
export function getStatsConfig() {
|
||
ensureStatsConfigLoaded()
|
||
return {
|
||
collectIntervalMs: runtime.collectIntervalMs,
|
||
broadcastIntervalMs: runtime.broadcastIntervalMs,
|
||
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,
|
||
peers: peers.size,
|
||
runningTracked: rosterRunning.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)
|
||
runtime = next
|
||
if (opts.persist !== false) saveStatsDisk()
|
||
// Interval is applied on the next schedule after a tick completes
|
||
logger.info('Stats config updated', {
|
||
collectIntervalMs: runtime.collectIntervalMs,
|
||
broadcastIntervalMs: runtime.broadcastIntervalMs,
|
||
concurrency: runtime.concurrency,
|
||
samplesPerTick: runtime.samplesPerTick,
|
||
listIntervalMs: runtime.listIntervalMs,
|
||
cacheTtlMs: runtime.cacheTtlMs,
|
||
persist: opts.persist !== false,
|
||
})
|
||
return getStatsConfig()
|
||
}
|
||
|
||
export function getStatsFilePath() {
|
||
return STATS_FILE
|
||
}
|
||
|
||
/** @type {Record<string, object>} */
|
||
const containerStats = {}
|
||
const statsCache = new Map()
|
||
const containerActivity = new Map()
|
||
|
||
/** @type {ReturnType<typeof setTimeout>|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<string>} */
|
||
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
|
||
const precpu = stats?.precpu_stats
|
||
if (!cpu?.cpu_usage || !precpu?.cpu_usage) return 0
|
||
|
||
const cpuDelta = (cpu.cpu_usage.total_usage || 0) - (precpu.cpu_usage.total_usage || 0)
|
||
const systemDelta = (cpu.system_cpu_usage || 0) - (precpu.system_cpu_usage || 0)
|
||
|
||
let cpuCount = cpu.online_cpus || 0
|
||
if (!cpuCount && Array.isArray(cpu.cpu_usage.percpu_usage)) {
|
||
cpuCount = cpu.cpu_usage.percpu_usage.length
|
||
}
|
||
if (!cpuCount) cpuCount = 1
|
||
|
||
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)
|
||
}
|
||
return 0
|
||
} catch {
|
||
return 0
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Working-set style memory (closer to `docker stats` MEM USAGE).
|
||
*/
|
||
function calculateMemoryUsage(stats) {
|
||
try {
|
||
const mem = stats?.memory_stats
|
||
if (!mem) return { usage: 0, limit: 0 }
|
||
const usage = Number(mem.usage) || 0
|
||
const limit = Number(mem.limit) || 0
|
||
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 }
|
||
} catch {
|
||
return { usage: 0, limit: 0 }
|
||
}
|
||
}
|
||
|
||
function sumNetworkBytes(stats) {
|
||
let rx = 0
|
||
let tx = 0
|
||
const nets = stats?.networks
|
||
if (!nets || typeof nets !== 'object') return { rx, tx }
|
||
for (const n of Object.values(nets)) {
|
||
rx += Number(n?.rx_bytes) || 0
|
||
tx += Number(n?.tx_bytes) || 0
|
||
}
|
||
return { rx, tx }
|
||
}
|
||
|
||
function sumBlkioBytes(stats) {
|
||
let read = 0
|
||
let write = 0
|
||
const arr =
|
||
stats?.blkio_stats?.io_service_bytes_recursive ||
|
||
stats?.blkio_stats?.io_service_bytes ||
|
||
[]
|
||
if (!Array.isArray(arr)) return { read, write }
|
||
for (const e of arr) {
|
||
const op = String(e?.op || '').toLowerCase()
|
||
const v = Number(e?.value) || 0
|
||
if (op === 'read') read += v
|
||
else if (op === 'write') write += v
|
||
}
|
||
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
|
||
if (dt > 0 && statsData._netRx != null) {
|
||
statsData.netRxRate = Math.max(0, (net.rx - statsData._netRx) / dt)
|
||
statsData.netTxRate = Math.max(0, (net.tx - statsData._netTx) / dt)
|
||
statsData.blkReadRate = Math.max(0, (blk.read - statsData._blkRead) / dt)
|
||
statsData.blkWriteRate = Math.max(0, (blk.write - statsData._blkWrite) / dt)
|
||
} else {
|
||
statsData.netRxRate = statsData.netRxRate || 0
|
||
statsData.netTxRate = statsData.netTxRate || 0
|
||
statsData.blkReadRate = statsData.blkReadRate || 0
|
||
statsData.blkWriteRate = statsData.blkWriteRate || 0
|
||
}
|
||
statsData._netRx = net.rx
|
||
statsData._netTx = net.tx
|
||
statsData._blkRead = blk.read
|
||
statsData._blkWrite = blk.write
|
||
statsData._ioTs = now
|
||
statsData.netRxTotal = net.rx
|
||
statsData.netTxTotal = net.tx
|
||
statsData.blkReadTotal = blk.read
|
||
statsData.blkWriteTotal = blk.write
|
||
}
|
||
|
||
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') {
|
||
for (const net of Object.values(nets)) {
|
||
if (net?.IPAddress) return net.IPAddress
|
||
}
|
||
}
|
||
if (containerInfo?.NetworkSettings?.IPAddress) {
|
||
return containerInfo.NetworkSettings.IPAddress
|
||
}
|
||
return 'No IP Assigned'
|
||
}
|
||
|
||
function ensureStatsEntry(containerInfo) {
|
||
const id = containerInfo.Id
|
||
let statsData = containerStats[id]
|
||
if (!statsData) {
|
||
statsData = {
|
||
id,
|
||
name: containerInfo.Names?.[0]?.replace(/^\//, '') || 'Unknown',
|
||
cpu: 0,
|
||
memory: 0,
|
||
memoryLimit: 0,
|
||
netRxRate: 0,
|
||
netTxRate: 0,
|
||
blkReadRate: 0,
|
||
blkWriteRate: 0,
|
||
netRxTotal: 0,
|
||
netTxTotal: 0,
|
||
blkReadTotal: 0,
|
||
blkWriteTotal: 0,
|
||
ip: ipFromListContainer(containerInfo),
|
||
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)
|
||
if (ip && ip !== 'No IP Assigned') statsData.ip = ip
|
||
}
|
||
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<object|null>}
|
||
*/
|
||
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<void>} 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
|
||
delete containerStats[id]
|
||
statsCache.delete(id)
|
||
}
|
||
|
||
/**
|
||
* Drop in-memory stats for a container before stop/remove.
|
||
* @param {string} id
|
||
*/
|
||
export function destroyStatsForContainer(id) {
|
||
if (!id) return
|
||
const needle = String(id)
|
||
destroyStatsEntry(needle)
|
||
for (const key of Object.keys(containerStats)) {
|
||
if (key === needle) continue
|
||
if (
|
||
(needle.length >= 12 && key.startsWith(needle)) ||
|
||
(key.length >= 12 && needle.startsWith(key.slice(0, Math.min(12, key.length))))
|
||
) {
|
||
destroyStatsEntry(key)
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Refresh container roster from Docker (throttled).
|
||
* @param {boolean} [force]
|
||
*/
|
||
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
|
||
try {
|
||
const now = Date.now()
|
||
if (now < dockerBackoffUntil) return
|
||
|
||
await collectContainerStats()
|
||
dockerDownLogged = false
|
||
|
||
// Peer may have left while we were collecting
|
||
if (peers.size === 0) {
|
||
pauseStatsCollection()
|
||
return
|
||
}
|
||
|
||
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
|
||
}
|
||
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)
|
||
}
|
||
} catch (err) {
|
||
const msg = err.message || ''
|
||
const dockerDown =
|
||
msg.includes('ENOENT') ||
|
||
msg.includes('ECONNREFUSED') ||
|
||
msg.includes('docker.sock')
|
||
if (dockerDown) {
|
||
dockerBackoffUntil = Date.now() + 15000
|
||
if (!dockerDownLogged) {
|
||
logger.error('Docker unavailable; stats paused', { error: msg })
|
||
dockerDownLogged = true
|
||
}
|
||
} else {
|
||
logger.error('Stats broadcast failed', { error: msg })
|
||
}
|
||
} finally {
|
||
tickInFlight = 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()
|
||
})
|
||
}
|
||
|
||
/**
|
||
* Resume collection when at least one peer is connected.
|
||
* Idempotent; kicks an immediate tick so first client is not waiting a full interval.
|
||
*/
|
||
export function resumeStatsCollection() {
|
||
if (!serviceArmed) return
|
||
if (peers.size === 0) return
|
||
if (collectionLoopActive) return
|
||
collectionLoopActive = true
|
||
lastBroadcast = 0
|
||
logger.debug('Stats collection resumed', {
|
||
peers: peers.size,
|
||
intervalMs: runtime.collectIntervalMs,
|
||
concurrency: runtime.concurrency,
|
||
samplesPerTick: runtime.samplesPerTick,
|
||
listIntervalMs: runtime.listIntervalMs,
|
||
})
|
||
// Force fresh roster on first tick after resume
|
||
rosterFetchedAt = 0
|
||
runLoopTick()
|
||
}
|
||
|
||
function onPeerCountChange(size, prevSize) {
|
||
if (!serviceArmed) return
|
||
if (size > 0 && prevSize === 0) {
|
||
resumeStatsCollection()
|
||
} else if (size === 0 && prevSize > 0) {
|
||
pauseStatsCollection()
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Arm the stats service. Collection only runs while peers.size > 0.
|
||
* Call once at server boot (safe if already started).
|
||
*/
|
||
export function startStatsBroadcast() {
|
||
if (serviceArmed) return
|
||
ensureStatsConfigLoaded()
|
||
serviceArmed = true
|
||
lastBroadcast = 0
|
||
dockerDownLogged = false
|
||
dockerBackoffUntil = 0
|
||
|
||
if (!unsubPeers) {
|
||
unsubPeers = peers.onChange(onPeerCountChange)
|
||
}
|
||
|
||
if (peers.size > 0) {
|
||
resumeStatsCollection()
|
||
} else {
|
||
// Explicit idle: no interval, no Docker stats requests
|
||
pauseStatsCollection()
|
||
logger.debug('Stats service armed (idle until first peer)')
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Fully stop the stats service (process shutdown).
|
||
*/
|
||
export function stopStatsBroadcast() {
|
||
serviceArmed = false
|
||
if (unsubPeers) {
|
||
try {
|
||
unsubPeers()
|
||
} catch {
|
||
// ignore
|
||
}
|
||
unsubPeers = null
|
||
}
|
||
pauseStatsCollection()
|
||
}
|
||
|
||
/** @returns {boolean} whether the collect loop is currently armed */
|
||
export function isStatsCollectionActive() {
|
||
return collectionLoopActive
|
||
}
|
||
|
||
/** Test/ops helpers — same shape as getStatsConfig subset */
|
||
export function getStatsRuntimeConfig() {
|
||
ensureStatsConfigLoaded()
|
||
return {
|
||
collectIntervalMs: runtime.collectIntervalMs,
|
||
broadcastIntervalMs: runtime.broadcastIntervalMs,
|
||
concurrency: runtime.concurrency,
|
||
samplesPerTick: runtime.samplesPerTick,
|
||
listIntervalMs: runtime.listIntervalMs,
|
||
cacheTtlMs: runtime.cacheTtlMs,
|
||
warnContainers: runtime.warnContainers,
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 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()
|
||
largeFleetWarned = false
|
||
}
|