Updates

server/services/collectors/peardock.js

@@ -0,0 +1,239 @@
+/**
+ * PearDock bridge collector (Phase 3 spike).
+ *
+ * Dials dock / agent peers and remaps their docker/container charts into
+ * peardock.* — no PearDock source copied (AGPL boundary).
+ *
+ * Enable: PEARDATA_PEARDOCK=1
+ * Peers:  PEARDATA_PEARDOCK_PEERS=hex,hex  (or PEARDATA_PEARDOCK_PEERS_FILE)
+ * Optional: PEARDATA_PEARDOCK_SEED, PEARDATA_PEARDOCK_POLL_MS
+ *
+ * Charts: peardock.containers, peardock.cpu.<id>, peardock.mem.<id>
+ *         (sourced from remote docker.* charts when present)
+ */
+import fs from 'fs'
+import { EventEmitter } from 'events'
+import { PearDataConnection } from '../../../client/connection.js'
+import { Methods } from '../../../shared/protocol.js'
+import { registerChart } from '../../../shared/metrics.js'
+import logger from '../../utils/logger.js'
+
+const log = logger.child('peardock')
+
+export function isPearDockEnabled() {
+  const v = process.env.PEARDATA_PEARDOCK
+  return v === '1' || v === 'on' || v === 'true'
+}
+
+/**
+ * @returns {string[]}
+ */
+export function parsePearDockPeers() {
+  const raw = process.env.PEARDATA_PEARDOCK_PEERS || ''
+  const fromEnv = raw
+    .split(/[,\s]+/)
+    .map((s) => s.trim().toLowerCase())
+    .filter((s) => /^[0-9a-f]{64}$/.test(s))
+  const file = process.env.PEARDATA_PEARDOCK_PEERS_FILE
+  if (!file) return [...new Set(fromEnv)]
+  try {
+    const text = fs.readFileSync(file, 'utf8')
+    const fromFile = text
+      .split(/\r?\n/)
+      .map((l) => l.replace(/#.*$/, '').trim().toLowerCase())
+      .filter((s) => /^[0-9a-f]{64}$/.test(s))
+    return [...new Set([...fromEnv, ...fromFile])]
+  } catch {
+    return [...new Set(fromEnv)]
+  }
+}
+
+/**
+ * Map remote docker/container chart id → peardock chart id.
+ * @param {string} chartId
+ * @returns {string|null}
+ */
+export function remapDockChart(chartId) {
+  const id = String(chartId || '')
+  if (id.startsWith('docker.')) return `peardock.${id.slice('docker.'.length)}`
+  if (id.startsWith('container.')) return `peardock.${id.slice('container.'.length)}`
+  if (id.startsWith('cgroup.docker.')) return `peardock.${id.slice('cgroup.docker.'.length)}`
+  return null
+}
+
+/**
+ * @param {any} metrics getAllMetrics-style payload
+ * @returns {Array<{ chart: string, context: string, values: Record<string, number|null>, sourceChart: string }>}
+ */
+export function extractDockCharts(metrics) {
+  const root = metrics?.body || metrics
+  const charts = root?.charts || {}
+  /** @type {Array<{ chart: string, context: string, values: Record<string, number|null>, sourceChart: string }>} */
+  const out = []
+  for (const [chartId, point] of Object.entries(charts)) {
+    const mapped = remapDockChart(chartId)
+    if (!mapped) continue
+    const values = point?.dimensions || point?.values || {}
+    /** @type {Record<string, number|null>} */
+    const nums = {}
+    for (const [k, v] of Object.entries(values)) {
+      const n = typeof v === 'object' && v != null ? Number(v.value) : Number(v)
+      nums[k] = Number.isFinite(n) ? n : null
+    }
+    const parts = mapped.split('.')
+    const context = parts.length >= 2 ? `${parts[0]}.${parts[1]}` : mapped
+    out.push({
+      chart: mapped,
+      context,
+      values: nums,
+      sourceChart: chartId,
+    })
+  }
+  return out
+}
+
+function registerMappedChart(mappedId, context, values) {
+  const dims = Object.keys(values).map((id) => ({
+    id,
+    name: id,
+    algorithm: 'absolute',
+  }))
+  if (!dims.length) dims.push({ id: 'value', name: 'value', algorithm: 'absolute' })
+  registerChart({
+    id: mappedId,
+    name: mappedId,
+    context,
+    title: mappedId,
+    units: context.includes('mem') ? 'MiB' : context.includes('cpu') ? 'percentage' : 'count',
+    family: 'peardock',
+    chartType: 'line',
+    priority: 8500,
+    plugin: 'peardock',
+    dimensions: dims,
+  })
+}
+
+export class PearDockCollector extends EventEmitter {
+  constructor(opts = {}) {
+    super()
+    this.pollMs = opts.pollMs || Number(process.env.PEARDATA_PEARDOCK_POLL_MS) || 5000
+    this.adminSeed = opts.adminSeed || process.env.PEARDATA_PEARDOCK_SEED || null
+    this.peerKeys = opts.peers || parsePearDockPeers()
+    /** @type {Map<string, PearDataConnection|null>} */
+    this.conns = new Map()
+    this._timer = null
+    this._busy = false
+  }
+
+  start() {
+    if (this._timer) return
+    registerChart({
+      id: 'peardock.containers',
+      name: 'peardock.containers',
+      context: 'peardock.containers',
+      title: 'PearDock bridged containers',
+      units: 'containers',
+      family: 'peardock',
+      chartType: 'line',
+      priority: 8490,
+      plugin: 'peardock',
+      dimensions: [
+        { id: 'charts', name: 'charts', algorithm: 'absolute' },
+        { id: 'peers', name: 'peers', algorithm: 'absolute' },
+      ],
+    })
+    log.info('PearDock bridge started', { peers: this.peerKeys.length, pollMs: this.pollMs })
+    this._tick()
+    this._timer = setInterval(() => this._tick(), this.pollMs)
+    if (typeof this._timer.unref === 'function') this._timer.unref()
+  }
+
+  stop() {
+    if (this._timer) {
+      clearInterval(this._timer)
+      this._timer = null
+    }
+    for (const [pk, conn] of this.conns) {
+      if (conn) conn.destroy().catch(() => {})
+      this.conns.set(pk, null)
+    }
+  }
+
+  async _ensure(pk) {
+    let conn = this.conns.get(pk)
+    if (conn?.connected) return conn
+    try {
+      if (conn) await conn.destroy().catch(() => {})
+      conn = new PearDataConnection(pk, {
+        adminSeed: this.adminSeed,
+        timeoutMs: 20_000,
+      })
+      await conn.connect()
+      this.conns.set(pk, conn)
+      conn.on('disconnected', () => this.conns.set(pk, null))
+      return conn
+    } catch (err) {
+      this.conns.set(pk, null)
+      log.warn('PearDock dial failed', { peer: pk.slice(0, 12), error: err.message })
+      return null
+    }
+  }
+
+  async _tick() {
+    if (this._busy) return
+    this._busy = true
+    try {
+      const ts = Date.now()
+      /** @type {Array<{ chart: string, context: string, ts: number, values: Record<string, number|null> }>} */
+      const batch = []
+      let peersUp = 0
+      let chartCount = 0
+
+      for (const pk of this.peerKeys) {
+        const conn = await this._ensure(pk)
+        if (!conn) continue
+        try {
+          const metrics = await conn.request(Methods.getAllMetrics, { format: 'json' })
+          const mapped = extractDockCharts(metrics)
+          peersUp++
+          for (const row of mapped) {
+            registerMappedChart(row.chart, row.context, row.values)
+            batch.push({
+              chart: row.chart,
+              context: row.context,
+              ts,
+              values: row.values,
+            })
+            chartCount++
+          }
+        } catch (err) {
+          log.warn('PearDock poll failed', { peer: pk.slice(0, 12), error: err.message })
+          try {
+            await conn.destroy()
+          } catch {
+            // ignore
+          }
+          this.conns.set(pk, null)
+        }
+      }
+
+      batch.unshift({
+        chart: 'peardock.containers',
+        context: 'peardock.containers',
+        ts,
+        values: { charts: chartCount, peers: peersUp },
+      })
+      this.emit('samples', batch)
+    } finally {
+      this._busy = false
+    }
+  }
+}
+
+/** @type {PearDockCollector|null} */
+let singleton = null
+
+export function getPearDockCollector() {
+  if (!singleton) singleton = new PearDockCollector()
+  return singleton
+}