Updates

2026-07-18 19:49:57 -04:00
parent 2e1a3e9b06
commit a639b3c953
19 changed files with 1705 additions and 293 deletions
@@ -0,0 +1,88 @@
+/**
+ * Kafka JMX/exporter HTTP probe (Burrow-style or prometheus jmx exporter).
+ * Enable: PEARDATA_KAFKA=1
+ * URL: PEARDATA_KAFKA_URL=http://127.0.0.1:9308/metrics (prometheus) OR
+ *      PEARDATA_KAFKA_URL=http://127.0.0.1:8080 (plain JSON under /stats)
+ */
+import { CollectorPlugin } from './plugin.js'
+import { registerChart } from '../../../shared/metrics.js'
+
+export function isKafkaEnabled() {
+  const v = process.env.PEARDATA_KAFKA
+  return v === '1' || v === 'on' || v === 'true'
+}
+
+function url() {
+  return (process.env.PEARDATA_KAFKA_URL || 'http://127.0.0.1:9308/metrics').replace(/\/$/, '')
+}
+
+export class KafkaCollector extends CollectorPlugin {
+  constructor() {
+    super({ name: 'kafka' })
+  }
+
+  isEnabled() {
+    return isKafkaEnabled()
+  }
+
+  async collect() {
+    const ts = Date.now()
+    registerChart({
+      id: 'kafka.up',
+      name: 'kafka.up',
+      context: 'kafka.up',
+      title: 'Kafka exporter up',
+      units: 'boolean',
+      family: 'kafka',
+      chartType: 'line',
+      priority: 8800,
+      plugin: 'kafka',
+      dimensions: [{ id: 'up', name: 'up', algorithm: 'absolute' }],
+    })
+    try {
+      const res = await fetch(url(), { signal: AbortSignal.timeout(2500) })
+      if (!res.ok) throw new Error(String(res.status))
+      const text = await res.text()
+      /** @type {Record<string, number>} */
+      const vals = { brokers: 0, under_replicated: 0, offline_count: 0 }
+      for (const line of text.split('\n')) {
+        if (line.startsWith('#') || !line.trim()) continue
+        const m = line.match(/^([a-zA-Z0-9_:]+)(?:\{[^}]*\})?\s+([0-9.eE+-]+)/)
+        if (!m) continue
+        const name = m[1]
+        const v = Number(m[2])
+        if (name.includes('broker') && name.includes('info')) vals.brokers += 1
+        if (name.includes('under_replicated')) vals.under_replicated += v
+        if (name.includes('topic') && name.includes('partitions')) vals.partition_count += v
+      }
+      registerChart({
+        id: 'kafka.cluster',
+        name: 'kafka.cluster',
+        context: 'kafka.cluster',
+        title: 'Kafka cluster',
+        units: 'count',
+        family: 'kafka',
+        chartType: 'line',
+        priority: 8810,
+        plugin: 'kafka',
+        dimensions: Object.keys(vals).map((id) => ({
+          id,
+          name: id,
+          algorithm: 'absolute',
+        })),
+      })
+      return [
+        { chart: 'kafka.up', context: 'kafka.up', ts, values: { up: 1 } },
+        { chart: 'kafka.cluster', context: 'kafka.cluster', ts, values: vals },
+      ]
+    } catch {
+      return [{ chart: 'kafka.up', context: 'kafka.up', ts, values: { up: 0 } }]
+    }
+  }
+}
+
+let singleton = null
+export function getKafkaCollector() {
+  if (!singleton) singleton = new KafkaCollector()
+  return singleton
+}