peardata/server/services/collectors/sockets.js

/**
 * TCP/UDP socket state timeseries from /proc/net/tcp{,6} and udp.
 * Always-on on Linux (cheap enough); disable with PEARDATA_SOCKETS=0.
 */
import fs from 'fs'
import os from 'os'
import { EventEmitter } from 'events'
import { SAMPLE_INTERVAL_MS, registerChart } from '../../../shared/metrics.js'
import { readFileBuf } from '../../utils/fd-cache.js'
import logger from '../../utils/logger.js'

const log = logger.child('sockets')

const TCP_STATES = {
  '01': 'established',
  '02': 'syn_sent',
  '03': 'syn_recv',
  '04': 'fin_wait1',
  '05': 'fin_wait2',
  '06': 'time_wait',
  '07': 'close',
  '08': 'close_wait',
  '09': 'last_ack',
  '0A': 'listen',
  '0B': 'closing',
}

export function isSocketsEnabled() {
  const v = process.env.PEARDATA_SOCKETS
  if (v === '0' || v === 'off' || v === 'false') return false
  if (v === '1' || v === 'on' || v === 'true') return true
  return os.platform() === 'linux'
}

function countTcp(file) {
  /** @type {Record<string, number>} */
  const counts = {
    established: 0,
    listen: 0,
    time_wait: 0,
    close_wait: 0,
    syn_sent: 0,
    syn_recv: 0,
    other: 0,
  }
  const raw = readFileBuf(file)
  if (!raw) return counts
  for (const line of raw.split('\n').slice(1)) {
    const parts = line.trim().split(/\s+/)
    if (parts.length < 4) continue
    const st = parts[3]
    const name = TCP_STATES[st]
    if (name && name in counts) counts[name]++
    else counts.other++
  }
  return counts
}

function countUdp(file) {
  const raw = readFileBuf(file)
  if (!raw) return 0
  return Math.max(0, raw.trim().split('\n').length - 1)
}

export class SocketsCollector extends EventEmitter {
  constructor(opts = {}) {
    super()
    this.intervalMs =
      opts.intervalMs ?? (Number(process.env.PEARDATA_SAMPLE_MS) || SAMPLE_INTERVAL_MS)
    this.timer = null
    this.running = false
  }

  start() {
    if (this.running) return
    this.running = true
    registerChart({
      id: 'net.socket_states',
      name: 'net.socket_states',
      context: 'net.socket_states',
      title: 'TCP socket states',
      units: 'sockets',
      family: 'sockets',
      chartType: 'stacked',
      priority: 775,
      plugin: 'sockets',
      dimensions: [
        'established',
        'listen',
        'time_wait',
        'close_wait',
        'syn_sent',
        'syn_recv',
        'other',
      ].map((id) => ({ id, name: id, algorithm: 'absolute' })),
    })
    registerChart({
      id: 'net.udp_sockets',
      name: 'net.udp_sockets',
      context: 'net.udp_sockets',
      title: 'UDP sockets',
      units: 'sockets',
      family: 'sockets',
      chartType: 'line',
      priority: 776,
      plugin: 'sockets',
      dimensions: [
        { id: 'udp', name: 'udp', algorithm: 'absolute' },
        { id: 'udp6', name: 'udp6', algorithm: 'absolute' },
      ],
    })
    this._tick()
    this.timer = setInterval(() => this._tick(), this.intervalMs)
    if (this.timer.unref) this.timer.unref()
    log.info('Sockets collector started')
  }

  stop() {
    this.running = false
    if (this.timer) clearInterval(this.timer)
    this.timer = null
  }

  _tick() {
    const t0 = performance.now()
    const ts = Date.now()
    const v4 = countTcp('/proc/net/tcp')
    const v6 = countTcp('/proc/net/tcp6')
    /** @type {Record<string, number>} */
    const merged = {}
    for (const k of Object.keys(v4)) merged[k] = (v4[k] || 0) + (v6[k] || 0)
    this.emit('samples', [
      { chart: 'net.socket_states', context: 'net.socket_states', ts, values: merged },
      {
        chart: 'net.udp_sockets',
        context: 'net.udp_sockets',
        ts,
        values: {
          udp: countUdp('/proc/net/udp'),
          udp6: countUdp('/proc/net/udp6'),
        },
      },
    ])
    const dur = performance.now() - t0
    if (dur > 5) log.debug('tick', { collector: 'sockets', durMs: Math.round(dur * 10) / 10 })
  }
}

let singleton = null
export function getSocketsCollector() {
  if (!singleton) singleton = new SocketsCollector()
  return singleton
}