/** * 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} */ export function parsePostgresStats(body) { /** @type {Record} */ 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 }>} */ 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 }