peardata/server/services/collectors/memcached.js

/**
 * Memcached stats collector (service plugin).
 *
 * Enable: PEARDATA_MEMCACHED=1
 * Addr:   PEARDATA_MEMCACHED_URL=127.0.0.1:11211
 *
 * Charts: memcached.ops, memcached.memory
 */
import net from 'net'
import { CollectorPlugin } from './plugin.js'
import { registerChart } from '../../../shared/metrics.js'
import logger from '../../utils/logger.js'

const log = logger.child('memcached')

const CHART_OPS = {
  id: 'memcached.ops',
  name: 'memcached.ops',
  context: 'memcached.ops',
  title: 'Memcached operations',
  units: 'ops/s',
  family: 'memcached',
  chartType: 'line',
  priority: 8500,
  plugin: 'memcached',
  dimensions: [
    { id: 'get_hits', name: 'get_hits', algorithm: 'absolute' },
    { id: 'get_misses', name: 'get_misses', algorithm: 'absolute' },
    { id: 'cmd_get', name: 'cmd_get', algorithm: 'absolute' },
    { id: 'cmd_set', name: 'cmd_set', algorithm: 'absolute' },
  ],
}

const CHART_MEMORY = {
  id: 'memcached.memory',
  name: 'memcached.memory',
  context: 'memcached.memory',
  title: 'Memcached memory',
  units: 'bytes',
  family: 'memcached',
  chartType: 'area',
  priority: 8510,
  plugin: 'memcached',
  dimensions: [
    { id: 'bytes', name: 'bytes', algorithm: 'absolute' },
    { id: 'limit_maxbytes', name: 'limit_maxbytes', algorithm: 'absolute' },
  ],
}

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

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

/**
 * @param {string} body
 * @returns {Record<string, number>}
 */
export function parseMemcachedStats(body) {
  /** @type {Record<string, number>} */
  const out = {}
  for (const line of String(body || '').split(/\r?\n/)) {
    const m = line.match(/^STAT\s+(\S+)\s+(\S+)/)
    if (!m) continue
    const n = Number(m[2])
    if (Number.isFinite(n)) out[m[1]] = n
  }
  return out
}

/**
 * @param {{ host: string, port: number }} addr
 * @param {number} [timeoutMs]
 * @returns {Promise<string>}
 */
export function fetchMemcachedStats(addr, timeoutMs = 3000) {
  return new Promise((resolve, reject) => {
    const socket = net.createConnection({ host: addr.host, port: addr.port })
    let buf = ''
    let settled = false
    const done = (err, data) => {
      if (settled) return
      settled = true
      clearTimeout(timer)
      try {
        socket.destroy()
      } catch {
        // ignore
      }
      if (err) reject(err)
      else resolve(data)
    }
    const timer = setTimeout(() => done(new Error('memcached timeout')), timeoutMs)
    socket.on('connect', () => {
      socket.write('stats\r\n')
    })
    socket.on('data', (chunk) => {
      buf += chunk.toString('utf8')
      if (buf.includes('END')) done(null, buf)
    })
    socket.on('error', (err) => done(err))
    socket.on('end', () => {
      if (buf) done(null, buf)
      else done(new Error('memcached closed with no data'))
    })
  })
}

export class MemcachedCollector extends CollectorPlugin {
  constructor(opts = {}) {
    super({ name: 'memcached', intervalMs: opts.intervalMs })
    this.addr = parseMemcachedAddr(
      opts.url || process.env.PEARDATA_MEMCACHED_URL || '127.0.0.1:11211'
    )
    /** @type {{ get_hits: number, get_misses: number, cmd_get: number, cmd_set: number, wallMs: number }|null} */
    this._prev = null
  }

  isEnabled() {
    return isMemcachedEnabled()
  }

  start() {
    if (!this.isEnabled()) return
    registerChart(CHART_OPS)
    registerChart(CHART_MEMORY)
    log.info('Memcached collector started', this.addr)
    super.start()
  }

  async collect() {
    const raw = await fetchMemcachedStats(this.addr)
    const stats = parseMemcachedStats(raw)
    const ts = Date.now()

    const cur = {
      get_hits: Number(stats.get_hits) || 0,
      get_misses: Number(stats.get_misses) || 0,
      cmd_get: Number(stats.cmd_get) || 0,
      cmd_set: Number(stats.cmd_set) || 0,
    }

    /** @type {Record<string, number>} */
    const rates = { get_hits: 0, get_misses: 0, cmd_get: 0, cmd_set: 0 }
    if (this._prev && ts > this._prev.wallMs) {
      const dt = (ts - this._prev.wallMs) / 1000
      if (dt > 0) {
        for (const k of Object.keys(rates)) {
          rates[k] = Math.max(0, (cur[k] - this._prev[k]) / dt)
        }
      }
    }
    this._prev = { ...cur, wallMs: ts }

    return [
      {
        chart: 'memcached.ops',
        context: 'memcached.ops',
        ts,
        values: rates,
      },
      {
        chart: 'memcached.memory',
        context: 'memcached.memory',
        ts,
        values: {
          bytes: Number(stats.bytes) || 0,
          limit_maxbytes: Number(stats.limit_maxbytes) || 0,
        },
      },
    ]
  }
}

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

export function getMemcachedCollector() {
  if (!singleton) singleton = new MemcachedCollector()
  return singleton
}