peardata/server/services/collectors/postgres.js

/**
 * Postgres collector (service plugin spike).
 *
 * Enable: PEARDATA_POSTGRES=1
 * TCP:    PEARDATA_POSTGRES_HOST / PEARDATA_POSTGRES_PORT (default 5432)
 * Optional HTTP stats (key=value lines): PEARDATA_POSTGRES_STATS_URL
 *
 * Charts:
 *   postgres.up          — 1/0 + connect latency
 *   postgres.stats       — from HTTP stats URL when set (connections, xact, tuples)
 */
import net from 'net'
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('postgres')

const CHART_UP = {
  id: 'postgres.up',
  name: 'postgres.up',
  context: 'postgres.up',
  title: 'Postgres availability',
  units: 'boolean',
  family: 'postgres',
  chartType: 'line',
  priority: 8200,
  plugin: 'postgres',
  dimensions: [
    { id: 'up', name: 'up', algorithm: 'absolute' },
    { id: 'latency_ms', name: 'latency_ms', algorithm: 'absolute' },
  ],
}

const CHART_STATS = {
  id: 'postgres.stats',
  name: 'postgres.stats',
  context: 'postgres.stats',
  title: 'Postgres stats',
  units: 'count',
  family: 'postgres',
  chartType: 'line',
  priority: 8210,
  plugin: 'postgres',
  dimensions: [
    { id: 'connections', name: 'connections', algorithm: 'absolute' },
    { id: 'xact_commit', name: 'xact_commit', algorithm: 'absolute' },
    { id: 'xact_rollback', name: 'xact_rollback', algorithm: 'absolute' },
    { id: 'tuples_returned', name: 'tuples_returned', algorithm: 'absolute' },
  ],
}

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

/**
 * Parse simple key=value postgres stats (custom exporter / sidecar).
 * @param {string} body
 * @returns {Record<string, number>}
 */
export function parsePostgresStats(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
    const m = t.match(/^([a-zA-Z0-9_]+)\s*[=:]\s*([0-9.]+)/)
    if (m) out[m[1]] = Number(m[2])
  }
  return out
}

/**
 * @param {{ host: string, port: number }} addr
 * @param {number} [timeoutMs]
 * @returns {Promise<{ up: number, latency_ms: number }>}
 */
export function probePostgresTcp(addr, timeoutMs = 3000) {
  return new Promise((resolve) => {
    const started = Date.now()
    const socket = net.createConnection({ host: addr.host, port: addr.port })
    let settled = false
    const finish = (up) => {
      if (settled) return
      settled = true
      clearTimeout(timer)
      try {
        socket.destroy()
      } catch {
        // ignore
      }
      resolve({ up, latency_ms: Date.now() - started })
    }
    const timer = setTimeout(() => finish(0), timeoutMs)
    socket.on('connect', () => finish(1))
    socket.on('error', () => finish(0))
  })
}

function fetchText(url, timeoutMs = 3000) {
  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 PostgresCollector extends CollectorPlugin {
  constructor(opts = {}) {
    super({ name: 'postgres', intervalMs: opts.intervalMs })
    this.host = opts.host || process.env.PEARDATA_POSTGRES_HOST || '127.0.0.1'
    this.port = Number(opts.port || process.env.PEARDATA_POSTGRES_PORT) || 5432
    this.statsUrl = opts.statsUrl || process.env.PEARDATA_POSTGRES_STATS_URL || ''
  }

  isEnabled() {
    return isPostgresEnabled()
  }

  start() {
    if (!this.isEnabled()) return
    registerChart(CHART_UP)
    if (this.statsUrl) registerChart(CHART_STATS)
    log.info('Postgres collector started', {
      host: this.host,
      port: this.port,
      statsUrl: this.statsUrl || null,
    })
    super.start()
  }

  async collect() {
    const ts = Date.now()
    const probe = await probePostgresTcp({ host: this.host, port: this.port })
    /** @type {Array<{ chart: string, context: string, ts: number, values: Record<string, number|null> }>} */
    const batch = [
      {
        chart: 'postgres.up',
        context: 'postgres.up',
        ts,
        values: { up: probe.up, latency_ms: probe.latency_ms },
      },
    ]
    if (this.statsUrl) {
      try {
        const body = await fetchText(this.statsUrl)
        const s = parsePostgresStats(body)
        registerChart(CHART_STATS)
        batch.push({
          chart: 'postgres.stats',
          context: 'postgres.stats',
          ts,
          values: {
            connections: s.connections ?? s.numbackends ?? null,
            xact_commit: s.xact_commit ?? null,
            xact_rollback: s.xact_rollback ?? null,
            tuples_returned: s.tuples_returned ?? s.tup_returned ?? null,
          },
        })
      } catch (err) {
        log.warn('Postgres stats URL failed', { error: err.message })
      }
    }
    return batch
  }
}

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

export function getPostgresCollector() {
  if (!singleton) singleton = new PostgresCollector()
  return singleton
}