diskless-workstation/serving/fleet-manager/influx.go

// 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,
	"uptime_seconds": 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()),
		"uptime_seconds":   snap.GetUptimeSeconds(),
		// 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)
}