Audiobuzz
9c86cd4c5a
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).
74 lines
2.1 KiB
Go
74 lines
2.1 KiB
Go
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")
|
|
}
|
|
}
|