Experimental CPU Optimize Techniques

server/services/anomaly.js

@@ -225,6 +225,7 @@ export class AnomalyEngine extends EventEmitter {
   evaluate(batch) {
     /** @type {AnomalyEvent[]} */
     const fired = []
+    if (!this.configs.size) return fired
     const byChart = new Map(batch.map((s) => [s.chart, s]))
 
     for (const id of [...this.configs.keys()]) {

server/services/collector.js

@@ -44,10 +44,13 @@ import {
   ifaceSpeedKbps,
 } from './collectors/host-deep.js'
 import { collectHostMore } from './collectors/host-more.js'
+import { readFileCached, closeFdCache } from '../utils/fd-cache.js'
 import logger from '../utils/logger.js'
 
 const log = logger.child('collector')
 
+const IS_LINUX = os.platform() === 'linux'
+
 function safeProcessUptime() {
   try {
     if (typeof process !== 'undefined' && typeof process.uptime === 'function') {
@@ -66,11 +69,7 @@ function bytesToGiB(n) {
   return n / (1024 * 1024 * 1024)
 }
 function readFile(path) {
-  try {
+  return IS_LINUX ? readFileCached(path) : null
-    return fs.readFileSync(path, 'utf8')
-  } catch {
-    return null
-  }
 }
 function rate(prev, cur, dtSec) {
   if (dtSec <= 0 || cur < prev) return 0
@@ -423,6 +422,7 @@ export class MetricsCollector extends EventEmitter {
     this.deepState = {}
     /** @type {Map<string, Record<string, number|null>>} */
     this.latest = new Map()
+    this._chartsRegistered = false
   }
 
   start() {
@@ -438,6 +438,7 @@ export class MetricsCollector extends EventEmitter {
     this.running = false
     if (this.timer) clearInterval(this.timer)
     this.timer = null
+    if (IS_LINUX) closeFdCache()
   }
 
   _tick() {
@@ -446,9 +447,11 @@ export class MetricsCollector extends EventEmitter {
       const dtSec = this.lastTs ? (ts - this.lastTs) / 1000 : this.intervalMs / 1000
       this.lastTs = ts
       const prevVmSnapshot = this.lastVm
+
       /** @type {Array<{ chart: string, context: string, ts: number, values: Record<string, number|null> }>} */
       const batch = []
 
+      if (IS_LINUX) {
         this._safeCollect('cpu', () => this._collectCpuAndProcs(batch, ts, dtSec))
         this._safeCollect('cpu_sysfs', () => collectCpuSysfs(batch, ts, dtSec, this.deepState))
         this._safeCollect('memory', () => this._collectMemory(batch, ts, dtSec))
@@ -464,14 +467,20 @@ export class MetricsCollector extends EventEmitter {
             prevVm: prevVmSnapshot,
           })
         })
+      } else {
+        this._safeCollect('cpu', () => this._collectCpuFallback(batch, ts, dtSec))
+        this._safeCollect('memory', () => this._collectMemoryFallback(batch, ts))
+      }
 
       let uptime = 0
       try {
         uptime = os.uptime()
       } catch {
+        if (IS_LINUX) {
           const raw = readFile('/proc/uptime')
           if (raw) uptime = Number(raw.trim().split(/\s+/)[0]) || 0
         }
+      }
       batch.push({
         chart: 'system.uptime',
         context: 'system.uptime',
@@ -479,6 +488,7 @@ export class MetricsCollector extends EventEmitter {
         values: { uptime },
       })
 
+      if (IS_LINUX) {
         const entropy = readFile('/proc/sys/kernel/random/entropy_avail')
         if (entropy != null) {
           batch.push({
@@ -488,9 +498,11 @@ export class MetricsCollector extends EventEmitter {
             values: { entropy: Number(entropy.trim()) || 0 },
           })
         }
+      }
 
       for (const s of batch) this.latest.set(s.chart, s.values)
       this.sampleCount++
+      this._chartsRegistered = true
       this.emit('samples', batch)
     } catch (err) {
       log.error('Collector tick failed', { error: err.message })
@@ -505,6 +517,32 @@ export class MetricsCollector extends EventEmitter {
     }
   }
 
+  _collectCpuFallback(batch, ts, dtSec) {
+    const cpus = os.cpus()
+    const load = os.loadavg()
+    const agg = { user: 0, nice: 0, system: 0, idle: 0, irq: 0, iowait: 0, softirq: 0, steal: 0, guest: 0, guest_nice: 0 }
+    for (let i = 0; i < cpus.length; i++) {
+      const c = cpus[i]
+      const cur = { id: String(i), user: c.times.user, nice: c.times.nice, system: c.times.sys, idle: c.times.idle, irq: c.times.irq, iowait: 0, softirq: 0, steal: 0, guest: 0, guest_nice: 0 }
+      if (!this._chartsRegistered) registerChart(makeCpuCoreChart(i))
+      const prev = this.lastStat?.cores?.find((x) => x.id === String(i))
+      batch.push({ chart: `cpu.cpu${i}`, context: 'cpu.cpu', ts, values: cpuDeltaPct(prev, cur) })
+      for (const k of Object.keys(agg)) agg[k] += cur[k] || 0
+    }
+    batch.push({ chart: 'system.cpu', context: 'system.cpu', ts, values: cpuDeltaPct(this.lastStat?.aggregate, agg) })
+    this.lastStat = { aggregate: agg, cores: cpus.map((c, i) => ({ id: String(i), ...c.times, system: c.times.sys })), intr: 0, ctxt: 0, processes: 0, procsRunning: 0, procsBlocked: 0 }
+    batch.push({ chart: 'system.processes', context: 'system.processes', ts, values: { running: load[0] < 0 ? 0 : Math.round(load[0]), blocked: 0 } })
+    batch.push({ chart: 'system.active_processes', context: 'system.active_processes', ts, values: { active: 0 } })
+    batch.push({ chart: 'system.load', context: 'system.load', ts, values: { load1: load[0], load5: load[1], load15: load[2] } })
+  }
+
+  _collectMemoryFallback(batch, ts) {
+    const total = os.totalmem()
+    const free = os.freemem()
+    batch.push({ chart: 'system.ram', context: 'system.ram', ts, values: { free: bytesToMiB(free), used: bytesToMiB(total - free), cached: 0, buffers: 0 } })
+    batch.push({ chart: 'mem.available', context: 'mem.available', ts, values: { avail: bytesToMiB(free) } })
+  }
+
   _collectCpuAndProcs(batch, ts, dtSec) {
     const stat = parseProcStat()
     const load = parseLoadavg()
@@ -514,7 +552,7 @@ export class MetricsCollector extends EventEmitter {
       batch.push({ chart: 'system.cpu', context: 'system.cpu', ts, values: cpuVals })
 
       for (const core of stat.cores) {
-        registerChart(makeCpuCoreChart(core.id))
+        if (!this._chartsRegistered) registerChart(makeCpuCoreChart(core.id))
         const prev = this.lastStat?.cores?.find((c) => c.id === core.id)
         batch.push({
           chart: `cpu.cpu${core.id}`,
@@ -764,6 +802,7 @@ export class MetricsCollector extends EventEmitter {
 
     for (const [name, d] of Object.entries(disks)) {
       if (name === '__agg') continue
+      if (!this._chartsRegistered) {
         registerChart(makeDiskIoChart(name))
         registerChart(makeDiskOpsChart(name))
         registerChart(makeDiskUtilChart(name))
@@ -776,6 +815,7 @@ export class MetricsCollector extends EventEmitter {
         registerChart(makeDiskSvctmChart(name))
         registerChart(makeDiskMergedChart(name))
         registerChart(makeDiskFlushChart(name))
+      }
       const prev = prevAll[name]
       const readBytes = d.readSectors * 512
       const writeBytes = d.writeSectors * 512
@@ -911,6 +951,7 @@ export class MetricsCollector extends EventEmitter {
 
     for (const [name, n] of Object.entries(ifaces)) {
       if (name === '__agg') continue
+      if (!this._chartsRegistered) {
         registerChart(makeNetChart(name))
         registerChart(makeNetPacketsChart(name))
         registerChart(makeNetErrorsChart(name))
@@ -919,6 +960,7 @@ export class MetricsCollector extends EventEmitter {
         registerChart(makeNetDuplexChart(name))
         registerChart(makeNetMtuChart(name))
         registerChart(makeNetQueueChart(name))
+      }
       const prev = prevAll[name]
       rx += n.rxBytes
       tx += n.txBytes
@@ -1189,8 +1231,10 @@ export class MetricsCollector extends EventEmitter {
       const id = mountId(m.path)
       const sample = sampleMount(m.path)
       if (!sample) continue
+      if (!this._chartsRegistered) {
         registerChart(makeDiskSpaceChart(id, m.path))
         registerChart(makeDiskInodesChart(id, m.path))
+      }
       batch.push({
         chart: `disk_space.${id}`,
         context: 'disk.space',

server/services/collectors/host-deep.js

@@ -6,13 +6,10 @@ import fs from 'fs'
 import os from 'os'
 import path from 'path'
 import { registerChart } from '../../../shared/metrics.js'
+import { readFileCached } from '../../utils/fd-cache.js'
 
 function readFile(p) {
-  try {
+  return readFileCached(p)
-    return fs.readFileSync(p, 'utf8')
-  } catch {
-    return null
-  }
 }
 
 function rate(prev, cur, dtSec) {

server/services/collectors/host-more.js

@@ -11,13 +11,10 @@ import {
   makeCpuIdleChart,
   makeNumaNodeMemChart,
 } from '../../../shared/metrics.js'
+import { readFileCached } from '../../utils/fd-cache.js'
 
 function readFile(p) {
-  try {
+  return readFileCached(p)
-    return fs.readFileSync(p, 'utf8')
-  } catch {
-    return null
-  }
 }
 
 function rate(prev, cur, dtSec) {

server/services/collectors/self-monitor.js

@@ -58,9 +58,11 @@ export function isSelfMonitorEnabled() {
   return true
 }
 
+const SELF_MONITOR_INTERVAL = Number(process.env.PEARDATA_SELF_MONITOR_MS) || 5000
+
 export class SelfMonitorCollector extends CollectorPlugin {
   constructor(opts = {}) {
-    super({ name: 'self-monitor', intervalMs: opts.intervalMs })
+    super({ name: 'self-monitor', intervalMs: opts.intervalMs || SELF_MONITOR_INTERVAL })
     /** @type {{ user: number, system: number }|null} */
     this._prevCpu = null
     /** @type {number|null} */

server/services/store.js

@@ -102,7 +102,7 @@ export class MetricStore extends EventEmitter {
         this.series.set(s.chart, entry)
       }
       entry.points.push({ ts: s.ts, values: s.values })
-      if (entry.points.length > this.tier0Max) {
+      if (entry.points.length > this.tier0Max + 60) {
         entry.points.splice(0, entry.points.length - this.tier0Max)
       }
 
@@ -125,7 +125,7 @@ export class MetricStore extends EventEmitter {
         }
         const warm = { ts: s.ts, values: avg }
         entry.tier1.push(warm)
-        if (entry.tier1.length > this.tier1Max) {
+        if (entry.tier1.length > this.tier1Max + 30) {
           entry.tier1.splice(0, entry.tier1.length - this.tier1Max)
         }
         entry.acc = null

server/services/subscriptions.js

@@ -46,7 +46,9 @@ export function unsubscribeAnomalies(session) {
  * @param {Array<{ chart: string, context: string, ts: number, values: object }>} batch
  */
 export function broadcastMetrics(batch) {
-  for (const session of peers.list()) {
+  const list = peers.list()
+  if (!list.length) return
+  for (const session of list) {
     const sub = session.state.get('metricSub')
     if (!sub || session.closed) continue
     const filtered =
@@ -73,7 +75,9 @@ export function broadcastMetrics(batch) {
  * @param {object} anomaly
  */
 export function broadcastAnomaly(anomaly) {
-  for (const session of peers.list()) {
+  const list = peers.list()
+  if (!list.length) return
+  for (const session of list) {
     if (!session.state.get('anomalySub') || session.closed) continue
     try {
       session.push(Pushes.anomaly, anomaly)
@@ -87,7 +91,9 @@ export function broadcastAnomaly(anomaly) {
  * @param {object} payload
  */
 export function broadcastHealth(payload) {
-  for (const session of peers.list()) {
+  const list = peers.list()
+  if (!list.length) return
+  for (const session of list) {
     if (session.closed) continue
     try {
       session.push(Pushes.health, payload)

server/utils/fd-cache.js

@@ -0,0 +1,34 @@
+import fs from 'fs'
+
+const READ_BUF = Buffer.alloc(65536)
+const fdCache = new Map()
+
+export function readFileCached(path) {
+  let entry = fdCache.get(path)
+  if (entry === undefined) {
+    try {
+      const fd = fs.openSync(path, 'r')
+      entry = { fd }
+      fdCache.set(path, entry)
+    } catch {
+      fdCache.set(path, null)
+      return null
+    }
+  }
+  if (!entry) return null
+  try {
+    const bytesRead = fs.readSync(entry.fd, READ_BUF, 0, READ_BUF.length, 0)
+    return READ_BUF.toString('utf8', 0, bytesRead)
+  } catch {
+    return null
+  }
+}
+
+export function closeFdCache() {
+  for (const [path, entry] of fdCache) {
+    if (entry) {
+      try { fs.closeSync(entry.fd) } catch {}
+    }
+  }
+  fdCache.clear()
+}