peardata/server/pipeline.js

/**
 * Agent data pipeline: collector → memory store → anomaly → push
 *                     ↘ warm HyperDB flush
 *                     (+ optional Docker collector)
 */
import os from 'os'
import { getCollector } from './services/collector.js'
import {
  getDockerCollector,
  isDockerCollectorEnabled,
} from './services/collectors/docker.js'
import {
  getProcessCollector,
  isProcessCollectorEnabled,
} from './services/collectors/processes.js'
import {
  getParentCollector,
  isParentEnabled,
} from './services/collectors/parent.js'
import {
  getNginxCollector,
  isNginxEnabled,
} from './services/collectors/nginx.js'
import {
  getRedisCollector,
  isRedisEnabled,
} from './services/collectors/redis.js'
import {
  getPostgresCollector,
  isPostgresEnabled,
} from './services/collectors/postgres.js'
import {
  getPearDockCollector,
  isPearDockEnabled,
} from './services/collectors/peardock.js'
import {
  getCgroupsCollector,
  isCgroupsEnabled,
} from './services/collectors/cgroups.js'
import {
  getSensorsCollector,
  isSensorsEnabled,
} from './services/collectors/sensors.js'
import {
  getMysqlCollector,
  isMysqlEnabled,
} from './services/collectors/mysql.js'
import {
  getApacheCollector,
  isApacheEnabled,
} from './services/collectors/apache.js'
import {
  getMemcachedCollector,
  isMemcachedEnabled,
} from './services/collectors/memcached.js'
import {
  getPrometheusCollector,
  isPrometheusEnabled,
} from './services/collectors/prometheus.js'
import {
  getMdstatCollector,
  isMdstatEnabled,
} from './services/collectors/mdstat.js'
import {
  getSmartCollector,
  isSmartEnabled,
} from './services/collectors/smart.js'
import {
  getNvidiaCollector,
  isNvidiaEnabled,
} from './services/collectors/nvidia.js'
import {
  getIpmiCollector,
  isIpmiEnabled,
} from './services/collectors/ipmi.js'
import {
  getSelfMonitorCollector,
  isSelfMonitorEnabled,
} from './services/collectors/self-monitor.js'
import {
  getStatsdCollector,
  isStatsdEnabled,
} from './services/collectors/statsd.js'
import {
  getRabbitmqCollector,
  isRabbitmqEnabled,
} from './services/collectors/rabbitmq.js'
import {
  getMongodbCollector,
  isMongodbEnabled,
} from './services/collectors/mongodb.js'
import {
  getEbpfCollector,
  isEbpfEnabled,
} from './services/collectors/ebpf.js'
import {
  getIopingCollector,
  isIopingEnabled,
} from './services/collectors/ioping.js'
import {
  getZfsCollector,
  isZfsEnabled,
} from './services/collectors/zfs.js'
import {
  getBcacheCollector,
  isBcacheEnabled,
} from './services/collectors/bcache.js'
import {
  getKafkaCollector,
  isKafkaEnabled,
} from './services/collectors/kafka.js'
import {
  getNatsCollector,
  isNatsEnabled,
} from './services/collectors/nats.js'
import {
  getUnboundCollector,
  isUnboundEnabled,
} from './services/collectors/unbound.js'
import {
  getSocketsCollector,
  isSocketsEnabled,
} from './services/collectors/sockets.js'
import {
  getDmcacheCollector,
  isDmcacheEnabled,
} from './services/collectors/dmcache.js'
import {
  getFsStatsCollector,
  isFsStatsEnabled,
} from './services/collectors/fs-stats.js'
import { notifyAnomaly } from './services/notify.js'
import { getStore } from './services/store.js'
import { getAnomalyEngine } from './services/anomaly.js'
import {
  broadcastMetrics,
  broadcastAnomaly,
  broadcastHealth,
} from './services/subscriptions.js'
import { enqueueWarmPoint, flushWarmPending, persistAlertEvent } from './services/warm-flush.js'
import { getDb } from './db/index.js'
import { Pushes } from '../shared/protocol.js'
import { peers } from './core/peer-registry.js'
import logger from './utils/logger.js'

const log = logger.child('pipeline')
let healthEvery = 0
let warmFlushEvery = 0

/**
 * @param {import('./services/store.js').MetricStore} store
 * @param {import('./services/anomaly.js').AnomalyEngine} anomalies
 * @param {Array<{ chart: string, context: string, ts: number, values: object }>} batch
 */
function ingestBatch(store, anomalies, batch) {
  const t0 = performance.now()
  store.ingest(batch)
  const tIngest = performance.now()
  broadcastMetrics(batch)
  const tBroadcast = performance.now()

  const fired = anomalies.evaluate(batch)
  const tAnomaly = performance.now()
  for (const ev of fired) {
    broadcastAnomaly(ev)
    persistAlertEvent(ev)
    notifyAnomaly(ev).catch(() => {})
    if (!ev.cleared) {
      peers.broadcast(Pushes.alert, {
        id: ev.id,
        status: ev.severity === 'critical' ? 'CRITICAL' : 'WARNING',
        ...ev,
      })
    }
  }

  healthEvery++
  if (healthEvery >= 15) {
    healthEvery = 0
    broadcastHealth(anomalies.getHealth())
  }

  warmFlushEvery++
  if (warmFlushEvery >= 10) {
    warmFlushEvery = 0
    flushWarmPending().catch(() => {})
  }
  const dur = performance.now() - t0
  if (dur > 10) {
    log.debug('ingest', {
      durMs: Math.round(dur * 10) / 10,
      ingestMs: Math.round((tIngest - t0) * 10) / 10,
      bcastMs: Math.round((tBroadcast - tIngest) * 10) / 10,
      anomalyMs: Math.round((tAnomaly - tBroadcast) * 10) / 10,
      charts: batch.length,
      fired: fired.length,
    })
  }
}

export function startPipeline() {
  const collector = getCollector()
  const store = getStore()
  const anomalies = getAnomalyEngine(os.cpus().length)

  store.on('warm', (point) => {
    enqueueWarmPoint(point)
  })

  collector.on('samples', (batch) => ingestBatch(store, anomalies, batch))
  collector.start()

  let docker = null
  if (isDockerCollectorEnabled()) {
    docker = getDockerCollector()
    docker.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    docker.start()
  }

  let processes = null
  if (isProcessCollectorEnabled()) {
    processes = getProcessCollector()
    processes.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    processes.start()
  }

  let parent = null
  if (isParentEnabled()) {
    parent = getParentCollector()
    parent.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    parent.start()
  }

  let nginx = null
  if (isNginxEnabled()) {
    nginx = getNginxCollector()
    nginx.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    nginx.on('error', (err) => log.warn('Nginx collector error', { error: err.message }))
    nginx.start()
  }

  let redis = null
  if (isRedisEnabled()) {
    redis = getRedisCollector()
    redis.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    redis.on('error', (err) => log.warn('Redis collector error', { error: err.message }))
    redis.start()
  }

  let postgres = null
  if (isPostgresEnabled()) {
    postgres = getPostgresCollector()
    postgres.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    postgres.on('error', (err) => log.warn('Postgres collector error', { error: err.message }))
    postgres.start()
  }

  let peardock = null
  if (isPearDockEnabled()) {
    peardock = getPearDockCollector()
    peardock.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    peardock.start()
  }

  let cgroups = null
  if (isCgroupsEnabled()) {
    cgroups = getCgroupsCollector()
    cgroups.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    cgroups.start()
  }

  let sensors = null
  if (isSensorsEnabled()) {
    sensors = getSensorsCollector()
    sensors.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    sensors.start()
  }

  let mysql = null
  if (isMysqlEnabled()) {
    mysql = getMysqlCollector()
    mysql.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    mysql.on('error', (err) => log.warn('MySQL collector error', { error: err.message }))
    mysql.start()
  }

  let apache = null
  if (isApacheEnabled()) {
    apache = getApacheCollector()
    apache.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    apache.on('error', (err) => log.warn('Apache collector error', { error: err.message }))
    apache.start()
  }

  let memcached = null
  if (isMemcachedEnabled()) {
    memcached = getMemcachedCollector()
    memcached.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    memcached.on('error', (err) => log.warn('Memcached collector error', { error: err.message }))
    memcached.start()
  }

  let prometheus = null
  if (isPrometheusEnabled()) {
    prometheus = getPrometheusCollector()
    prometheus.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    prometheus.on('error', (err) => log.warn('Prometheus collector error', { error: err.message }))
    prometheus.start()
  }

  let mdstat = null
  if (isMdstatEnabled()) {
    mdstat = getMdstatCollector()
    mdstat.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    mdstat.start()
  }

  let smart = null
  if (isSmartEnabled()) {
    smart = getSmartCollector()
    smart.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    smart.on('error', (err) => log.warn('SMART collector error', { error: err.message }))
    smart.start()
  }

  let nvidia = null
  if (isNvidiaEnabled()) {
    nvidia = getNvidiaCollector()
    nvidia.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    nvidia.on('error', (err) => log.warn('NVIDIA collector error', { error: err.message }))
    nvidia.start()
  }

  let ipmi = null
  if (isIpmiEnabled()) {
    ipmi = getIpmiCollector()
    ipmi.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    ipmi.on('error', (err) => log.warn('IPMI collector error', { error: err.message }))
    ipmi.start()
  }

  let selfMonitor = null
  if (isSelfMonitorEnabled()) {
    selfMonitor = getSelfMonitorCollector()
    selfMonitor.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    selfMonitor.on('error', (err) => log.warn('Self-monitor error', { error: err.message }))
    selfMonitor.start()
  }

  let statsd = null
  if (isStatsdEnabled()) {
    statsd = getStatsdCollector()
    statsd.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    statsd.start()
  }

  let rabbitmq = null
  if (isRabbitmqEnabled()) {
    rabbitmq = getRabbitmqCollector()
    rabbitmq.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    rabbitmq.on('error', (err) => log.warn('RabbitMQ collector error', { error: err.message }))
    rabbitmq.start()
  }

  let mongodb = null
  if (isMongodbEnabled()) {
    mongodb = getMongodbCollector()
    mongodb.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    mongodb.on('error', (err) => log.warn('MongoDB collector error', { error: err.message }))
    mongodb.start()
  }

  let ebpf = null
  if (isEbpfEnabled()) {
    ebpf = getEbpfCollector()
    ebpf.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    ebpf.start()
  }

  let ioping = null
  if (isIopingEnabled()) {
    ioping = getIopingCollector()
    ioping.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    ioping.on('error', (err) => log.warn('ioping error', { error: err.message }))
    ioping.start()
  }

  let zfs = null
  if (isZfsEnabled()) {
    zfs = getZfsCollector()
    zfs.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    zfs.on('error', (err) => log.warn('ZFS error', { error: err.message }))
    zfs.start()
  }

  let bcache = null
  if (isBcacheEnabled()) {
    bcache = getBcacheCollector()
    bcache.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    bcache.on('error', (err) => log.warn('BCache error', { error: err.message }))
    bcache.start()
  }

  let kafka = null
  if (isKafkaEnabled()) {
    kafka = getKafkaCollector()
    kafka.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    kafka.on('error', (err) => log.warn('Kafka error', { error: err.message }))
    kafka.start()
  }

  let nats = null
  if (isNatsEnabled()) {
    nats = getNatsCollector()
    nats.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    nats.on('error', (err) => log.warn('NATS error', { error: err.message }))
    nats.start()
  }

  let unbound = null
  if (isUnboundEnabled()) {
    unbound = getUnboundCollector()
    unbound.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    unbound.on('error', (err) => log.warn('Unbound error', { error: err.message }))
    unbound.start()
  }

  let sockets = null
  if (isSocketsEnabled()) {
    sockets = getSocketsCollector()
    sockets.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    sockets.start()
  }

  let dmcache = null
  if (isDmcacheEnabled()) {
    dmcache = getDmcacheCollector()
    dmcache.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    dmcache.on('error', (err) => log.warn('dm-cache error', { error: err.message }))
    dmcache.start()
  }

  let fsStats = null
  if (isFsStatsEnabled()) {
    fsStats = getFsStatsCollector()
    fsStats.on('samples', (batch) => ingestBatch(store, anomalies, batch))
    fsStats.on('error', (err) => log.warn('fs-stats error', { error: err.message }))
    fsStats.start()
  }

  log.info('Metrics pipeline started', {
    hyperdb: Boolean(getDb()),
    docker: Boolean(docker),
    processes: Boolean(processes),
    parent: Boolean(parent),
    nginx: Boolean(nginx),
    redis: Boolean(redis),
    postgres: Boolean(postgres),
    peardock: Boolean(peardock),
    cgroups: Boolean(cgroups),
    sensors: Boolean(sensors),
    mysql: Boolean(mysql),
    apache: Boolean(apache),
    memcached: Boolean(memcached),
    prometheus: Boolean(prometheus),
    mdstat: Boolean(mdstat),
    smart: Boolean(smart),
    nvidia: Boolean(nvidia),
    ipmi: Boolean(ipmi),
    selfMonitor: Boolean(selfMonitor),
    statsd: Boolean(statsd),
    rabbitmq: Boolean(rabbitmq),
    mongodb: Boolean(mongodb),
    ebpf: Boolean(ebpf),
    ioping: Boolean(ioping),
    zfs: Boolean(zfs),
    bcache: Boolean(bcache),
    kafka: Boolean(kafka),
    nats: Boolean(nats),
    unbound: Boolean(unbound),
    sockets: Boolean(sockets),
    dmcache: Boolean(dmcache),
    fsStats: Boolean(fsStats),
  })
  return {
    collector,
    store,
    anomalies,
    docker,
    processes,
    parent,
    nginx,
    redis,
    postgres,
    peardock,
    cgroups,
    sensors,
    mysql,
    apache,
    memcached,
    prometheus,
    mdstat,
    smart,
    nvidia,
    ipmi,
    selfMonitor,
    statsd,
    rabbitmq,
    mongodb,
    ebpf,
    ioping,
    zfs,
    bcache,
    kafka,
    nats,
    unbound,
    sockets,
    dmcache,
    fsStats,
  }
}