feat: Prometheus metrics endpoint for knox nodes

Refs #2 - /metrics served on a dedicated port (KNOX_METRICS_ADDR, default localhost:8932) via prometheus/client_golang, with Go runtime + process collectors - DB-derived gauges refreshed per scrape: observations by source, last-24h observations, entries, projects, sessions, pending reflections, threads by status, peers, observations by origin node, knowledge vector (max hcl per node) - live gossip counters (pulls/pushes, observations pulled/pushed, errors) incremented during the anti-entropy sweep; Run accepts an optional metrics handle (nil for one-shot CLI) - knox_node_info{node_id,name} for scrape identification - internal/metrics package + db MetricsSnapshot; tests for snapshot, scrape output, and counter increments
2026-08-29 06:06:49 -07:00
parent 25a7112d8a
commit 876d2aa45f
10 changed files with 494 additions and 20 deletions
@@ -0,0 +1,164 @@
+// Package metrics exposes Prometheus-format metrics for a knox node.
+//
+// Gauges are recomputed from the database on each scrape (cheap aggregates);
+// gossip counters are in-memory and incremented as the daemon exchanges data
+// with peers. The registry also gains the standard Go runtime and process
+// collectors from prometheus/client_golang.
+package metrics
+
+import (
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+
+	"github.com/david/knox/internal/db"
+)
+
+// Metrics holds the gossip event counters (incremented by watch) and the
+// scrape-time gauges derived from the database (refreshed on each scrape).
+type Metrics struct {
+	// Gossip counters (live).
+	pullsTotal          prometheus.Counter
+	pushesTotal         prometheus.Counter
+	obsPulledTotal      prometheus.Counter
+	obsPushedTotal      prometheus.Counter
+	errorsTotal         prometheus.Counter
+
+	// Snapshot gauges (updated per scrape).
+	observationsGauge      *prometheus.GaugeVec
+	entriesGauge           prometheus.Gauge
+	projectsGauge          prometheus.Gauge
+	sessionsGauge          prometheus.Gauge
+	pendingReflections     prometheus.Gauge
+	peersGauge             prometheus.Gauge
+	threadsByStatus        *prometheus.GaugeVec
+	byOriginNode           *prometheus.GaugeVec
+	knowledgeVector        *prometheus.GaugeVec
+	observationsLast24h    prometheus.Gauge
+
+	registry *prometheus.Registry
+	kdb      *db.KnoxDB
+}
+
+// New builds the metrics registry bound to a knowledge index.
+func New(kdb *db.KnoxDB, nodeName string) *Metrics {
+	reg := prometheus.NewRegistry()
+
+	m := &Metrics{
+		registry: reg,
+		kdb:      kdb,
+	}
+
+	// Node identity aids scraping: which node produced this output.
+	m.nodeInfo(kdb.NodeID(), nodeName)
+
+	m.pullsTotal = newCounter(reg, "knox_gossip_pulls_total", "Peer pull round-trips completed.")
+	m.pushesTotal = newCounter(reg, "knox_gossip_pushes_total", "Peer push round-trips completed.")
+	m.obsPulledTotal = newCounter(reg, "knox_gossip_observations_pulled_total", "Observations received from peers.")
+	m.obsPushedTotal = newCounter(reg, "knox_gossip_observations_pushed_total", "Observations sent to peers.")
+	m.errorsTotal = newCounter(reg, "knox_gossip_errors_total", "Gossip errors (ping/pull/push failures).")
+
+	m.observationsGauge = newGaugeVec(reg, "knox_observations_total", "Observation log size.", "source_id")
+	m.observationsLast24h = newGauge(reg, "knox_observations_last_24h", "Observations collected in the last 24h.")
+	m.entriesGauge = newGauge(reg, "knox_entries_total", "Materialized entry cache size.")
+	m.projectsGauge = newGauge(reg, "knox_projects_total", "Distinct projects in the entry cache.")
+	m.sessionsGauge = newGauge(reg, "knox_sessions_total", "Sessions tracked.")
+	m.pendingReflections = newGauge(reg, "knox_pending_reflections", "Sessions awaiting reflection.")
+	m.peersGauge = newGauge(reg, "knox_peers_total", "Known peer nodes.")
+	m.threadsByStatus = newGaugeVec(reg, "knox_threads_total", "Threads by status.", "status")
+	m.byOriginNode = newGaugeVec(reg, "knox_observations_by_node", "Observations per originating node.", "node_id")
+	m.knowledgeVector = newGaugeVec(reg, "knox_knowledge_max_hcl", "Highest HCL seen per originating node.", "node_id")
+
+	// Go runtime + process collectors come from the official library.
+	reg.MustRegister(prometheus.NewGoCollector())
+	reg.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))
+
+	return m
+}
+
+func newCounter(reg *prometheus.Registry, name, help string) prometheus.Counter {
+	c := prometheus.NewCounter(prometheus.CounterOpts{Name: name, Help: help})
+	reg.MustRegister(c)
+	return c
+}
+
+func newGauge(reg *prometheus.Registry, name, help string) prometheus.Gauge {
+	g := prometheus.NewGauge(prometheus.GaugeOpts{Name: name, Help: help})
+	reg.MustRegister(g)
+	return g
+}
+
+func newGaugeVec(reg *prometheus.Registry, name, help string, labels ...string) *prometheus.GaugeVec {
+	g := prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: name, Help: help}, labels)
+	reg.MustRegister(g)
+	return g
+}
+
+func (m *Metrics) nodeInfo(nodeID, name string) {
+	info := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "knox_node_info",
+		Help: "Node identity (always 1).",
+		ConstLabels: prometheus.Labels{
+			"node_id": nodeID,
+			"name":    name,
+		},
+	})
+	m.registry.MustRegister(info)
+	info.Set(1)
+}
+
+// Capture refresh the DB-derived gauges from a fresh snapshot.
+func (m *Metrics) Capture(s *db.MetricsSnapshot) {
+	m.observationsGauge.Reset()
+	for src, n := range s.BySource {
+		m.observationsGauge.WithLabelValues(src).Set(float64(n))
+	}
+	m.observationsLast24h.Set(float64(s.ObservationsLast24h))
+	m.entriesGauge.Set(float64(s.Entries))
+	m.projectsGauge.Set(float64(s.Projects))
+	m.sessionsGauge.Set(float64(s.Sessions))
+	m.pendingReflections.Set(float64(s.PendingReflections))
+	m.peersGauge.Set(float64(s.Peers))
+
+	m.threadsByStatus.Reset()
+	for st, n := range s.ThreadsByStatus {
+		m.threadsByStatus.WithLabelValues(st).Set(float64(n))
+	}
+
+	m.byOriginNode.Reset()
+	for nid, n := range s.ByOriginNode {
+		m.byOriginNode.WithLabelValues(nid).Set(float64(n))
+	}
+	m.knowledgeVector.Reset()
+	for nid, hcl := range s.KnowledgeVector {
+		m.knowledgeVector.WithLabelValues(nid).Set(float64(hcl))
+	}
+}
+
+// IncrementPull records a completed pull and its accepted observation count.
+func (m *Metrics) IncrementPull(newObs int) {
+	m.pullsTotal.Inc()
+	m.obsPulledTotal.Add(float64(newObs))
+}
+
+// IncrementPush records a completed push and its accepted observation count.
+func (m *Metrics) IncrementPush(newObs int) {
+	m.pushesTotal.Inc()
+	m.obsPushedTotal.Add(float64(newObs))
+}
+
+// IncrementErrors counts a failed gossip attempt.
+func (m *Metrics) IncrementErrors() { m.errorsTotal.Inc() }
+
+// Handler returns the /metrics scrape handler. On each scrape it refreshes
+// DB-derived gauges before rendering (cost is a few cheap aggregates).
+func (m *Metrics) Handler() http.Handler {
+	h := promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{})
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if s, err := m.kdb.MetricsSnapshot(); err == nil {
+			m.Capture(s)
+		}
+		h.ServeHTTP(w, r)
+	})
+}
@@ -0,0 +1,114 @@
+package metrics
+
+import (
+	"io"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/david/knox/internal/db"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+)
+
+func tmpKdb(t *testing.T) *db.KnoxDB {
+	t.Helper()
+	k, err := db.Open(filepath.Join(t.TempDir(), "index.db"))
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	t.Cleanup(func() { k.Close() })
+	return k
+}
+
+func seed(t *testing.T, k *db.KnoxDB, src string, n int) {
+	t.Helper()
+	for i := 0; i < n; i++ {
+		_, _, err := k.RecordObservation(db.ObservationRecord{
+			Fingerprint:     "fp-" + src + "-" + string(rune('a'+i)),
+			SourceID:        src,
+			SourcePath:      src,
+			Project:         "test",
+			ContentType:     "test",
+			Title:           src,
+			Summary:         "s",
+			CreatedAt:       "2026-08-29T00:00:00Z",
+			Confidence:      0.9,
+			IngesterVersion: "itest/v1",
+		})
+		if err != nil {
+			t.Fatalf("seed: %v", err)
+		}
+	}
+}
+
+func TestMetricsSnapshot(t *testing.T) {
+	k := tmpKdb(t)
+	seed(t, k, "git", 2)
+	seed(t, k, "browser-history", 3)
+
+	s, err := k.MetricsSnapshot()
+	if err != nil {
+		t.Fatalf("snapshot: %v", err)
+	}
+	if s.Observations != 5 {
+		t.Errorf("observations = %d, want 5", s.Observations)
+	}
+	if s.BySource["git"] != 2 || s.BySource["browser-history"] != 3 {
+		t.Errorf("by source = %v", s.BySource)
+	}
+	if s.KnowledgeVector[k.NodeID()] == 0 {
+		t.Errorf("knowledge vector missing own node")
+	}
+	if s.ByOriginNode[k.NodeID()] != 5 {
+		t.Errorf("by origin node = %v", s.ByOriginNode)
+	}
+}
+
+func TestMetricsScrape(t *testing.T) {
+	k := tmpKdb(t)
+	seed(t, k, "git", 2)
+
+	m := New(k, "testnode")
+	h := m.Handler()
+
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, httptest.NewRequest("GET", "/metrics", nil))
+
+	body, _ := io.ReadAll(rec.Body)
+	out := string(body)
+	for _, want := range []string{
+		`knox_node_info{name="testnode"`,
+		`knox_observations_total{source_id="git"} 2`,
+		`knox_gossip_pulls_total 0`,
+		"go_goroutines",
+		"process_cpu_seconds_total",
+	} {
+		if !strings.Contains(out, want) {
+			t.Errorf("scrape output missing %q", want)
+		}
+	}
+}
+
+func TestMetricsCounters(t *testing.T) {
+	k := tmpKdb(t)
+	m := New(k, "t")
+
+	// Manually drive counters through the Metrics API.
+	m.IncrementPull(3)
+	m.IncrementPush(7)
+	m.IncrementErrors()
+
+	if got := testutil.ToFloat64(m.pullsTotal); got != 1 {
+		t.Errorf("pullsTotal = %v, want 1", got)
+	}
+	if got := testutil.ToFloat64(m.obsPulledTotal); got != 3 {
+		t.Errorf("obsPulled = %v, want 3", got)
+	}
+	if got := testutil.ToFloat64(m.obsPushedTotal); got != 7 {
+		t.Errorf("obsPushed = %v, want 7", got)
+	}
+	if got := testutil.ToFloat64(m.errorsTotal); got != 1 {
+		t.Errorf("errorsTotal = %v, want 1", got)
+	}
+}