915 lines
27 KiB
JavaScript
915 lines
27 KiB
JavaScript
/**
|
|
* Container stats collection and broadcast.
|
|
*
|
|
* Demand-driven real-time fleet stats:
|
|
*
|
|
* - 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'
|
|
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')
|
|
|
|
export const STATS_LIMITS = Object.freeze({
|
|
/** 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 },
|
|
cacheTtlMs: { min: 0, max: 30_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)))
|
|
}
|
|
|
|
/**
|
|
* 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 broadcastIntervalMs = clampInt(
|
|
envInt(
|
|
'PEARDOCK_STATS_BROADCAST_MS',
|
|
CONFIG.STATS?.INTERVAL_MS ?? CONFIG.STATS?.ACTIVE_INTERVAL_MS ?? 1000
|
|
),
|
|
STATS_LIMITS.broadcastIntervalMs.min,
|
|
STATS_LIMITS.broadcastIntervalMs.max,
|
|
1000
|
|
)
|
|
const listIntervalMs = clampInt(
|
|
envInt('PEARDOCK_STATS_LIST_INTERVAL_MS', CONFIG.STATS?.LIST_INTERVAL_MS ?? 10_000),
|
|
STATS_LIMITS.listIntervalMs.min,
|
|
STATS_LIMITS.listIntervalMs.max,
|
|
10_000
|
|
)
|
|
const warnContainers = clampInt(
|
|
envInt('PEARDOCK_STATS_WARN_CONTAINERS', 80),
|
|
STATS_LIMITS.warnContainers.min,
|
|
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 {
|
|
broadcastIntervalMs,
|
|
listIntervalMs,
|
|
warnContainers,
|
|
collectIntervalMs,
|
|
concurrency: 2,
|
|
samplesPerTick: 4,
|
|
cacheTtlMs: 1000,
|
|
}
|
|
}
|
|
|
|
/** @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: 2,
|
|
mode: 'interest-stream',
|
|
updatedAt: new Date().toISOString(),
|
|
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,
|
|
cacheTtlMs: runtime.cacheTtlMs,
|
|
}
|
|
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 })
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @param {object} partial
|
|
* @param {ReturnType<typeof defaultStatsConfig>} [base]
|
|
*/
|
|
export function normalizeStatsConfig(partial = {}, base = defaultStatsConfig()) {
|
|
const src = partial && typeof partial === 'object' ? partial : {}
|
|
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(
|
|
src.collectIntervalMs ?? broadcastIntervalMs,
|
|
STATS_LIMITS.collectIntervalMs.min,
|
|
STATS_LIMITS.collectIntervalMs.max,
|
|
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
|
|
),
|
|
cacheTtlMs: clampInt(
|
|
src.cacheTtlMs ?? base.cacheTtlMs,
|
|
STATS_LIMITS.cacheTtlMs.min,
|
|
STATS_LIMITS.cacheTtlMs.max,
|
|
base.cacheTtlMs
|
|
),
|
|
}
|
|
}
|
|
|
|
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,
|
|
broadcastIntervalMs: runtime.broadcastIntervalMs,
|
|
})
|
|
}
|
|
return runtime
|
|
}
|
|
|
|
// ─── Interest (view-driven) ───────────────────────────────────────────────────
|
|
|
|
/** Peer ids that currently want live fleet stats */
|
|
const interestedPeers = new Set()
|
|
/** @type {Map<string, string>} peerId → view name (debug) */
|
|
const peerViews = new Map()
|
|
|
|
/**
|
|
* 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<string, object>} */
|
|
const containerStats = {}
|
|
const statsCache = new Map()
|
|
const containerActivity = new Map()
|
|
|
|
/** @type {ReturnType<typeof setTimeout>|null} */
|
|
let broadcastTimer = null
|
|
/** @type {ReturnType<typeof setTimeout>|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 {
|
|
mode: 'interest-stream',
|
|
broadcastIntervalMs: runtime.broadcastIntervalMs,
|
|
listIntervalMs: runtime.listIntervalMs,
|
|
warnContainers: runtime.warnContainers,
|
|
collectIntervalMs: runtime.collectIntervalMs,
|
|
concurrency: runtime.concurrency,
|
|
samplesPerTick: runtime.samplesPerTick,
|
|
cacheTtlMs: runtime.cacheTtlMs,
|
|
collectIntervalSec: Math.round(runtime.collectIntervalMs / 1000),
|
|
broadcastIntervalSec: Math.round(runtime.broadcastIntervalMs / 1000),
|
|
listIntervalSec: Math.round(runtime.listIntervalMs / 1000),
|
|
active: collectionActive,
|
|
peers: peers.size,
|
|
watchers: interestedPeers.size,
|
|
streams: Object.values(containerStats).filter((s) => s.stream).length,
|
|
runningTracked: Object.keys(containerStats).length,
|
|
limits: { ...STATS_LIMITS },
|
|
path: STATS_FILE,
|
|
}
|
|
}
|
|
|
|
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()
|
|
if (collectionActive) {
|
|
if (broadcastChanged) armBroadcastLoop()
|
|
if (listChanged) armRosterLoop()
|
|
}
|
|
logger.info('Stats config updated', {
|
|
broadcastIntervalMs: runtime.broadcastIntervalMs,
|
|
listIntervalMs: runtime.listIntervalMs,
|
|
watchers: interestedPeers.size,
|
|
persist: opts.persist !== false,
|
|
})
|
|
return getStatsConfig()
|
|
}
|
|
|
|
export function getStatsFilePath() {
|
|
return STATS_FILE
|
|
}
|
|
|
|
// ─── CPU / mem helpers ────────────────────────────────────────────────────────
|
|
|
|
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
|
|
if (!Number.isFinite(pct) || pct < 0) return 0
|
|
return Math.min(pct, cpuCount * 100)
|
|
}
|
|
return 0
|
|
} catch {
|
|
return 0
|
|
}
|
|
}
|
|
|
|
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) {
|
|
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) {
|
|
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 }
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
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 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]
|
|
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),
|
|
stream: null,
|
|
updatedAt: 0,
|
|
}
|
|
containerStats[id] = statsData
|
|
} else {
|
|
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 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 live 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)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Sync streams with running containers (list + attach/detach).
|
|
*/
|
|
async function refreshRosterAndStreams() {
|
|
if (!collectionActive || interestedPeers.size === 0) return
|
|
if (rosterInFlight) return
|
|
if (Date.now() < dockerBackoffUntil) return
|
|
rosterInFlight = true
|
|
try {
|
|
const running = await docker.listContainers({ all: false })
|
|
dockerDownLogged = false
|
|
const runningIds = new Set(running.map((c) => c.Id))
|
|
|
|
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',
|
|
}
|
|
)
|
|
}
|
|
|
|
for (const info of running) {
|
|
const entry = ensureStatsEntry(info)
|
|
if (!entry.stream && !entry._attaching) {
|
|
attachStatsStream(entry, docker.getContainer(info.Id))
|
|
}
|
|
}
|
|
|
|
for (const id of Object.keys(containerStats)) {
|
|
if (!runningIds.has(id)) {
|
|
destroyStatsEntry(id)
|
|
}
|
|
}
|
|
} catch (err) {
|
|
const msg = err.message || ''
|
|
const dockerDown =
|
|
msg.includes('ENOENT') ||
|
|
msg.includes('ECONNREFUSED') ||
|
|
msg.includes('docker.sock')
|
|
if (dockerDown) {
|
|
dockerBackoffUntil = Date.now() + 15_000
|
|
if (!dockerDownLogged) {
|
|
logger.error('Docker unavailable; stats roster paused', { error: msg })
|
|
dockerDownLogged = true
|
|
}
|
|
} else {
|
|
logger.debug('Stats roster refresh failed', { error: msg })
|
|
}
|
|
} finally {
|
|
rosterInFlight = false
|
|
}
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
/**
|
|
* 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 (interestedPeers.size === 0) return
|
|
if (collectionActive) return
|
|
collectionActive = true
|
|
lastBroadcast = 0
|
|
logger.info('Stats collection resumed (view interest)', {
|
|
watchers: interestedPeers.size,
|
|
broadcastMs: runtime.broadcastIntervalMs,
|
|
listMs: runtime.listIntervalMs,
|
|
})
|
|
armRosterLoop()
|
|
armBroadcastLoop()
|
|
}
|
|
|
|
function onPeerCountChange(size, prevSize) {
|
|
if (!serviceArmed) return
|
|
// 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 at boot. Idle until a peer sets stats interest.
|
|
*/
|
|
export function startStatsBroadcast() {
|
|
if (serviceArmed) return
|
|
ensureStatsConfigLoaded()
|
|
serviceArmed = true
|
|
lastBroadcast = 0
|
|
dockerDownLogged = false
|
|
dockerBackoffUntil = 0
|
|
|
|
if (!unsubPeers) {
|
|
unsubPeers = peers.onChange(onPeerCountChange)
|
|
}
|
|
|
|
if (interestedPeers.size > 0) {
|
|
resumeStatsCollection()
|
|
} else {
|
|
pauseStatsCollection()
|
|
logger.debug('Stats service armed (idle until a client opens Containers/Dashboard)')
|
|
}
|
|
}
|
|
|
|
export function stopStatsBroadcast() {
|
|
serviceArmed = false
|
|
if (unsubPeers) {
|
|
try {
|
|
unsubPeers()
|
|
} catch {
|
|
// ignore
|
|
}
|
|
unsubPeers = null
|
|
}
|
|
interestedPeers.clear()
|
|
peerViews.clear()
|
|
pauseStatsCollection()
|
|
}
|
|
|
|
/** @returns {boolean} whether live streams are currently active */
|
|
export function isStatsCollectionActive() {
|
|
return collectionActive
|
|
}
|
|
|
|
export function getStatsRuntimeConfig() {
|
|
ensureStatsConfigLoaded()
|
|
return {
|
|
broadcastIntervalMs: runtime.broadcastIntervalMs,
|
|
listIntervalMs: runtime.listIntervalMs,
|
|
collectIntervalMs: runtime.collectIntervalMs,
|
|
concurrency: runtime.concurrency,
|
|
samplesPerTick: runtime.samplesPerTick,
|
|
cacheTtlMs: runtime.cacheTtlMs,
|
|
warnContainers: runtime.warnContainers,
|
|
watchers: interestedPeers.size,
|
|
}
|
|
}
|
|
|
|
export function resetStatsConfigForTests() {
|
|
diskLoaded = false
|
|
runtime = defaultStatsConfig()
|
|
interestedPeers.clear()
|
|
peerViews.clear()
|
|
largeFleetWarned = false
|
|
pauseStatsCollection()
|
|
}
|