peardata/server/services/collectors/mysql.js

/**
 * MySQL collector (service plugin).
 *
 * Enable: PEARDATA_MYSQL=1
 * Addr:   PEARDATA_MYSQL_URL=127.0.0.1:3306
 * Optional HTTP stats (key=value): PEARDATA_MYSQL_STATS_URL
 *
 * Charts: mysql.up, mysql.stats
 */
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('mysql')

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

const CHART_STATS = {
  id: 'mysql.stats',
  name: 'mysql.stats',
  context: 'mysql.stats',
  title: 'MySQL stats',
  units: 'count',
  family: 'mysql',
  chartType: 'line',
  priority: 8310,
  plugin: 'mysql',
  dimensions: [
    { id: 'questions', name: 'questions', algorithm: 'absolute' },
    { id: 'threads_connected', name: 'threads_connected', algorithm: 'absolute' },
    { id: 'slow_queries', name: 'slow_queries', algorithm: 'absolute' },
  ],
}

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

/**
 * @param {string} urlOrHost
 * @returns {{ host: string, port: number }}
 */
export function parseMysqlAddr(urlOrHost) {
  const raw = String(urlOrHost || '127.0.0.1:3306').trim()
  if (raw.includes('://')) {
    try {
      const u = new URL(raw)
      return {
        host: u.hostname || '127.0.0.1',
        port: Number(u.port) || 3306,
      }
    } catch {
      // fall through
    }
  }
  const [host, port] = raw.split(':')
  return { host: host || '127.0.0.1', port: Number(port) || 3306 }
}

/**
 * @param {string} body
 * @returns {Record<string, number>}
 */
export function parseMysqlStats(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 probeMysqlTcp(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 MysqlCollector extends CollectorPlugin {
  constructor(opts = {}) {
    super({ name: 'mysql', intervalMs: opts.intervalMs })
    this.addr = parseMysqlAddr(opts.url || process.env.PEARDATA_MYSQL_URL || '127.0.0.1:3306')
    this.statsUrl = opts.statsUrl || process.env.PEARDATA_MYSQL_STATS_URL || ''
  }

  isEnabled() {
    return isMysqlEnabled()
  }

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

  async collect() {
    const ts = Date.now()
    const probe = await probeMysqlTcp(this.addr)
    /** @type {Array<{ chart: string, context: string, ts: number, values: Record<string, number|null> }>} */
    const batch = [
      {
        chart: 'mysql.up',
        context: 'mysql.up',
        ts,
        values: { up: probe.up, latency_ms: probe.latency_ms },
      },
    ]
    if (this.statsUrl) {
      try {
        const body = await fetchText(this.statsUrl)
        const s = parseMysqlStats(body)
        registerChart(CHART_STATS)
        batch.push({
          chart: 'mysql.stats',
          context: 'mysql.stats',
          ts,
          values: {
            questions: s.questions ?? s.Queries ?? null,
            threads_connected: s.threads_connected ?? s.Threads_connected ?? null,
            slow_queries: s.slow_queries ?? s.Slow_queries ?? null,
          },
        })
      } catch (err) {
        log.warn('MySQL stats URL failed', { error: err.message })
      }
    }
    return batch
  }
}

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

export function getMysqlCollector() {
  if (!singleton) singleton = new MysqlCollector()
  return singleton
}