fleet-manager: checkpoint 4 — InfluxDB metrics gateway (write + Flux read)
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).
This commit is contained in:
@@ -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)
|
||||
}
|
||||
Reference in New Issue
Block a user