Raven Scott
af6609c2d3
Cache cpuidle names (was re-reading name sysfs every tick × cores) Cache CPU sysfs dir list + register cpufreq/throttle once Map for per-core prev lookups (was O(n²) .find) Dedupe meminfo/vmstat once per tick Cache iface speed/duplex/mtu/qlen (30s) Cache inotify limits (30s) Gate registerChart for IRQs, cpuidle, peardock mapped charts Remove leftover profile() stderr timers Cache os.cpus().length in docker collector Grow fd-cache buffer past 64KB for /proc/interrupts etc.
570 lines
16 KiB
JavaScript
570 lines
16 KiB
JavaScript
/**
|
|
* Opt-in Docker / container collector (Phase 3 spike).
|
|
*
|
|
* Enable: PEARDATA_DOCKER=1
|
|
* Discovers containers via cgroup v2 scopes (docker-*.scope / libpod-*.scope)
|
|
* and optionally Docker Engine API over a unix socket for names.
|
|
*
|
|
* Emits:
|
|
* docker.containers — running / total counts
|
|
* docker.cpu.<shortId> — % of one host CPU
|
|
* docker.mem.<shortId> — usage / limit MiB
|
|
*/
|
|
import fs from 'fs'
|
|
import path from 'path'
|
|
import net from 'net'
|
|
import os from 'os'
|
|
import { EventEmitter } from 'events'
|
|
import { execFile } from '../../utils/exec.js'
|
|
import { readFileBuf } from '../../utils/fd-cache.js'
|
|
import {
|
|
SAMPLE_INTERVAL_MS,
|
|
registerChart,
|
|
DOCKER_CONTAINERS_CHART,
|
|
makeDockerCpuChart,
|
|
makeDockerMemChart,
|
|
} from '../../../shared/metrics.js'
|
|
import {
|
|
pickContainerDisplayName,
|
|
resolveContainerLabel,
|
|
indexContainerName,
|
|
} from '../../../shared/container-names.js'
|
|
import logger from '../../utils/logger.js'
|
|
|
|
const log = logger.child('docker')
|
|
|
|
export function isDockerCollectorEnabled() {
|
|
const v = process.env.PEARDATA_DOCKER
|
|
return v === '1' || v === 'on' || v === 'true'
|
|
}
|
|
|
|
/** Candidate Docker / Podman engine sockets (first existing wins). */
|
|
export function resolveDockerSocket(preferred) {
|
|
const candidates = [
|
|
preferred,
|
|
process.env.PEARDATA_DOCKER_SOCKET,
|
|
'/var/run/docker.sock',
|
|
'/run/docker.sock',
|
|
'/var/run/podman/podman.sock',
|
|
'/run/podman/podman.sock',
|
|
].filter(Boolean)
|
|
for (const p of candidates) {
|
|
try {
|
|
if (fs.existsSync(p)) return p
|
|
} catch {
|
|
// ignore
|
|
}
|
|
}
|
|
// Prefer a real default path when nothing exists yet (docker not started)
|
|
return process.env.PEARDATA_DOCKER_SOCKET || '/var/run/docker.sock'
|
|
}
|
|
|
|
function readFile(p) {
|
|
return readFileBuf(p)
|
|
}
|
|
|
|
function bytesToMiB(n) {
|
|
return n / (1024 * 1024)
|
|
}
|
|
|
|
const HOST_CPUS = (() => {
|
|
try {
|
|
return os.cpus().length || 1
|
|
} catch {
|
|
return 1
|
|
}
|
|
})()
|
|
|
|
function hostCpus() {
|
|
return HOST_CPUS
|
|
}
|
|
|
|
/**
|
|
* @returns {Array<{ id: string, shortId: string, name: string, cgroupPath: string }>}
|
|
*/
|
|
export function discoverCgroupContainers() {
|
|
const roots = [
|
|
'/sys/fs/cgroup/system.slice',
|
|
'/sys/fs/cgroup/docker',
|
|
'/sys/fs/cgroup',
|
|
]
|
|
/** @type {Map<string, { id: string, shortId: string, name: string, cgroupPath: string }>} */
|
|
const found = new Map()
|
|
|
|
for (const root of roots) {
|
|
let entries
|
|
try {
|
|
entries = fs.readdirSync(root, { withFileTypes: true })
|
|
} catch {
|
|
continue
|
|
}
|
|
for (const ent of entries) {
|
|
if (!ent.isDirectory()) continue
|
|
const name = ent.name
|
|
let id = null
|
|
let label = name
|
|
const dockerScope = name.match(/^docker-([0-9a-f]{12,64})\.scope$/i)
|
|
const podmanScope = name.match(/^libpod-([0-9a-f]{12,64})\.scope$/i)
|
|
if (dockerScope) {
|
|
id = dockerScope[1].toLowerCase()
|
|
label = id.slice(0, 12)
|
|
} else if (podmanScope) {
|
|
id = podmanScope[1].toLowerCase()
|
|
label = id.slice(0, 12)
|
|
} else if (/^[0-9a-f]{64}$/i.test(name) && root.endsWith('/docker')) {
|
|
id = name.toLowerCase()
|
|
label = id.slice(0, 12)
|
|
}
|
|
if (!id || found.has(id)) continue
|
|
const cgroupPath = path.join(root, name)
|
|
if (
|
|
!readFile(path.join(cgroupPath, 'cpu.stat')) &&
|
|
!readFile(path.join(cgroupPath, 'cpuacct.usage'))
|
|
) {
|
|
continue
|
|
}
|
|
found.set(id, {
|
|
id,
|
|
shortId: id.slice(0, 12),
|
|
name: label,
|
|
cgroupPath,
|
|
})
|
|
}
|
|
}
|
|
return [...found.values()]
|
|
}
|
|
|
|
/**
|
|
* @param {string} cgroupPath
|
|
* @returns {{ usageNs: number }|null}
|
|
*/
|
|
export function readCgroupCpu(cgroupPath) {
|
|
const raw = readFile(path.join(cgroupPath, 'cpu.stat'))
|
|
if (raw) {
|
|
/** @type {Record<string, number>} */
|
|
const m = {}
|
|
for (const line of raw.split('\n')) {
|
|
const [k, v] = line.trim().split(/\s+/)
|
|
if (k && v != null) m[k] = Number(v)
|
|
}
|
|
if (m.usage_usec != null) {
|
|
return { usageNs: m.usage_usec * 1000 }
|
|
}
|
|
}
|
|
const acct = readFile(path.join(cgroupPath, 'cpuacct.usage'))
|
|
if (acct) {
|
|
const usageNs = Number(acct.trim())
|
|
if (Number.isFinite(usageNs)) return { usageNs }
|
|
}
|
|
return null
|
|
}
|
|
|
|
/**
|
|
* @param {string} cgroupPath
|
|
* @returns {{ usage: number, limit: number }|null} bytes
|
|
*/
|
|
export function readCgroupMemory(cgroupPath) {
|
|
const current = readFile(path.join(cgroupPath, 'memory.current'))
|
|
if (current) {
|
|
const usage = Number(current.trim())
|
|
let limit = Number(readFile(path.join(cgroupPath, 'memory.max'))?.trim())
|
|
if (!Number.isFinite(limit) || limit <= 0 || limit > 1e15) limit = 0
|
|
if (Number.isFinite(usage)) return { usage, limit }
|
|
}
|
|
const usageFile = readFile(path.join(cgroupPath, 'memory.usage_in_bytes'))
|
|
if (usageFile) {
|
|
const usage = Number(usageFile.trim())
|
|
const limitRaw = readFile(path.join(cgroupPath, 'memory.limit_in_bytes'))
|
|
let limit = limitRaw ? Number(limitRaw.trim()) : 0
|
|
if (limit > 1e15) limit = 0
|
|
if (Number.isFinite(usage)) return { usage, limit }
|
|
}
|
|
return null
|
|
}
|
|
|
|
/**
|
|
* Parse Docker Engine `/containers/json` payload into id → display name.
|
|
* Exported for tests.
|
|
* @param {any[]} list
|
|
* @returns {Map<string, string>}
|
|
*/
|
|
export function mapDockerContainerNames(list) {
|
|
/** @type {Map<string, string>} */
|
|
const map = new Map()
|
|
if (!Array.isArray(list)) return map
|
|
for (const c of list) {
|
|
const id = String(c.Id || '').toLowerCase()
|
|
if (!id) continue
|
|
const name = pickContainerDisplayName(c)
|
|
indexContainerName(map, id, name)
|
|
for (const n of c.Names || []) {
|
|
const bare = String(n || '')
|
|
.replace(/^\//, '')
|
|
.trim()
|
|
.toLowerCase()
|
|
if (bare) map.set(bare, name)
|
|
}
|
|
}
|
|
return map
|
|
}
|
|
|
|
/**
|
|
* GET over a Unix domain socket (works under Node + Bare; bare-http1 has no socketPath).
|
|
* @param {string} socketPath
|
|
* @param {string} urlPath
|
|
* @param {number} [timeoutMs]
|
|
* @returns {Promise<string>} response body
|
|
*/
|
|
export function httpGetUnix(socketPath, urlPath, timeoutMs = 3000) {
|
|
return new Promise((resolve, reject) => {
|
|
// String form is reliable for bare-net → bare-pipe unix sockets
|
|
const socket = net.connect(socketPath)
|
|
let buf = ''
|
|
let settled = false
|
|
const finish = (err, body) => {
|
|
if (settled) return
|
|
settled = true
|
|
clearTimeout(timer)
|
|
try {
|
|
socket.destroy()
|
|
} catch {
|
|
// ignore
|
|
}
|
|
if (err) reject(err)
|
|
else resolve(body)
|
|
}
|
|
const timer = setTimeout(() => {
|
|
finish(new Error(`unix http timeout after ${timeoutMs}ms`))
|
|
}, timeoutMs)
|
|
socket.on('connect', () => {
|
|
socket.write(
|
|
`GET ${urlPath} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\nAccept: application/json\r\n\r\n`
|
|
)
|
|
})
|
|
socket.on('data', (chunk) => {
|
|
buf += typeof chunk === 'string' ? chunk : Buffer.from(chunk).toString('utf8')
|
|
})
|
|
socket.on('end', () => {
|
|
try {
|
|
finish(null, parseHttpBody(buf))
|
|
} catch (err) {
|
|
finish(err)
|
|
}
|
|
})
|
|
socket.on('error', (err) => finish(err))
|
|
})
|
|
}
|
|
|
|
/**
|
|
* @param {string} raw
|
|
*/
|
|
export function parseHttpBody(raw) {
|
|
let sep = raw.indexOf('\r\n\r\n')
|
|
let hdrEnd = 4
|
|
if (sep < 0) {
|
|
sep = raw.indexOf('\n\n')
|
|
hdrEnd = 2
|
|
}
|
|
if (sep < 0) throw new Error('invalid http response')
|
|
const header = raw.slice(0, sep)
|
|
let body = raw.slice(sep + hdrEnd)
|
|
const status = header.split(/\r?\n/)[0] || ''
|
|
if (!/\s200(\s|$)/.test(status)) {
|
|
throw new Error(`http ${status || 'error'}`)
|
|
}
|
|
if (/transfer-encoding:\s*chunked/i.test(header)) {
|
|
body = decodeChunked(body)
|
|
}
|
|
return body
|
|
}
|
|
|
|
function decodeChunked(body) {
|
|
let out = ''
|
|
let i = 0
|
|
while (i < body.length) {
|
|
const nl = body.indexOf('\r\n', i)
|
|
if (nl < 0) break
|
|
const size = parseInt(body.slice(i, nl), 16)
|
|
if (!Number.isFinite(size) || size <= 0) break
|
|
const start = nl + 2
|
|
out += body.slice(start, start + size)
|
|
i = start + size + 2
|
|
}
|
|
return out
|
|
}
|
|
|
|
/**
|
|
* Docker Engine API via unix socket.
|
|
* @param {string} socketPath
|
|
* @returns {Promise<Map<string, string>>}
|
|
*/
|
|
export async function fetchDockerNames(socketPath) {
|
|
const sock = resolveDockerSocket(socketPath)
|
|
const paths = [
|
|
'/containers/json?all=true',
|
|
'/v1.41/containers/json?all=true',
|
|
'/v1.44/containers/json?all=true',
|
|
]
|
|
let lastErr = null
|
|
for (const urlPath of paths) {
|
|
try {
|
|
const body = await httpGetUnix(sock, urlPath)
|
|
const map = mapDockerContainerNames(JSON.parse(body))
|
|
if (map.size) return map
|
|
} catch (err) {
|
|
lastErr = err
|
|
}
|
|
}
|
|
if (lastErr) throw lastErr
|
|
return new Map()
|
|
}
|
|
|
|
/**
|
|
* Parse `docker ps` TSV lines into a name map. Exported for tests.
|
|
* @param {string} stdout
|
|
* @returns {Map<string, string>}
|
|
*/
|
|
export function parseDockerPsNames(stdout) {
|
|
/** @type {Map<string, string>} */
|
|
const map = new Map()
|
|
for (const line of String(stdout || '').split('\n')) {
|
|
const tab = line.indexOf('\t')
|
|
if (tab < 0) continue
|
|
const id = line.slice(0, tab).trim().toLowerCase()
|
|
const names = line
|
|
.slice(tab + 1)
|
|
.trim()
|
|
.split(',')
|
|
.map((n) => n.trim())
|
|
.filter(Boolean)
|
|
if (!id || !names.length) continue
|
|
const name = pickContainerDisplayName({ Id: id, Names: names })
|
|
indexContainerName(map, id, name)
|
|
for (const n of names) map.set(n.replace(/^\//, '').toLowerCase(), name)
|
|
}
|
|
return map
|
|
}
|
|
|
|
/**
|
|
* `docker ps` CLI fallback (Bare-safe via server/utils/exec.js).
|
|
* @returns {Promise<Map<string, string>>}
|
|
*/
|
|
export async function fetchDockerNamesCli() {
|
|
try {
|
|
const { stdout } = await execFile(
|
|
'docker',
|
|
['ps', '-a', '--no-trunc', '--format', '{{.ID}}\t{{.Names}}'],
|
|
{ timeout: 5000, maxBuffer: 8 * 1024 * 1024 }
|
|
)
|
|
return parseDockerPsNames(stdout)
|
|
} catch {
|
|
return new Map()
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Read Names from Docker's on-disk config.v2.json (no socket needed if readable).
|
|
* @returns {Map<string, string>}
|
|
*/
|
|
export function readDockerNamesFromFs() {
|
|
/** @type {Map<string, string>} */
|
|
const map = new Map()
|
|
const roots = [
|
|
process.env.PEARDATA_DOCKER_ROOT,
|
|
'/var/lib/docker/containers',
|
|
path.join(os.homedir(), '.local/share/docker/containers'),
|
|
].filter(Boolean)
|
|
|
|
for (const root of roots) {
|
|
let ents
|
|
try {
|
|
ents = fs.readdirSync(root)
|
|
} catch {
|
|
continue
|
|
}
|
|
for (const id of ents) {
|
|
if (!/^[0-9a-f]{64}$/i.test(id)) continue
|
|
const raw = readFile(path.join(root, id, 'config.v2.json'))
|
|
if (!raw) continue
|
|
try {
|
|
const cfg = JSON.parse(raw)
|
|
const labels = cfg.Config?.Labels || cfg.Labels || {}
|
|
const name = pickContainerDisplayName({
|
|
Id: id,
|
|
Names: cfg.Name ? [cfg.Name] : [],
|
|
Labels: labels,
|
|
Image: cfg.Config?.Image || cfg.Image || '',
|
|
})
|
|
indexContainerName(map, id, name)
|
|
if (cfg.Name) {
|
|
map.set(String(cfg.Name).replace(/^\//, '').toLowerCase(), name)
|
|
}
|
|
} catch {
|
|
// ignore bad configs
|
|
}
|
|
}
|
|
}
|
|
return map
|
|
}
|
|
|
|
/**
|
|
* Best-effort name map: Engine API → docker CLI → filesystem.
|
|
* @param {string} [socketPath]
|
|
* @returns {Promise<{ map: Map<string, string>, source: string }>}
|
|
*/
|
|
export async function loadContainerNameMap(socketPath) {
|
|
const sock = resolveDockerSocket(socketPath)
|
|
if (fs.existsSync(sock)) {
|
|
try {
|
|
const map = await fetchDockerNames(sock)
|
|
if (map.size) return { map, source: 'docker-api' }
|
|
} catch (err) {
|
|
log.warn('Docker API name fetch failed', { socket: sock, error: err.message })
|
|
}
|
|
}
|
|
|
|
try {
|
|
const map = await fetchDockerNamesCli()
|
|
if (map.size) return { map, source: 'docker-cli' }
|
|
} catch (err) {
|
|
log.warn('docker CLI name fetch failed', { error: err.message })
|
|
}
|
|
|
|
const fsMap = readDockerNamesFromFs()
|
|
if (fsMap.size) return { map: fsMap, source: 'docker-fs' }
|
|
|
|
return { map: new Map(), source: 'none' }
|
|
}
|
|
|
|
export class DockerCollector extends EventEmitter {
|
|
constructor(opts = {}) {
|
|
super()
|
|
this.intervalMs = opts.intervalMs || Number(process.env.PEARDATA_SAMPLE_MS) || SAMPLE_INTERVAL_MS
|
|
this.socketPath = resolveDockerSocket(opts.socketPath)
|
|
this._timer = null
|
|
/** @type {Map<string, { usageNs: number, wallMs: number }>} */
|
|
this._prevCpu = new Map()
|
|
/** @type {Map<string, string>} */
|
|
this._names = new Map()
|
|
this._nameRefreshAt = 0
|
|
}
|
|
|
|
start() {
|
|
if (this._timer) return
|
|
this.socketPath = resolveDockerSocket(this.socketPath)
|
|
registerChart(DOCKER_CONTAINERS_CHART)
|
|
log.info('Docker collector started', { socket: this.socketPath })
|
|
this._tick()
|
|
this._timer = setInterval(() => this._tick(), this.intervalMs)
|
|
if (typeof this._timer.unref === 'function') this._timer.unref()
|
|
}
|
|
|
|
stop() {
|
|
if (this._timer) {
|
|
clearInterval(this._timer)
|
|
this._timer = null
|
|
}
|
|
}
|
|
|
|
async _refreshNames() {
|
|
const now = Date.now()
|
|
// Retry sooner while empty so a late socket/group fix shows names quickly
|
|
const interval = this._names.size ? 30_000 : 5_000
|
|
if (now - this._nameRefreshAt < interval) return
|
|
this._nameRefreshAt = now
|
|
try {
|
|
this.socketPath = resolveDockerSocket(this.socketPath)
|
|
const { map, source } = await loadContainerNameMap(this.socketPath)
|
|
if (map.size) {
|
|
const prev = this._names.size
|
|
this._names = map
|
|
if (prev !== map.size) {
|
|
log.info('Docker container names loaded', {
|
|
count: map.size,
|
|
source,
|
|
socket: this.socketPath,
|
|
})
|
|
}
|
|
} else if (!this._names.size) {
|
|
log.warn(
|
|
'No Docker container names yet — check peardata ∈ docker group, PEARDATA_DOCKER=1, and socket path',
|
|
{ socket: this.socketPath, source }
|
|
)
|
|
}
|
|
} catch (err) {
|
|
log.warn('Docker name refresh failed', { error: err.message, socket: this.socketPath })
|
|
}
|
|
}
|
|
|
|
async _tick() {
|
|
try {
|
|
await this._refreshNames()
|
|
const containers = discoverCgroupContainers()
|
|
const ts = Date.now()
|
|
const wallMs = ts
|
|
const ncpu = hostCpus()
|
|
/** @type {Array<{ chart: string, context: string, ts: number, values: Record<string, number|null> }>} */
|
|
const batch = [
|
|
{
|
|
chart: 'docker.containers',
|
|
context: 'docker.containers',
|
|
ts,
|
|
values: { running: containers.length, total: containers.length },
|
|
},
|
|
]
|
|
|
|
for (const c of containers) {
|
|
const display = resolveContainerLabel(c.id, this._names)
|
|
const cpuDef = makeDockerCpuChart(c.shortId, display)
|
|
const memDef = makeDockerMemChart(c.shortId, display)
|
|
registerChart(cpuDef)
|
|
registerChart(memDef)
|
|
|
|
const cpu = readCgroupCpu(c.cgroupPath)
|
|
let usagePct = 0
|
|
if (cpu) {
|
|
const prev = this._prevCpu.get(c.id)
|
|
if (prev && wallMs > prev.wallMs) {
|
|
const dNs = cpu.usageNs - prev.usageNs
|
|
const dWallNs = (wallMs - prev.wallMs) * 1e6
|
|
if (dNs >= 0 && dWallNs > 0) {
|
|
usagePct = Math.min(100 * ncpu, (dNs / dWallNs) * 100)
|
|
}
|
|
}
|
|
this._prevCpu.set(c.id, { usageNs: cpu.usageNs, wallMs })
|
|
}
|
|
|
|
batch.push({
|
|
chart: cpuDef.id,
|
|
context: 'docker.cpu',
|
|
ts,
|
|
values: { usage: usagePct },
|
|
})
|
|
|
|
const mem = readCgroupMemory(c.cgroupPath)
|
|
batch.push({
|
|
chart: memDef.id,
|
|
context: 'docker.mem',
|
|
ts,
|
|
values: {
|
|
usage: mem ? bytesToMiB(mem.usage) : 0,
|
|
limit: mem && mem.limit ? bytesToMiB(mem.limit) : null,
|
|
},
|
|
})
|
|
}
|
|
|
|
this.emit('samples', batch)
|
|
} catch (err) {
|
|
log.warn('Docker tick failed', { error: err.message })
|
|
}
|
|
}
|
|
}
|
|
|
|
/** @type {DockerCollector|null} */
|
|
let singleton = null
|
|
|
|
export function getDockerCollector() {
|
|
if (!singleton) singleton = new DockerCollector()
|
|
return singleton
|
|
}
|