diff --git a/docs/gossip-protocol.md b/docs/gossip-protocol.md index e60ce96..71a57d3 100644 --- a/docs/gossip-protocol.md +++ b/docs/gossip-protocol.md @@ -249,6 +249,17 @@ reach a booting node. Config via env `KNOX_PEER_ADDR` / `KNOX_PEERS`. Verified e2e: two daemons, `diff` previewed 1655 peer-only fingerprints, `sync` pulled all and converged B to 1655 observations. +**M5 — Prometheus metrics.** DONE. `/metrics` served on a dedicated port +(`KNOX_METRICS_ADDR`, default `:8932`, separate from gossip). Uses +`prometheus/client_golang`; ships with the Go runtime and process collectors. +Gauge set: DB-derived gauges refreshed per scrape (`observations_total{source_id}`, +`observations_last_24h`, `entries`, `projects`, `sessions`, `pending_reflections`, +`threads_total{status}`, `peers`, `observations_by_node{node_id}`, +`knowledge_max_hcl{node_id}`) plus live gossip counters +(`pulls/pushes`, `observations_pulled/pushed`, `errors`) incremented during the +anti-entropy sweep, and `knox_node_info{node_id,name}` for scrape identity +(tracks issue #2). + ## 9. Future Work (explicitly out of M1–M4) - mDNS / rendezvous peer discovery. diff --git a/go.mod b/go.mod index b99e3f8..8c7ff8c 100644 --- a/go.mod +++ b/go.mod @@ -5,21 +5,30 @@ go 1.26 require ( github.com/fsnotify/fsnotify v1.8.0 github.com/mark3labs/mcp-go v0.17.0 + github.com/prometheus/client_golang v1.24.1 github.com/spf13/cobra v1.9.1 modernc.org/sqlite v1.37.1 ) require ( + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/google/uuid v1.6.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/ncruces/go-strftime v0.1.9 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.70.1 // indirect + github.com/prometheus/procfs v0.21.1 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/spf13/pflag v1.0.6 // indirect github.com/yosida95/uritemplate/v3 v3.0.2 // indirect golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0 // indirect - golang.org/x/sys v0.33.0 // indirect + golang.org/x/sys v0.47.0 // indirect + google.golang.org/protobuf v1.36.11 // indirect modernc.org/libc v1.65.7 // indirect modernc.org/mathutil v1.7.1 // indirect modernc.org/memory v1.11.0 // indirect diff --git a/go.sum b/go.sum index 1358997..65a8ea4 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,7 @@ +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -5,20 +9,36 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/fsnotify/fsnotify v1.8.0 h1:dAwr6QBTBZIkG8roQaJjGof0pp0EeF+tNV7YBP3F/8M= github.com/fsnotify/fsnotify v1.8.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= +github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/mark3labs/mcp-go v0.17.0 h1:5Ps6T7qXr7De/2QTqs9h6BKeZ/qdeUeGrgM5lPzi930= github.com/mark3labs/mcp-go v0.17.0/go.mod h1:KmJndYv7GIgcPVwEKJjNcbhVQ+hJGJhrCCB/9xITzpE= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4= github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU= +github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY= +github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc= +github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI= +github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= @@ -26,21 +46,27 @@ github.com/spf13/cobra v1.9.1 h1:CXSaggrXdbHK9CF+8ywj8Amf7PBRmPCOJugH954Nnlo= github.com/spf13/cobra v1.9.1/go.mod h1:nDyEzZ8ogv936Cinf6g1RU9MRY64Ir93oCnqb9wxYW0= github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o= github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= -github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= -github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0 h1:R84qjqJb5nVJMxqWYb3np9L5ZsaDtB+a39EqjV0JSUM= golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0/go.mod h1:S9Xr4PYopiDyqSyp5NjCrhFrqg6A5zA2E/iPHPhqnS8= golang.org/x/mod v0.24.0 h1:ZfthKaKaT4NrhGVZHO1/WDTwGES4De8KtWO0SIbNJMU= golang.org/x/mod v0.24.0/go.mod h1:IXM97Txy2VM4PJ3gI61r1YEk/gAj6zAHN3AdZt6S9Ww= -golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ= -golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= +golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw= -golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/tools v0.33.0 h1:4qz2S3zmRxbGIhDIAgjxvFutSvH5EfnsYrRBj0UI0bc= golang.org/x/tools v0.33.0/go.mod h1:CIJMaWEY88juyUfo7UbgPqbC8rU2OqfAV1h2Qp0oMYI= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/cmd/gossip.go b/internal/cmd/gossip.go index 695fc9f..dc3bf63 100644 --- a/internal/cmd/gossip.go +++ b/internal/cmd/gossip.go @@ -61,7 +61,7 @@ func NewGossipCmd(kdb *db.KnoxDB) *cobra.Command { if len(peers) == 0 { return fmt.Errorf("no peers configured (set KNOX_PEERS)") } - watch.Run(kdb, peers) + watch.Run(kdb, nil, peers) created, linked, err := watch.Reconcile(kdb) if err != nil { return err diff --git a/internal/db/metrics.go b/internal/db/metrics.go new file mode 100644 index 0000000..765ecf8 --- /dev/null +++ b/internal/db/metrics.go @@ -0,0 +1,108 @@ +package db + +// MetricsSnapshot holds the scrape-time gauges derived from the database. +type MetricsSnapshot struct { + Observations int + ObservationsLast24h int + BySource map[string]int + Entries int + Projects int + Sessions int + PendingReflections int + ThreadsByStatus map[string]int + Peers int + ByOriginNode map[string]int // node_id → observation count + KnowledgeVector map[string]int64 // node_id → max hcl + EarliestObservation string +} + +// MetricsSnapshot computes database-derived gauges for Prometheus scraping. +// All queries are cheap aggregations; nothing is stored or mutated. +func (k *KnoxDB) MetricsSnapshot() (*MetricsSnapshot, error) { + s := &MetricsSnapshot{ + BySource: make(map[string]int), + ThreadsByStatus: make(map[string]int), + ByOriginNode: make(map[string]int), + } + + if err := k.db.QueryRow("SELECT COUNT(*) FROM observations").Scan(&s.Observations); err != nil { + return nil, err + } + if err := k.db.QueryRow("SELECT COUNT(*) FROM observations WHERE collected_at > datetime('now', '-1 day')").Scan(&s.ObservationsLast24h); err != nil { + return nil, err + } + if err := k.db.QueryRow("SELECT COUNT(*) FROM entries").Scan(&s.Entries); err != nil { + return nil, err + } + if err := k.db.QueryRow("SELECT COUNT(DISTINCT project) FROM entries").Scan(&s.Projects); err != nil { + return nil, err + } + if err := k.db.QueryRow("SELECT COUNT(*) FROM sessions").Scan(&s.Sessions); err != nil { + return nil, err + } + if err := k.db.QueryRow("SELECT COUNT(*) FROM sessions WHERE indexed=0").Scan(&s.PendingReflections); err != nil { + return nil, err + } + if err := k.db.QueryRow("SELECT COUNT(*) FROM peers").Scan(&s.Peers); err != nil { + return nil, err + } + if err := k.db.QueryRow("SELECT COALESCE(MIN(collected_at),'') FROM observations").Scan(&s.EarliestObservation); err != nil { + return nil, err + } + + rows, err := k.db.Query("SELECT COALESCE(source_id,'unknown'), COUNT(*) FROM observations GROUP BY source_id") + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var src string + var n int + if err := rows.Scan(&src, &n); err != nil { + return nil, err + } + s.BySource[src] = n + } + if err := rows.Err(); err != nil { + return nil, err + } + + rows2, err := k.db.Query("SELECT COALESCE(status,'active'), COUNT(*) FROM threads GROUP BY status") + if err != nil { + return nil, err + } + defer rows2.Close() + for rows2.Next() { + var st string + var n int + if err := rows2.Scan(&st, &n); err != nil { + return nil, err + } + s.ThreadsByStatus[st] = n + } + if err := rows2.Err(); err != nil { + return nil, err + } + + rows3, err := k.db.Query("SELECT node_id, COUNT(*) FROM observations WHERE node_id<>'' GROUP BY node_id") + if err != nil { + return nil, err + } + defer rows3.Close() + for rows3.Next() { + var nid string + var n int + if err := rows3.Scan(&nid, &n); err != nil { + return nil, err + } + s.ByOriginNode[nid] = n + } + if err := rows3.Err(); err != nil { + return nil, err + } + + if s.KnowledgeVector, err = k.KnowledgeVector(); err != nil { + return nil, err + } + return s, nil +} \ No newline at end of file diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go new file mode 100644 index 0000000..2652583 --- /dev/null +++ b/internal/metrics/metrics.go @@ -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) + }) +} \ No newline at end of file diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go new file mode 100644 index 0000000..283ad5c --- /dev/null +++ b/internal/metrics/metrics_test.go @@ -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) + } +} \ No newline at end of file diff --git a/internal/watch/gossip.go b/internal/watch/gossip.go index bf9382a..313b845 100644 --- a/internal/watch/gossip.go +++ b/internal/watch/gossip.go @@ -19,14 +19,16 @@ import ( "time" "github.com/david/knox/internal/db" + "github.com/david/knox/internal/metrics" ) const defaultPort = "8931" type Node struct { - Kdb *db.KnoxDB - Name string - Addr string // advertised base URL, e.g. http://192.168.1.20:8931 + Kdb *db.KnoxDB + Name string + Addr string // advertised base URL, e.g. http://192.168.1.20:8931 + Metrics *metrics.Metrics } // pingResponse is the anti-entropy summary returned by /v1/ping. @@ -269,7 +271,9 @@ func (c *Client) Diff() (*DiffSummary, error) { // with any one node reveals who else is in the swarm, and those nodes are then // swept too. There is no relay of observations — only membership is shared; each // node pulls/pushes directly with every other node it learns about. -func Run(kdb *db.KnoxDB, static []string) { +// +// m, when non-nil, receives gossip event counters (nil for one-shot CLI runs). +func Run(kdb *db.KnoxDB, m *metrics.Metrics, static []string) { myID := kdb.NodeID() // Seed the work queue with static config plus persisted discoveries. @@ -292,6 +296,9 @@ func Run(kdb *db.KnoxDB, static []string) { p, err := c.Ping() if err != nil { log.Printf("[gossip] ping %s: %v", addr, err) + if m != nil { + m.IncrementErrors() + } continue } if p.NodeID == myID { @@ -323,13 +330,22 @@ func Run(kdb *db.KnoxDB, static []string) { n, err := c.Pull(remoteNode, localHCL, kdb) if err != nil { log.Printf("[gossip] pull %s@%s: %v", remoteNode, addr, err) + if m != nil { + m.IncrementErrors() + } continue } pulled += n } } + if m != nil { + m.IncrementPull(pulled) + } pushed, _ := c.Push(kdb, p.Vector) + if m != nil { + m.IncrementPush(pushed) + } if err := kdb.UpsertPeer(p.NodeID, addr, p.Name, vectorMax(p.Vector)); err != nil { log.Printf("[gossip] peer upsert: %v", err) @@ -368,6 +384,18 @@ func ListenAddr() (addr string) { return "localhost:" + defaultPort } +const defaultMetricsPort = "8932" + +// MetricsAddr returns the Prometheus scrape address (KNOX_METRICS_ADDR or +// default). It is a separate port from gossip so scraping never contends with +// the peer protocol. +func MetricsAddr() (addr string) { + if addr = os.Getenv("KNOX_METRICS_ADDR"); addr != "" { + return addr + } + return "localhost:" + defaultMetricsPort +} + // PeerAddrs returns the configured peer list (KNOX_PEERS, comma-separated). func PeerAddrs() []string { raw := os.Getenv("KNOX_PEERS") diff --git a/internal/watch/gossip_test.go b/internal/watch/gossip_test.go index 5e0f916..5d3ba34 100644 --- a/internal/watch/gossip_test.go +++ b/internal/watch/gossip_test.go @@ -60,8 +60,8 @@ func TestGossipConvergence(t *testing.T) { defer sb.Close() // A pulls from B, then B pulls from A (bidirectional sweep). - Run(a, []string{sb.URL}) - Run(b, []string{sa.URL}) + Run(a, nil, []string{sb.URL}) + Run(b, nil, []string{sa.URL}) av, err := a.KnowledgeVector() if err != nil { @@ -107,11 +107,11 @@ func TestGossipSwarmDiscovery(t *testing.T) { defer sc.Close() // A discovers B (A pings B) so A can advertise B to the swarm. - Run(a, []string{sb.URL}) + Run(a, nil, []string{sb.URL}) // C only knows A. A single sweep should surface B (membership in ping) // and pull B's observations directly. - Run(c, []string{sa.URL}) + Run(c, nil, []string{sa.URL}) // C must know B and hold all three origin logs. peers, err := c.ListPeers() @@ -227,12 +227,12 @@ func TestGossipIdempotent(t *testing.T) { sb := httptest.NewServer(nodeB.Handler()) defer sb.Close() - Run(b, []string{sa.URL}) + Run(b, nil, []string{sa.URL}) before, err := b.EntryCount() if err != nil { t.Fatal(err) } - Run(b, []string{sa.URL}) + Run(b, nil, []string{sa.URL}) after, err := b.EntryCount() if err != nil { t.Fatal(err) diff --git a/internal/watch/watch.go b/internal/watch/watch.go index ce2fa28..a921344 100644 --- a/internal/watch/watch.go +++ b/internal/watch/watch.go @@ -10,6 +10,7 @@ import ( "github.com/david/knox/internal/db" "github.com/david/knox/internal/ingest" + "github.com/david/knox/internal/metrics" "github.com/fsnotify/fsnotify" ) @@ -24,6 +25,7 @@ type Watcher struct { vault string debounce time.Duration fileIngesters []ingest.Ingester + metrics *metrics.Metrics } func New(kdb *db.KnoxDB, dirs []string) *Watcher { @@ -40,6 +42,7 @@ func New(kdb *db.KnoxDB, dirs []string) *Watcher { ingest.NewLogIngester(), ingest.NewSkillsIngester(), }, + metrics: metrics.New(kdb, "knox"), } } @@ -53,7 +56,7 @@ func (w *Watcher) Start() error { // Start the gossip server first so peers can reach us while the initial // seed is still ingesting. Vault/dirs are logged after the seed below. gossipAddr := ListenAddr() - node := &Node{Kdb: w.knoxDB, Name: "knox", Addr: gossipAddr} + node := &Node{Kdb: w.knoxDB, Name: "knox", Addr: gossipAddr, Metrics: w.metrics} srv := &http.Server{Addr: gossipAddr, Handler: node.Handler()} go func() { log.Printf("[knox] gossip listening on %s", gossipAddr) @@ -61,6 +64,17 @@ func (w *Watcher) Start() error { log.Printf("[knox] gossip server: %v", err) } }() + + // Prometheus scraping on a dedicated port (KNOX_METRICS_ADDR). + metricsAddr := MetricsAddr() + metricsSrv := &http.Server{Addr: metricsAddr, Handler: node.Metrics.Handler()} + go func() { + log.Printf("[knox] metrics listening on %s", metricsAddr) + if err := metricsSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Printf("[knox] metrics server: %v", err) + } + }() + if peers := PeerAddrs(); len(peers) > 0 { log.Printf("[knox] gossip peers: %v", peers) } @@ -447,7 +461,7 @@ func (w *Watcher) syncGossip() { before = n } - Run(w.knoxDB, peers) + Run(w.knoxDB, w.metrics, peers) // If new observations arrived, reconcile to pick up entries/threads they // imply (deterministic log → derived rebuild).