From 9c86cd4c5a6fbb79ad298d7d36fa54602d4b5703 Mon Sep 17 00:00:00 2001 From: Audiobuzz Date: Sat, 4 Jul 2026 03:59:18 +0000 Subject: [PATCH] =?UTF-8?q?fleet-manager:=20checkpoint=204=20=E2=80=94=20I?= =?UTF-8?q?nfluxDB=20metrics=20gateway=20(write=20+=20Flux=20read)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The manager is the ONLY thing that talks to InfluxDB, both directions; hosts hold no token and never reach it. - influx.go: influxSink implements MetricsSink — each agent HealthSnapshot is written as a fleet_health point, identity (serial/hostnqn/starship) as TAGS from the AUTHENTICATED stream (a host can only write its own series), telemetry as fields. QueryField answers the SPA's read: a Flux query built with a strict FIELD WHITELIST + hex-serial guard + parsed-duration range, so no caller input is interpolated raw (injection-safe). Server-side influxdb-client-go/v2. - admin.go: GET /api/hosts/{serial}/metrics?field=&range= -> time-series JSON; 503 when InfluxDB isn't wired, 400 on a bad range, 404 for unknown host. - main.go: sink created when FLEET_MGR_INFLUX_URL+_TOKEN set, fed to the agent hub; nil -> no-op sink so the manager runs without metrics. Tests: snapshotPoint tags/fields/time, fluxQuery contents + %q quoting, and QueryField rejects a non-hex serial + non-whitelisted field. build/test/vet/gofmt clean. Live InfluxDB path validated in the compose bench (ckpt 6). --- serving/fleet-manager/README.md | 14 ++- serving/fleet-manager/admin.go | 39 +++++++- serving/fleet-manager/go.mod | 5 + serving/fleet-manager/go.sum | 23 +++++ serving/fleet-manager/influx.go | 140 +++++++++++++++++++++++++++ serving/fleet-manager/influx_test.go | 73 ++++++++++++++ serving/fleet-manager/main.go | 28 +++++- 7 files changed, 316 insertions(+), 6 deletions(-) create mode 100644 serving/fleet-manager/influx.go create mode 100644 serving/fleet-manager/influx_test.go diff --git a/serving/fleet-manager/README.md b/serving/fleet-manager/README.md index 938827e..89355d8 100644 --- a/serving/fleet-manager/README.md +++ b/serving/fleet-manager/README.md @@ -66,7 +66,7 @@ inspectable, reproducible; a DB swap is a drop-in behind the `Store` interface. | `GET /api/enrol/pending` | ✅ ckpt 3 — proxies the broker queue | | `POST /api/enrol/{id}/approve` | ✅ ckpt 3 — signs via broker + **births inventory** (identity/binding/enrolled) | | `POST /api/enrol/{id}/deny` | ✅ ckpt 3 | -| `GET /api/hosts/{serial}/metrics` | ⏳ ckpt 4 (InfluxDB/Flux) | +| `GET /api/hosts/{serial}/metrics?field=&range=` | ✅ ckpt 4 — Flux time-series (field whitelist) | The authorise pillar drives the enrol-broker's admin plane over mTLS (the manager is a service client, gated by the dedicated admin CA) asking for JSON, so an @@ -95,6 +95,10 @@ inventory for offline delivery. Metrics writing is behind a `MetricsSink` | `FLEET_MGR_BROKER_URL` | — | enrol-broker admin base URL (e.g. `https://127.0.0.1:9443`). Authorise pillar enabled only when set. | | `FLEET_MGR_BROKER_CLIENT_CERT` / `_KEY` | — | manager's client cert for the broker admin mTLS (admin CA) | | `FLEET_MGR_BROKER_CA` | — | CA that signs the broker's admin server cert | +| `FLEET_MGR_INFLUX_URL` | — | InfluxDB URL (private mgmt net). Metrics enabled only when URL + token set. | +| `FLEET_MGR_INFLUX_TOKEN` | — | InfluxDB token — **server-side only**, never on hosts | +| `FLEET_MGR_INFLUX_ORG` | `fleet` | InfluxDB org | +| `FLEET_MGR_INFLUX_BUCKET` | `fleet_health` | InfluxDB bucket | | `FLEET_MGR_INVENTORY` | `/var/lib/fleet-manager/inventory.json` | inventory file (mount a volume) | | `FLEET_MGR_SEED` | `false` | inject demo hosts so the SPA has data — dev only | | `HOSTNQN_BASE` | `nqn.2026-06.org.example.fleet:host-` | hostnqn prefix (match the broker) | @@ -121,5 +125,9 @@ curl -s localhost:9500/api/hosts | jq '.hosts[] | {name:.identity.starship_name, `Accept`-gated JSON approve response (browsers/`verify-serving.sh` unchanged); the manager drives it over mTLS and births the inventory record on approve. httptest-mock-broker test. -- Next: InfluxDB read/write (ckpt 4) → themed SPA (ckpt 5) → Docker/compose + PKI - + docs (ckpt 6). +- **Checkpoint 4 (done):** InfluxDB metrics gateway — snapshots written as points + tagged with the authenticated identity (`MetricsSink`→`influxSink`), and a + read API (`/api/hosts/{serial}/metrics`) that runs a Flux query with a **field + whitelist** + hex-serial guard (no caller input reaches the query raw). Unit + tests for point construction, query building, and unsafe-input rejection. +- Next: themed SPA (ckpt 5) → Docker/compose + PKI + docs (ckpt 6). diff --git a/serving/fleet-manager/admin.go b/serving/fleet-manager/admin.go index 85ae22d..d561680 100644 --- a/serving/fleet-manager/admin.go +++ b/serving/fleet-manager/admin.go @@ -8,6 +8,7 @@ package main import ( + "context" "encoding/json" "io" "log" @@ -35,7 +36,8 @@ func (s *server) routes(mux *http.ServeMux) { mux.HandleFunc("POST /api/enrol/{id}/approve", s.handleEnrolApprove) mux.HandleFunc("POST /api/enrol/{id}/deny", s.handleEnrolDeny) - // METRICS routes (/api/hosts/{serial}/metrics) — checkpoint 4. + // METRICS — time-series for one host/field, read from InfluxDB via the manager. + mux.HandleFunc("GET /api/hosts/{serial}/metrics", s.handleMetrics) } func (s *server) handleListHosts(w http.ResponseWriter, r *http.Request) { @@ -141,6 +143,41 @@ func (s *server) handleEnrolApprove(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, born) } +// ---- METRICS (InfluxDB time-series via the manager gateway) --------------- + +func (s *server) handleMetrics(w http.ResponseWriter, r *http.Request) { + if s.influx == nil { + writeErr(w, http.StatusServiceUnavailable, "metrics not configured (no InfluxDB wired)") + return + } + serial := r.PathValue("serial") + if _, ok := s.store.Get(serial); !ok { + writeErr(w, http.StatusNotFound, "no such host") + return + } + field := r.URL.Query().Get("field") + if field == "" { + field = "tmpfs_used_pct" + } + rng := time.Hour + if rs := r.URL.Query().Get("range"); rs != "" { + d, err := time.ParseDuration(rs) + if err != nil || d <= 0 { + writeErr(w, http.StatusBadRequest, "invalid range (use e.g. 30m, 1h, 24h)") + return + } + rng = d + } + ctx, cancel := context.WithTimeout(r.Context(), 15*time.Second) + defer cancel() + pts, err := s.influx.QueryField(ctx, serial, field, rng) + if err != nil { + writeErr(w, http.StatusBadGateway, "metrics query: "+err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"serial": serial, "field": field, "range": rng.String(), "points": pts}) +} + func (s *server) handleEnrolDeny(w http.ResponseWriter, r *http.Request) { if s.broker == nil { writeErr(w, http.StatusServiceUnavailable, "authorise plane not configured (no enrol-broker wired)") diff --git a/serving/fleet-manager/go.mod b/serving/fleet-manager/go.mod index c50807c..cece8ff 100644 --- a/serving/fleet-manager/go.mod +++ b/serving/fleet-manager/go.mod @@ -14,11 +14,16 @@ module fleet-manager go 1.25.0 require ( + github.com/influxdata/influxdb-client-go/v2 v2.14.0 google.golang.org/grpc v1.82.0 google.golang.org/protobuf v1.36.11 ) require ( + github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 // indirect + github.com/oapi-codegen/runtime v1.0.0 // indirect golang.org/x/net v0.53.0 // indirect golang.org/x/sys v0.43.0 // indirect golang.org/x/text v0.36.0 // indirect diff --git a/serving/fleet-manager/go.sum b/serving/fleet-manager/go.sum index 0815dbd..c1b50cf 100644 --- a/serving/fleet-manager/go.sum +++ b/serving/fleet-manager/go.sum @@ -1,5 +1,12 @@ +github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk= +github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ= +github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk= +github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvFv1sNto9p6w= 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/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +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= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= @@ -10,6 +17,20 @@ 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/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/influxdata/influxdb-client-go/v2 v2.14.0 h1:AjbBfJuq+QoaXNcrova8smSjwJdUHnwvfjMF71M1iI4= +github.com/influxdata/influxdb-client-go/v2 v2.14.0/go.mod h1:Ahpm3QXKMJslpXl3IftVLVezreAUtBOTZssDrjZEFHI= +github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 h1:W9WBk7wlPfJLvMCdtV4zPulc4uCPrlywQOmbFOhgQNU= +github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo= +github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE= +github.com/oapi-codegen/runtime v1.0.0 h1:P4rqFX5fMFWqRzY9M/3YF9+aPSPPB06IzP2P7oOxrWo= +github.com/oapi-codegen/runtime v1.0.0/go.mod h1:LmCUMQuPB4M/nLXilQXhHw+BLZdDb18B34OO356yJ/A= +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/spkg/bom v0.0.0-20160624110644-59b7046e48ad/go.mod h1:qLr4V1qq6nMqFKkMo8ZTx3f+BZEkzsRUY10Xsm2mwU0= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= @@ -36,3 +57,5 @@ google.golang.org/grpc v1.82.0 h1:vguDnZUPjE26w09A63VoxZPnvPjB5Riyc0mkXPFmAIU= google.golang.org/grpc v1.82.0/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +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/serving/fleet-manager/influx.go b/serving/fleet-manager/influx.go new file mode 100644 index 0000000..1ee09e9 --- /dev/null +++ b/serving/fleet-manager/influx.go @@ -0,0 +1,140 @@ +// influx.go — the metrics gateway. The manager is the ONLY thing that talks to +// InfluxDB (both directions): it writes each agent HealthSnapshot as a time-series +// point (tagged with the AUTHENTICATED host identity, so a host can only write its +// own series), and it answers the SPA's read queries via Flux — hosts never hold +// an InfluxDB token or reach InfluxDB. See docs/fleet-management-design.md. +package main + +import ( + "context" + "fmt" + "log" + "regexp" + "time" + + influxdb2 "github.com/influxdata/influxdb-client-go/v2" + "github.com/influxdata/influxdb-client-go/v2/api" + "github.com/influxdata/influxdb-client-go/v2/api/write" + + fleetpb "fleet-manager/proto" +) + +const influxMeasurement = "fleet_health" + +// metricFields is the whitelist of queryable numeric fields. A read request's +// ?field= MUST be one of these — never interpolate caller input into Flux. +var metricFields = map[string]bool{ + "load_avg": true, "cpu_pct": true, "temp_c": true, "mem_used_pct": true, + "tmpfs_used_pct": true, "chromium_crashes": true, "kiosk_active": true, +} + +var serialRe = regexp.MustCompile(`^[0-9a-fA-F]+$`) // cert serials are hex + +type influxConfig struct { + url, token, org, bucket string +} + +type influxSink struct { + client influxdb2.Client + write api.WriteAPIBlocking + query api.QueryAPI + bucket string +} + +func newInfluxSink(c influxConfig) *influxSink { + cl := influxdb2.NewClient(c.url, c.token) + return &influxSink{ + client: cl, + write: cl.WriteAPIBlocking(c.org, c.bucket), + query: cl.QueryAPI(c.org), + bucket: c.bucket, + } +} + +func (s *influxSink) Close() { s.client.Close() } + +// WriteSnapshot implements MetricsSink. +func (s *influxSink) WriteSnapshot(serial, hostnqn, starship string, snap *fleetpb.HealthSnapshot, at time.Time) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := s.write.WritePoint(ctx, snapshotPoint(serial, hostnqn, starship, snap, at)); err != nil { + log.Printf("influx write %s: %v", serial, err) + } +} + +// snapshotPoint builds the InfluxDB point (pure — unit-tested). Identity rides as +// TAGS (indexed, low-cardinality-ish per fleet); telemetry as FIELDS. +func snapshotPoint(serial, hostnqn, starship string, snap *fleetpb.HealthSnapshot, at time.Time) *write.Point { + tags := map[string]string{"serial": serial} + if hostnqn != "" { + tags["hostnqn"] = hostnqn + } + if starship != "" { + tags["starship"] = starship + } + fields := map[string]any{ + "load_avg": snap.GetLoadAvg(), + "cpu_pct": snap.GetCpuPct(), + "temp_c": snap.GetTempC(), + "mem_used_pct": snap.GetMemUsedPct(), + "tmpfs_used_pct": snap.GetTmpfsUsedPct(), + "chromium_crashes": int64(snap.GetChromiumCrashes()), + "kiosk_active": boolToInt(snap.GetKioskActive()), + // string status fields kept as fields too (handy for annotating charts) + "nvme_root": snap.GetNvmeRoot(), + "auth": snap.GetAuth(), + "selinux": snap.GetSelinux(), + "chrony": snap.GetChrony(), + } + return influxdb2.NewPoint(influxMeasurement, tags, fields, at) +} + +func boolToInt(b bool) int64 { + if b { + return 1 + } + return 0 +} + +type metricPoint struct { + Time time.Time `json:"t"` + Value any `json:"v"` +} + +// QueryField returns the time-series for one whitelisted field of one host. +func (s *influxSink) QueryField(ctx context.Context, serial, field string, rng time.Duration) ([]metricPoint, error) { + if !serialRe.MatchString(serial) { + return nil, fmt.Errorf("invalid serial") + } + if !metricFields[field] { + return nil, fmt.Errorf("unknown field %q", field) + } + res, err := s.query.Query(ctx, fluxQuery(s.bucket, serial, field, rng)) + if err != nil { + return nil, err + } + defer res.Close() + var out []metricPoint + for res.Next() { + out = append(out, metricPoint{Time: res.Record().Time(), Value: res.Record().Value()}) + } + if err := res.Err(); err != nil { + return nil, err + } + return out, nil +} + +// fluxQuery builds the read query (pure — unit-tested). serial and field are +// validated by the caller (hex / whitelist), and the range is a parsed duration +// rendered as an integer second count, so NOTHING here is raw caller input. +func fluxQuery(bucket, serial, field string, rng time.Duration) string { + secs := int64(rng.Seconds()) + if secs <= 0 { + secs = 3600 + } + return fmt.Sprintf(`from(bucket:%q) + |> range(start: -%ds) + |> filter(fn: (r) => r._measurement == %q and r.serial == %q and r._field == %q) + |> keep(columns: ["_time","_value"])`, + bucket, secs, influxMeasurement, serial, field) +} diff --git a/serving/fleet-manager/influx_test.go b/serving/fleet-manager/influx_test.go new file mode 100644 index 0000000..ea4a86d --- /dev/null +++ b/serving/fleet-manager/influx_test.go @@ -0,0 +1,73 @@ +package main + +import ( + "context" + "strings" + "testing" + "time" + + fleetpb "fleet-manager/proto" +) + +func TestSnapshotPoint(t *testing.T) { + at := time.Unix(1700000000, 0).UTC() + snap := &fleetpb.HealthSnapshot{ + LoadAvg: 0.5, CpuPct: 12, TmpfsUsedPct: 63, ChromiumCrashes: 2, KioskActive: true, NvmeRoot: "ok", + } + p := snapshotPoint("beef01", "nqn:host-beef01", "orion-drifter-07", snap, at) + + if p.Name() != influxMeasurement { + t.Errorf("measurement = %q", p.Name()) + } + tags := map[string]string{} + for _, tg := range p.TagList() { + tags[tg.Key] = tg.Value + } + if tags["serial"] != "beef01" || tags["starship"] != "orion-drifter-07" || tags["hostnqn"] != "nqn:host-beef01" { + t.Errorf("identity tags = %+v", tags) + } + fields := map[string]any{} + for _, f := range p.FieldList() { + fields[f.Key] = f.Value + } + if fields["tmpfs_used_pct"] != float64(63) { + t.Errorf("tmpfs_used_pct field = %v (%T)", fields["tmpfs_used_pct"], fields["tmpfs_used_pct"]) + } + if fields["kiosk_active"] != int64(1) { + t.Errorf("kiosk_active = %v", fields["kiosk_active"]) + } + if fields["chromium_crashes"] != int64(2) { + t.Errorf("chromium_crashes = %v", fields["chromium_crashes"]) + } + if fields["nvme_root"] != "ok" { + t.Errorf("nvme_root = %v", fields["nvme_root"]) + } + if !p.Time().Equal(at) { + t.Errorf("time = %v", p.Time()) + } +} + +func TestFluxQueryContentsAndQuoting(t *testing.T) { + q := fluxQuery("fleet_health", "beef01", "tmpfs_used_pct", time.Hour) + for _, want := range []string{ + `from(bucket:"fleet_health")`, + `range(start: -3600s)`, + `r._measurement == "fleet_health"`, + `r.serial == "beef01"`, + `r._field == "tmpfs_used_pct"`, + } { + if !strings.Contains(q, want) { + t.Errorf("flux query missing %q in:\n%s", want, q) + } + } +} + +func TestQueryFieldRejectsUnsafeInput(t *testing.T) { + s := &influxSink{bucket: "b"} // query is never reached for these + if _, err := s.QueryField(context.Background(), "not-hex!", "tmpfs_used_pct", time.Hour); err == nil { + t.Error("expected rejection of a non-hex serial") + } + if _, err := s.QueryField(context.Background(), "beef01", "evil) |> drop(", time.Hour); err == nil { + t.Error("expected rejection of a non-whitelisted field") + } +} diff --git a/serving/fleet-manager/main.go b/serving/fleet-manager/main.go index 96c6dc2..10b21c4 100644 --- a/serving/fleet-manager/main.go +++ b/serving/fleet-manager/main.go @@ -42,6 +42,9 @@ type config struct { brokerClientKey string brokerCA string + // InfluxDB (the manager is the only reader/writer; token is server-side only). + influx influxConfig + inventoryPath string seed bool hostnqnBase string @@ -65,6 +68,13 @@ func loadConfig() config { brokerClientKey: env("FLEET_MGR_BROKER_CLIENT_KEY", ""), brokerCA: env("FLEET_MGR_BROKER_CA", ""), + influx: influxConfig{ + url: env("FLEET_MGR_INFLUX_URL", ""), + token: env("FLEET_MGR_INFLUX_TOKEN", ""), + org: env("FLEET_MGR_INFLUX_ORG", "fleet"), + bucket: env("FLEET_MGR_INFLUX_BUCKET", "fleet_health"), + }, + inventoryPath: env("FLEET_MGR_INVENTORY", "/var/lib/fleet-manager/inventory.json"), seed: env("FLEET_MGR_SEED", "") == "true", hostnqnBase: env("HOSTNQN_BASE", "nqn.2026-06.org.example.fleet:host-"), @@ -77,6 +87,7 @@ type server struct { store Store hub *agentHub // live agent streams; nil if the agent plane isn't configured broker *brokerClient // enrol-broker admin client; nil if authorise isn't configured + influx *influxSink // metrics gateway; nil if InfluxDB isn't configured } func main() { @@ -106,11 +117,24 @@ func main() { log.Printf("authorise pillar DISABLED (set FLEET_MGR_BROKER_URL + client cert/key + CA to enable)") } + // Metrics gateway: the manager is the only reader/writer of InfluxDB. nil (a + // no-op MetricsSink) when unconfigured, so the manager runs without metrics. + var sink MetricsSink + if cfg.influx.url != "" && cfg.influx.token != "" { + s.influx = newInfluxSink(cfg.influx) + defer s.influx.Close() + sink = s.influx + log.Printf("InfluxDB metrics gateway -> %s (org=%s bucket=%s)", cfg.influx.url, cfg.influx.org, cfg.influx.bucket) + } else { + log.Printf("InfluxDB DISABLED (set FLEET_MGR_INFLUX_URL + _TOKEN to enable metrics)") + } + // Agent plane (gRPC/mTLS, fleet CA). Started only when configured; the admin // plane can run standalone (e.g. before any host is enrolled). The hub is the - // live-stream registry the lifecycle API pushes desired-state through. + // live-stream registry the lifecycle API pushes desired-state through, and it + // writes each snapshot to the metrics sink. if cfg.agentCert != "" && cfg.agentKey != "" && cfg.agentClientCA != "" { - s.hub = newAgentHub(store, nil) // InfluxDB sink wired in checkpoint 4 + s.hub = newAgentHub(store, sink) go func() { if err := serveAgent(cfg, s.hub); err != nil { log.Fatalf("agent plane: %v", err)