Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 david
|
876d2aa45f | ||
|
david
|
25a7112d8a |
+22
-4
@@ -125,14 +125,16 @@ a single node the rowid remains monotonic, so the current dedup query
|
||||
|
||||
### 5.1 Transport
|
||||
|
||||
Plain HTTP/JSON on a per-node advertized address (default port `8931`). Nodes
|
||||
discover peers via a static list in `settings` (M3). mDNS/rendezvous is future
|
||||
work.
|
||||
Plain HTTP/JSON on a per-node advertized address (default port `8931`). Peers
|
||||
are seeded from a static list (`KNOX_PEERS`), then the swarm discovers itself:
|
||||
each node advertises its known peer addresses in `/v1/ping`, and every sweep
|
||||
enqueues newly-learned nodes for direct contact (membership gossip — no relay of
|
||||
observations). mDNS/rendezvous is future work.
|
||||
|
||||
Endpoints:
|
||||
|
||||
```
|
||||
GET /v1/ping → { node_id, name, max_hcl }
|
||||
GET /v1/ping → { node_id, name, max_hcl, peers: [{peer_id, addr, name}] }
|
||||
GET /v1/log?after=<seq>&node=<id> → { cursor, rows: [observation...] } # pull
|
||||
POST /v1/obs/batch → body: [observation...]; reply: { accepted n, conflict n } # push
|
||||
GET /v1/diff → divergence summary (M4)
|
||||
@@ -142,6 +144,11 @@ GET /v1/diff → divergence summary (M4)
|
||||
|
||||
- **Knowledge vector:** each node tracks `peer_id → max_hcl consumed`. Anti-entropy
|
||||
is a pull: periodically (and on handshake) query each peer's `/v1/log?after=...`.
|
||||
- **Membership gossip:** `/v1/ping` includes the responding node's known peers
|
||||
(`peer_id`, `addr`, `name`). The caller merges them into its `peers` table and
|
||||
enqueues their addresses for direct sweeps. A new node therefore joins the
|
||||
whole swarm by configuring just one seed peer. Membership flows independently
|
||||
of data — a node never relays another's observations, only its address.
|
||||
- **Push:** on a new local observation, best-effort `POST /v1/obs/batch` to known
|
||||
peers. A node does **not** re-broadcast something it merely received (that peer
|
||||
already has it and will pull from its origin) — this is the echo/loop
|
||||
@@ -242,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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -142,6 +142,12 @@ func Open(path string) (*KnoxDB, error) {
|
||||
return nil, fmt.Errorf("create locator index: %w", err)
|
||||
}
|
||||
|
||||
// Thread idempotency index must come after the cluster_key migration (a
|
||||
// fresh DB has the column from Schema; an existing DB gets it above).
|
||||
if _, err := db.Exec(`CREATE UNIQUE INDEX IF NOT EXISTS idx_threads_cluster ON threads(cluster_key) WHERE cluster_key IS NOT NULL AND cluster_key != ''`); err != nil {
|
||||
return nil, fmt.Errorf("create cluster_key index: %w", err)
|
||||
}
|
||||
|
||||
return kdb, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -141,6 +141,18 @@ func (k *KnoxDB) UpsertPeer(peerID, addr, name string, maxHCL int64) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// MergePeer records a peer discovered indirectly (via another peer's ping).
|
||||
// Unlike UpsertPeer it never clobbers the cursor — a freshly learned address
|
||||
// has no known knowledge yet; the anti-entropy pull will set it on contact.
|
||||
func (k *KnoxDB) MergePeer(peerID, addr, name string) error {
|
||||
_, err := k.db.Exec(
|
||||
`INSERT INTO peers (peer_id, addr, name) VALUES (?, ?, ?)
|
||||
ON CONFLICT(peer_id) DO UPDATE SET addr=?, name=?`,
|
||||
peerID, addr, name, addr, name,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// ListPeers returns known peers ordered by first-seen.
|
||||
func (k *KnoxDB) ListPeers() ([]Peer, error) {
|
||||
rows, err := k.db.Query(`SELECT peer_id, COALESCE(addr,''), COALESCE(name,''), COALESCE(last_handshake,''), COALESCE(cursor,0), COALESCE(created_at,'') FROM peers ORDER BY created_at`)
|
||||
@@ -181,6 +193,61 @@ type Peer struct {
|
||||
CreatedAt string
|
||||
}
|
||||
|
||||
// PeerInfo is the shareable (non-secret) subset of a peer that /v1/ping
|
||||
// advertises so other nodes can discover the swarm.
|
||||
type PeerInfo struct {
|
||||
PeerID string `json:"peer_id"`
|
||||
Addr string `json:"addr"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
|
||||
// ShareablePeers returns the peers this node knows about, for dissemination in
|
||||
// ping responses. Self and peers without an address are excluded.
|
||||
func (k *KnoxDB) ShareablePeers() ([]PeerInfo, error) {
|
||||
rows, err := k.db.Query(
|
||||
`SELECT peer_id, COALESCE(addr,''), COALESCE(name,'') FROM peers WHERE addr<>'' AND peer_id<>? ORDER BY peer_id`,
|
||||
k.NodeID(),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []PeerInfo
|
||||
for rows.Next() {
|
||||
var p PeerInfo
|
||||
if err := rows.Scan(&p.PeerID, &p.Addr, &p.Name); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, p)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// MaxHCLForNode returns the maximum HCL this node holds for a given origin
|
||||
// node, or 0 if none.
|
||||
func (k *KnoxDB) MaxHCLForNode(nodeID string) int64 {
|
||||
var m int64
|
||||
if err := k.db.QueryRow(`SELECT MAX(hcl) FROM observations WHERE node_id=?`, nodeID).Scan(&m); err != nil {
|
||||
return 0
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// SwarmPeerAddrs returns the addresses of all known peers (shareable set). It
|
||||
// is what the sweep iterates after a bootstrap join.
|
||||
func (k *KnoxDB) SwarmPeerAddrs() ([]string, error) {
|
||||
peers, err := k.ShareablePeers()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
addrs := make([]string, 0, len(peers))
|
||||
for _, p := range peers {
|
||||
addrs = append(addrs, p.Addr)
|
||||
}
|
||||
return addrs, nil
|
||||
}
|
||||
|
||||
// DistinctFingerprints returns the set of all observed fingerprints — the
|
||||
// ground-truth index of what this node knows. Used by gossip diff.
|
||||
func (k *KnoxDB) DistinctFingerprints() (map[string]bool, error) {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -105,8 +105,6 @@ CREATE TABLE IF NOT EXISTS threads (
|
||||
cluster_key TEXT
|
||||
);
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_threads_cluster ON threads(cluster_key) WHERE cluster_key IS NOT NULL AND cluster_key != '';
|
||||
|
||||
-- SETTINGS: key-value store for runtime configuration
|
||||
CREATE TABLE IF NOT EXISTS settings (
|
||||
key TEXT PRIMARY KEY,
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
+80
-15
@@ -19,22 +19,25 @@ 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.
|
||||
type pingResponse struct {
|
||||
NodeID string `json:"node_id"`
|
||||
Name string `json:"name"`
|
||||
Vector map[string]int64 `json:"vector"` // node_id → max hcl
|
||||
MaxHCL *int64 `json:"max_hcl,omitempty"`
|
||||
NodeID string `json:"node_id"`
|
||||
Name string `json:"name"`
|
||||
Vector map[string]int64 `json:"vector"` // node_id → max hcl
|
||||
Peers []db.PeerInfo `json:"peers"` // swarm membership this node knows
|
||||
MaxHCL *int64 `json:"max_hcl,omitempty"`
|
||||
}
|
||||
|
||||
func (n *Node) Handler() http.Handler {
|
||||
@@ -52,10 +55,16 @@ func (n *Node) handlePing(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
peers, err := n.Kdb.ShareablePeers()
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
writeJSON(w, pingResponse{
|
||||
NodeID: n.Kdb.NodeID(),
|
||||
Name: n.Name,
|
||||
Vector: vector,
|
||||
Peers: peers,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -255,27 +264,62 @@ func (c *Client) Diff() (*DiffSummary, error) {
|
||||
return &d, nil
|
||||
}
|
||||
|
||||
// Run executes one anti-entropy sweep against the given peer addresses.
|
||||
func Run(kdb *db.KnoxDB, peers []string) {
|
||||
// Run executes one anti-entropy + membership sweep.
|
||||
//
|
||||
// Peers are a fusion of the statically configured list (KNOX_PEERS) and peers
|
||||
// previously discovered and persisted in the peers table (swarm join): syncing
|
||||
// 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.
|
||||
//
|
||||
// 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()
|
||||
for _, addr := range peers {
|
||||
addr = strings.TrimSpace(addr)
|
||||
if addr == "" {
|
||||
continue
|
||||
}
|
||||
if strings.HasPrefix(addr, kdb.NodeID()+":") {
|
||||
|
||||
// Seed the work queue with static config plus persisted discoveries.
|
||||
persisted, _ := kdb.SwarmPeerAddrs()
|
||||
work := make([]string, 0, len(static)+len(persisted))
|
||||
work = append(work, static...)
|
||||
work = append(work, persisted...)
|
||||
|
||||
seen := make(map[string]bool) // addr → handled (also suppresses self)
|
||||
queue := 0
|
||||
for queue < len(work) {
|
||||
addr := strings.TrimSpace(work[queue])
|
||||
queue++
|
||||
if addr == "" || seen[addr] {
|
||||
continue
|
||||
}
|
||||
seen[addr] = true
|
||||
|
||||
c := &Client{Addr: addr, Timeout: 10 * time.Second}
|
||||
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 {
|
||||
continue // never talk to ourselves (or an aliased address)
|
||||
}
|
||||
|
||||
// Membership discovery: learn who else is in the swarm and enqueue
|
||||
// their addresses for direct sweeps.
|
||||
for _, pi := range p.Peers {
|
||||
if pi.PeerID == "" || pi.PeerID == myID || pi.Addr == "" {
|
||||
continue
|
||||
}
|
||||
if err := kdb.MergePeer(pi.PeerID, pi.Addr, pi.Name); err != nil {
|
||||
log.Printf("[gossip] merge peer %s: %v", pi.PeerID, err)
|
||||
continue
|
||||
}
|
||||
if !seen[pi.Addr] {
|
||||
work = append(work, pi.Addr)
|
||||
}
|
||||
}
|
||||
|
||||
pulled := 0
|
||||
for remoteNode, remoteHCL := range p.Vector {
|
||||
localHCL := kdbVectorGet(kdb, remoteNode)
|
||||
@@ -286,13 +330,22 @@ func Run(kdb *db.KnoxDB, peers []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)
|
||||
@@ -331,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")
|
||||
|
||||
@@ -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 {
|
||||
@@ -84,8 +84,77 @@ func TestGossipConvergence(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestGossipDiff ensures /v1/diff reports per-node observation fingerprints and
|
||||
// tombstoned thread divergence.
|
||||
// TestGossipSwarmDiscovery: C only knows A. A knows B. When C sweeps A, it must
|
||||
// learn about B through A's ping, enqueue B, and pull B's observations — with no
|
||||
// direct configuration of B (no relay of data, only membership).
|
||||
func TestGossipSwarmDiscovery(t *testing.T) {
|
||||
a := tmpKnoxDB(t)
|
||||
b := tmpKnoxDB(t)
|
||||
c := tmpKnoxDB(t)
|
||||
|
||||
seedObs(a, "AAA")
|
||||
seedObs(b, "BBB")
|
||||
seedObs(c, "CCC")
|
||||
|
||||
nodeA := &Node{Kdb: a, Name: "A"}
|
||||
sa := httptest.NewServer(nodeA.Handler())
|
||||
defer sa.Close()
|
||||
nodeB := &Node{Kdb: b, Name: "B"}
|
||||
sb := httptest.NewServer(nodeB.Handler())
|
||||
defer sb.Close()
|
||||
nodeC := &Node{Kdb: c, Name: "C"}
|
||||
sc := httptest.NewServer(nodeC.Handler())
|
||||
defer sc.Close()
|
||||
|
||||
// A discovers B (A pings B) so A can advertise B to the swarm.
|
||||
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, nil, []string{sa.URL})
|
||||
|
||||
// C must know B and hold all three origin logs.
|
||||
peers, err := c.ListPeers()
|
||||
if err != nil {
|
||||
t.Fatalf("list peers: %v", err)
|
||||
}
|
||||
foundB := false
|
||||
for _, p := range peers {
|
||||
if p.PeerID == b.NodeID() {
|
||||
foundB = true
|
||||
}
|
||||
}
|
||||
if !foundB {
|
||||
t.Fatalf("C did not discover B via A's membership list; peers=%v", peers)
|
||||
}
|
||||
|
||||
vec, err := c.KnowledgeVector()
|
||||
if err != nil {
|
||||
t.Fatalf("c vector: %v", err)
|
||||
}
|
||||
if len(vec) != 3 {
|
||||
t.Errorf("C should hold 3 origin logs (A, B, C), got %v", vec)
|
||||
}
|
||||
|
||||
// C's copy of B's log must match B's own max hcl.
|
||||
bmax := c.MaxHCLForNode(b.NodeID())
|
||||
bm, err := bNodeMax(b)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if bmax != bm {
|
||||
t.Errorf("C max hcl for B=%d, B reports %d", bmax, bm)
|
||||
}
|
||||
}
|
||||
|
||||
// bNodeMax is a helper reading the max hcl B holds for its own node.
|
||||
func bNodeMax(b *db.KnoxDB) (int64, error) {
|
||||
rows, err := b.KnowledgeVector()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return rows[b.NodeID()], nil
|
||||
}
|
||||
func TestGossipDiff(t *testing.T) {
|
||||
a := tmpKnoxDB(t)
|
||||
b := tmpKnoxDB(t)
|
||||
@@ -158,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)
|
||||
|
||||
+16
-2
@@ -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).
|
||||
|
||||
Reference in New Issue
Block a user