peardata/server/services/collectors/prometheus.js

/**
 * Prometheus text exposition scraper (service plugin).
 *
 * Enable: PEARDATA_PROMETHEUS=1
 * URLs:   PEARDATA_PROMETHEUS_URLS=comma-separated scrape URLs
 *
 * Charts: prom.{sanitized_metric} — one gauge/counter value per metric (max 40)
 */
import http from 'http'
import https from 'https'
import { CollectorPlugin } from './plugin.js'
import { registerChart } from '../../../shared/metrics.js'
import logger from '../../utils/logger.js'

const log = logger.child('prometheus')

const MAX_METRICS = 40

export function isPrometheusEnabled() {
  const v = process.env.PEARDATA_PROMETHEUS
  return v === '1' || v === 'on' || v === 'true'
}

/**
 * @param {string} raw
 * @returns {string[]}
 */
export function parsePrometheusUrls(raw) {
  return String(raw || '')
    .split(',')
    .map((s) => s.trim())
    .filter(Boolean)
}

/**
 * Sanitize metric name for chart id.
 * @param {string} name
 */
export function sanitizeMetricName(name) {
  return String(name)
    .replace(/[^a-zA-Z0-9_.-]/g, '_')
    .replace(/^_+|_+$/g, '')
    .slice(0, 96)
}

/**
 * Parse Prometheus text exposition format.
 * Skips histogram/summary _bucket lines; returns latest scalar per metric name.
 * @param {string} body
 * @returns {Record<string, number>}
 */
export function parsePrometheusText(body) {
  /** @type {Record<string, number>} */
  const out = {}
  for (const line of String(body || '').split(/\r?\n/)) {
    const t = line.trim()
    if (!t || t.startsWith('#')) continue
    if (t.includes('_bucket{') || t.endsWith('_bucket')) continue
    if (t.includes('_sum{') || t.endsWith('_sum')) continue
    if (t.includes('_count{') || t.endsWith('_count')) continue
    const sp = t.lastIndexOf(' ')
    if (sp < 0) continue
    let namePart = t.slice(0, sp).trim()
    const valStr = t.slice(sp + 1).trim()
    const brace = namePart.indexOf('{')
    if (brace >= 0) namePart = namePart.slice(0, brace)
    if (namePart.endsWith('_bucket') || namePart.endsWith('_sum') || namePart.endsWith('_count')) {
      continue
    }
    const val = Number(valStr)
    if (!Number.isFinite(val)) continue
    out[namePart] = val
  }
  return out
}

function makePromChart(metric) {
  const safe = sanitizeMetricName(metric)
  return {
    id: `prom.${safe}`,
    name: `prom.${safe}`,
    context: 'prometheus.metric',
    title: `Prometheus ${metric}`,
    units: 'value',
    family: 'prometheus',
    chartType: 'line',
    priority: 8600,
    plugin: 'prometheus',
    dimensions: [{ id: 'value', name: 'value', algorithm: 'absolute' }],
  }
}

function fetchText(url, timeoutMs = 5000) {
  return new Promise((resolve, reject) => {
    const mod = String(url).startsWith('https') ? https : http
    const req = mod.get(url, (res) => {
      let body = ''
      res.on('data', (c) => {
        body += c
      })
      res.on('end', () => {
        if (res.statusCode && res.statusCode >= 400) {
          reject(new Error(`HTTP ${res.statusCode}`))
          return
        }
        resolve(body)
      })
    })
    req.setTimeout(timeoutMs, () => {
      req.destroy()
      reject(new Error('timeout'))
    })
    req.on('error', reject)
  })
}

export class PrometheusCollector extends CollectorPlugin {
  constructor(opts = {}) {
    super({ name: 'prometheus', intervalMs: opts.intervalMs })
    this.urls = parsePrometheusUrls(
      opts.urls || process.env.PEARDATA_PROMETHEUS_URLS || ''
    )
  }

  isEnabled() {
    return isPrometheusEnabled() && this.urls.length > 0
  }

  start() {
    if (!this.isEnabled()) return
    log.info('Prometheus collector started', { urls: this.urls.length })
    super.start()
  }

  async collect() {
    /** @type {Record<string, number>} */
    const merged = {}
    for (const url of this.urls) {
      try {
        const body = await fetchText(url)
        const parsed = parsePrometheusText(body)
        Object.assign(merged, parsed)
      } catch (err) {
        log.warn('Prometheus scrape failed', { url, error: err.message })
      }
    }

    const names = Object.keys(merged).slice(0, MAX_METRICS)
    const ts = Date.now()
    /** @type {Array<{ chart: string, context: string, ts: number, values: Record<string, number> }>} */
    const batch = []
    for (const name of names) {
      const def = makePromChart(name)
      registerChart(def)
      batch.push({
        chart: def.id,
        context: def.context,
        ts,
        values: { value: merged[name] },
      })
    }
    return batch
  }
}

/** @type {PrometheusCollector|null} */
let singleton = null

export function getPrometheusCollector() {
  if (!singleton) singleton = new PrometheusCollector()
  return singleton
}