113 lines
3.2 KiB
JavaScript
113 lines
3.2 KiB
JavaScript
/**
|
|
* StatsD UDP ingest — maps gauges/counters into prom-like charts.
|
|
* Enable: PEARDATA_STATSD=1
|
|
* Bind: PEARDATA_STATSD_PORT=8125 (default)
|
|
*/
|
|
import dgram from 'dgram'
|
|
import { EventEmitter } from 'events'
|
|
import { SAMPLE_INTERVAL_MS, registerChart } from '../../../shared/metrics.js'
|
|
import logger from '../../utils/logger.js'
|
|
|
|
const log = logger.child('statsd')
|
|
|
|
export function isStatsdEnabled() {
|
|
const v = process.env.PEARDATA_STATSD
|
|
return v === '1' || v === 'on' || v === 'true'
|
|
}
|
|
|
|
function sanitize(name) {
|
|
return String(name)
|
|
.replace(/[^a-zA-Z0-9_.-]/g, '_')
|
|
.slice(0, 80)
|
|
}
|
|
|
|
export class StatsdCollector extends EventEmitter {
|
|
constructor(opts = {}) {
|
|
super()
|
|
this.port = opts.port ?? (Number(process.env.PEARDATA_STATSD_PORT) || 8125)
|
|
this.intervalMs =
|
|
opts.intervalMs ?? (Number(process.env.PEARDATA_SAMPLE_MS) || SAMPLE_INTERVAL_MS)
|
|
/** @type {Map<string, number>} */
|
|
this.values = new Map()
|
|
this.socket = null
|
|
this.timer = null
|
|
this.running = false
|
|
}
|
|
|
|
start() {
|
|
if (this.running) return
|
|
this.running = true
|
|
this.socket = dgram.createSocket('udp4')
|
|
this.socket.on('message', (msg) => this._onMessage(msg.toString('utf8')))
|
|
this.socket.on('error', (err) => log.warn('StatsD socket error', { error: err.message }))
|
|
this.socket.bind(this.port, '127.0.0.1', () => {
|
|
log.info('StatsD listening', { port: this.port })
|
|
})
|
|
this.timer = setInterval(() => this._flush(), this.intervalMs)
|
|
if (this.timer.unref) this.timer.unref()
|
|
}
|
|
|
|
stop() {
|
|
this.running = false
|
|
if (this.timer) clearInterval(this.timer)
|
|
this.timer = null
|
|
try {
|
|
this.socket?.close()
|
|
} catch {
|
|
// ignore
|
|
}
|
|
this.socket = null
|
|
}
|
|
|
|
_onMessage(text) {
|
|
for (const line of text.split('\n')) {
|
|
if (!line.trim()) continue
|
|
// name:value|type
|
|
const m = line.trim().match(/^([^:]+):(-?[\d.]+)\|([a-z]+)(?:\|@[\d.]+)?$/i)
|
|
if (!m) continue
|
|
const name = sanitize(m[1])
|
|
const value = Number(m[2])
|
|
const type = m[3].toLowerCase()
|
|
if (!Number.isFinite(value)) continue
|
|
if (type === 'c') {
|
|
this.values.set(name, (this.values.get(name) || 0) + value)
|
|
} else {
|
|
this.values.set(name, value)
|
|
}
|
|
}
|
|
}
|
|
|
|
_flush() {
|
|
if (!this.values.size) return
|
|
const ts = Date.now()
|
|
/** @type {Array<{ chart: string, context: string, ts: number, values: object }>} */
|
|
const batch = []
|
|
let n = 0
|
|
for (const [name, value] of this.values) {
|
|
if (n++ >= 64) break
|
|
const id = `statsd.${name}`
|
|
registerChart({
|
|
id,
|
|
name: id,
|
|
context: 'statsd.metric',
|
|
title: `StatsD ${name}`,
|
|
units: 'value',
|
|
family: 'statsd',
|
|
chartType: 'line',
|
|
priority: 8500,
|
|
plugin: 'statsd',
|
|
dimensions: [{ id: 'value', name: 'value', algorithm: 'absolute' }],
|
|
})
|
|
batch.push({ chart: id, context: 'statsd.metric', ts, values: { value } })
|
|
}
|
|
if (batch.length) this.emit('samples', batch)
|
|
}
|
|
}
|
|
|
|
/** @type {StatsdCollector|null} */
|
|
let singleton = null
|
|
export function getStatsdCollector() {
|
|
if (!singleton) singleton = new StatsdCollector()
|
|
return singleton
|
|
}
|