diff --git a/Makefile b/Makefile index 54b46bb..3b37609 100644 --- a/Makefile +++ b/Makefile @@ -23,6 +23,7 @@ fuzz: $(GO) test ./plugins/push -run '^$$' -fuzz '^FuzzParse$$' -fuzztime $(FUZZTIME) $(GO) test ./plugins/push -run '^$$' -fuzz '^FuzzDecode$$' -fuzztime $(FUZZTIME) $(GO) test ./plugins/vmagent -run '^$$' -fuzz '^FuzzDecode$$' -fuzztime $(FUZZTIME) + $(GO) test ./plugins/boxagent -run '^$$' -fuzz '^FuzzDecode$$' -fuzztime $(FUZZTIME) lint: golangci-lint run ./... diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index 0844dcc..34cfb33 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -271,9 +271,9 @@ ingest: key_file: /etc/monitor/tls/ingest.key trusted_proxies: [] max_body: 64KiB - rate_limit: # per ingest token - per_second: 1 - burst: 10 + rate_limit: # per ingest token (box.agent beats once a second) + per_second: 2 + burst: 20 mqtt: # Embedded broker. External-broker mode is configured per tenant through @@ -407,6 +407,7 @@ usage: rtsp.stream: 2 mqtt.connection: 0 # an MQTT sensor bills once, through its push.mqtt check (confirmed 2026-10-07) monitor.self: 0 # the engine's own health check (platform tenant) + box.agent: 2 # each monitored LLM box (TBD with billing) vm.agent: 2 # each monitored PlusClouds VM (confirmed 2026-10-07) default_weight: 1 # plugins not listed (push.http, push.mqtt) # A check with Whoopsy! (band alerting) on is billed as "+whoopsy" diff --git a/docs/features/F08-push-ingestion.md b/docs/features/F08-push-ingestion.md index a9835ed..96c3b75 100644 --- a/docs/features/F08-push-ingestion.md +++ b/docs/features/F08-push-ingestion.md @@ -91,6 +91,16 @@ Pushed timestamps are accepted from up to 24 h in the past (devices that buffer - **Billing.** Weight 2 per monitored VM (`vm.agent`, confirmed by PlusClouds on 2026-10-07, from `v0.12.1`); objects (disks, interfaces) are not billed. - **Not yet:** service states (the agent sends none in telemetry; a watch list in the agent is planned), heartbeats on `agent.vm..evt`. +## As built: box.agent (`v0.13.0`) + +- **Source.** `box.agent` (Go, on servers that run vLLM) POSTs a JSON heartbeat about once a second to `POST /ingest/v1/{check_id}` with `Authorization: Bearer mpush_…`. No NATS: it is an independent service. The token is minted for monitoring alone (not the token box.agent uses with llmocean.api). +- **Check.** `box.agent`, config `{"box_id": "", "missed_count": 5}`, unique on the server (409). The provisioner (llmocean.api) creates the device (external `{source: "greenference", type: "llmbox", id: }`) and the check with `interval_seconds: 1`, receives `push_token` once, and gives the box `-monitor-url`/`-monitor-token`. `POST /v1/checks/{id}/rotate-token` rotates it; disabling or deleting the check on deprovision stops the silence alert. The tenant's `min_check_interval_seconds` must be 1 or less. +- **Objects.** `host` (load, memory, disk, OOM kills, router connection, the box's own count of this server's answers by class), `vllm` (up, health, restarts, queue) and `gpu:` (temperature, utilisation, memory, ECC, last XID, throttle). A missing GPU array is unknown, not failed. +- **State.** XID 79, 48 or 64, or uncorrectable ECC, is CRITICAL at once. `booting`, `downloading`, `restarting` and `stopped` are never incidents. Otherwise the agent's `severity` (ok, degraded, failed) maps to OK, WARNING, CRITICAL; the server decides what happens next. Silence for `missed_count` intervals (5) is CRITICAL through the last-seen rule (swept every 5 s). +- **Counters.** `ingest_http_responses_total{code}` counts every answer of the push endpoint by status code. +- **Billing.** Weight 2 per box is a placeholder until PlusClouds confirms it. +- **Not yet:** the alert route to the llmocean.api webhook and the provisioning calls are llmocean.api's side. + ## Open questions - The MQTT broker, topic structure and payload format are now taken from `fixleanplus.metric.collector` ([F12](F12-fixlean-collector-replacement.md)). Open questions specific to that migration are listed there. diff --git a/docs/progress.md b/docs/progress.md index dbe8789..06fa7f8 100644 --- a/docs/progress.md +++ b/docs/progress.md @@ -167,6 +167,10 @@ Suppression: a pool host that goes down suppresses its VMs' incidents and the ch Verified by `TestPool`, `TestPoolFollowsMaster`, `TestPoolFailures` (fake pool), `TestDiscoveredDevices` (engine, devices, incidents, metrics, limits, retention) and `TestDiscoveredHostSuppression`. +### v0.13.0 — box.agent heartbeats + +`box.agent` checks receive an LLM box's 1 s heartbeat over HTTPS with their own push token (separate from the token box.agent uses with llmocean.api): host, vllm and GPU objects, hard GPU faults as immediate CRITICAL, boot states exempt, silence as CRITICAL after 5 beats. Push answers are counted by status code (`ingest_http_responses_total`). Verified by `TestBoxAgent` (not yet run in CI at the time of writing) and `FuzzDecode`. + ### v0.12.0 — PlusClouds VM agent telemetry `vm.agent` checks receive each PlusClouds VM's agent telemetry from the platform's NATS (`vm.*.telemetry`): host, disk and interface objects with thresholds, silence as CRITICAL. Opt-in per VM: the panel creates the device and check when a customer enables monitoring. Verified by `TestVMAgent`, `FuzzDecode` and a live read of the platform's NATS (14 VMs, every message parsed). Off until `MONITOR_NATS_ENABLED=true` with the NATS client certificate and the `monitoring` user (in the auth callout's `auth_users`). diff --git a/internal/api/api_integration_test.go b/internal/api/api_integration_test.go index efceec4..5f0f181 100644 --- a/internal/api/api_integration_test.go +++ b/internal/api/api_integration_test.go @@ -380,7 +380,7 @@ func TestPlugins(t *testing.T) { t.Errorf("%s: config_schema is not an object schema", p["type"]) } } - if strings.Join(types, ",") != "camera.snapshot,http,icmp,monitor.self,mqtt.connection,push.http,push.mqtt,redfish.health,rtsp.stream,snmp.get,snmp.interfaces,snmp.pdu,snmp.sensor,snmp.system,snmp.ups,vm.agent,xapi.pool" { + if strings.Join(types, ",") != "box.agent,camera.snapshot,http,icmp,monitor.self,mqtt.connection,push.http,push.mqtt,redfish.health,rtsp.stream,snmp.get,snmp.interfaces,snmp.pdu,snmp.sensor,snmp.system,snmp.ups,vm.agent,xapi.pool" { t.Errorf("plugins = %v, want the built-in plugins sorted by type", types) } if strings.Join(required, ",") != "redfish.health,snmp.get,snmp.interfaces,snmp.pdu,snmp.sensor,snmp.system,snmp.ups,xapi.pool" { diff --git a/internal/api/boxagent_integration_test.go b/internal/api/boxagent_integration_test.go new file mode 100644 index 0000000..c28b95b --- /dev/null +++ b/internal/api/boxagent_integration_test.go @@ -0,0 +1,262 @@ +package api_test + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + + "github.com/plusclouds/monitoring.server/internal/config" + "github.com/plusclouds/monitoring.server/internal/ingest" + "github.com/plusclouds/monitoring.server/internal/metrics" + "github.com/plusclouds/monitoring.server/internal/runner" +) + +func boxBeat(box, state, severity string, xid float64, extra map[string]any) string { + m := map[string]any{ + "v": 1, "box_id": box, "seq": 1, "ts": time.Now().Unix(), "agent_version": "1.4.0", + "state": state, "severity": severity, "backend_healthy": true, "router_connected": true, + "vllm": map[string]any{"up": true, "restart_count": 0}, + "queue": map[string]any{"running": 1, "waiting": 0}, + "host": map[string]any{"load": 1.0, "mem_pct": 40, "disk_pct": 50, "oom_kills": 0}, + "gpu": []any{map[string]any{"idx": 0, "temp": 70, "util": 90, "mem_used": 40e9, "mem_total": 80e9, + "ecc_dbe": 0, "xid_last": xid, "throttle": false}}, + } + for k, v := range extra { + m[k] = v + } + b, _ := json.Marshal(m) + return string(b) +} + +// box.agent (LLM box heartbeats over HTTPS with their own push token): the +// provisioner creates a device and a check per box and gets the token once; +// heartbeats become host, vllm and GPU objects; boot states are not +// incidents, a hard GPU fault and silence are; a rotated token stops the +// old one; every answer is counted by status code. +func TestBoxAgent(t *testing.T) { + ctx := context.Background() + x := newNoise(t, 0) + const box = "box-7f3a" + dev := x.id(x.must(x.do("POST", "/v1/devices", x.key, map[string]any{"name": box, "type": "server", + "external": map[string]any{"source": "greenference", "type": "llmbox", "id": box}}), 201)) + body := map[string]any{"name": "box", "plugin": "box.agent", "is_host_check": true, "interval_seconds": 1, + "config": map[string]any{"box_id": box}, + "thresholds": []any{map[string]any{"metric": "gpu_temp_c", "object": "*", "warning": map[string]any{"op": ">", "value": 85}}}} + + // The tenant's minimum interval applies: a 1 s check needs it lowered. + x.must(x.do("POST", "/v1/devices/"+dev+"/checks", x.key, body), 422) + if _, err := db.System.Exec(ctx, `UPDATE tenants SET min_check_interval_seconds = 1 WHERE id = $1`, x.tenant); err != nil { + t.Fatal(err) + } + created := x.must(x.do("POST", "/v1/devices/"+dev+"/checks", x.key, body), 201) + chk := x.id(created) + token, _ := created.body["push_token"].(string) + if !strings.HasPrefix(token, "mpush_") || created.body["failure_count"] != float64(1) { + t.Fatalf("created: %s", created.raw) + } + if got := x.must(x.do("GET", "/v1/checks/"+chk, x.key, nil), 200); got.body["push_token"] != nil { + t.Errorf("the token is shown again: %s", got.raw) + } + // One check per box on the whole server; a bad box_id is refused. + other := x.device("box-copy") + x.must(x.do("POST", "/v1/devices/"+other+"/checks", x.key, map[string]any{"name": "box", "plugin": "box.agent", + "interval_seconds": 1, "config": map[string]any{"box_id": strings.ToUpper(box)}}), 409) + x.must(x.do("POST", "/v1/devices/"+other+"/checks", x.key, map[string]any{"name": "box", "plugin": "box.agent", + "interval_seconds": 1, "config": map[string]any{"box_id": "bad id"}}), 422) + + results := make(chan runner.Result, 100) + reg := prometheus.NewRegistry() + in, err := ingest.New(ingest.Options{Config: config.IngestHTTP{MaxBody: 64 << 10, RateLimit: config.Rate{PerSecond: 1000, Burst: 1000}}, + System: db.System, Results: results, Logger: slog.New(slog.DiscardHandler), Registry: reg}) + if err != nil { + t.Fatal(err) + } + srv := httptest.NewServer(in.Handler()) + defer srv.Close() + w, err := metrics.NewWriter(metrics.WriterOptions{System: db.System}) + if err != nil { + t.Fatal(err) + } + post := func(id, tok, payload string) int { + t.Helper() + req, _ := http.NewRequestWithContext(ctx, "POST", srv.URL+"/ingest/v1/"+id, strings.NewReader(payload)) + if tok != "" { + req.Header.Set("Authorization", "Bearer "+tok) + } + res, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer func() { _ = res.Body.Close() }() + _, _ = io.Copy(io.Discard, res.Body) + return res.StatusCode + } + apply := func() int { + t.Helper() + n := 0 + for { + select { + case r := <-results: + if _, err := x.eng.Apply(ctx, r); err != nil { + t.Fatal(err) + } + w.Add(r) + n++ + default: + if err := w.Flush(ctx); err != nil { + t.Fatal(err) + } + return n + } + } + } + state := func() map[string]any { + return x.must(x.do("GET", "/v1/checks/"+chk+"/state", x.key, nil), 200).body + } + + // Auth and identity: no token, a wrong token, another box's heartbeat. + if code := post(chk, "", boxBeat(box, "ready", "ok", 0, nil)); code != 401 { + t.Errorf("no token: %d", code) + } + if code := post(chk, "mpush_nope", boxBeat(box, "ready", "ok", 0, nil)); code != 401 { + t.Errorf("wrong token: %d", code) + } + if code := post(chk, token, boxBeat("other-box", "ready", "ok", 0, nil)); code != 400 { + t.Errorf("another box's heartbeat: %d", code) + } + + // A healthy beat, with the box's own count of the server's answers. + if code := post(chk, token, boxBeat(box, "ready", "ok", 0, + map[string]any{"monitor_responses": map[string]any{"2xx": 10, "429": 2, "5xx": 1}})); code != 202 { + t.Fatalf("beat: %d", code) + } + apply() + if st := state(); st["phase"] != "OK" { + t.Fatalf("state after a beat: %v", st) + } + objs := x.must(x.do("GET", "/v1/checks/"+chk+"/objects", x.key, nil), 200).body["items"].([]any) + byKey := map[string]map[string]any{} + for _, o := range objs { + m := o.(map[string]any) + byKey[m["key"].(string)] = m + } + if len(byKey) != 3 || byKey["host"] == nil || byKey["vllm"] == nil || byKey["gpu:0"] == nil { + t.Fatalf("objects: %v", objs) + } + if m := byKey["host"]["last_metrics"].(map[string]any); m["monitor_responses_429"] != float64(2) || m["monitor_responses_2xx"] != float64(10) { + t.Errorf("response counts: %v", m) + } + if m := byKey["gpu:0"]["last_metrics"].(map[string]any); m["gpu_mem_usage_pct"] != float64(50) { + t.Errorf("gpu: %v", m) + } + + // Boot, download and restart are not incidents, even when the agent says failed. + for _, s := range []string{"booting", "downloading", "restarting", "stopped"} { + if code := post(chk, token, boxBeat(box, s, "failed", 0, nil)); code != 202 { + t.Fatalf("%s: %d", s, code) + } + apply() + if st := state(); st["phase"] != "OK" { + t.Errorf("%s is an incident: %v", s, st) + } + } + + // The GPU gets hot: the threshold opens a warning on that object only. + if code := post(chk, token, strings.Replace(boxBeat(box, "ready", "ok", 0, nil), `"temp":70`, `"temp":91`, 1)); code != 202 { + t.Fatal(code) + } + apply() + if inc := x.incident(chk); inc["object_key"] != "gpu:0" || inc["severity"] != "warning" { + t.Errorf("temperature threshold: %v", inc) + } + + // A hard GPU fault (XID 79) is CRITICAL at once, whatever the state. + if code := post(chk, token, boxBeat(box, "ready", "ok", 79, nil)); code != 202 { + t.Fatal(code) + } + apply() + if st := state(); st["phase"] != "PROBLEM" || !strings.Contains(st["last_output"].(string), "XID 79") { + t.Errorf("XID 79: %v", st) + } + // Recovery. + if code := post(chk, token, boxBeat(box, "ready", "ok", 0, nil)); code != 202 { + t.Fatal(code) + } + apply() + if st := state(); st["phase"] != "OK" { + t.Errorf("after recovery: %v", st) + } + + // Silence: 5 intervals without a beat is CRITICAL, once per interval. + if _, err := db.System.Exec(ctx, `UPDATE push_sources SET last_push_at = now() - interval '1 minute', + created_at = now() - interval '1 hour' WHERE check_id = $1`, chk); err != nil { + t.Fatal(err) + } + if n, err := in.Sweep(ctx); err != nil || n != 1 { + t.Fatalf("sweep: %d %v", n, err) + } + apply() + if st := state(); st["phase"] != "PROBLEM" || !strings.Contains(st["last_output"].(string), "5 intervals") { + t.Errorf("silent: %v", st) + } + if code := post(chk, token, boxBeat(box, "ready", "ok", 0, nil)); code != 202 { + t.Fatal(code) + } + apply() + if st := state(); st["phase"] != "OK" { + t.Errorf("after the box came back: %v", st) + } + + // Rotation: the old token stops at once. + rot := x.must(x.do("POST", "/v1/checks/"+chk+"/rotate-token", x.key, nil), 200) + fresh, _ := rot.body["push_token"].(string) + if fresh == "" || fresh == token { + t.Fatalf("rotate: %s", rot.raw) + } + if code := post(chk, token, boxBeat(box, "ready", "ok", 0, nil)); code != 401 { + t.Errorf("old token after rotation: %d", code) + } + if code := post(chk, fresh, boxBeat(box, "ready", "ok", 0, nil)); code != 202 { + t.Errorf("new token: %d", code) + } + apply() + + // A disabled check (deprovisioned box) refuses data. + x.must(x.do("PATCH", "/v1/checks/"+chk, x.key, map[string]any{"enabled": false}), 200) + if code := post(chk, fresh, boxBeat(box, "ready", "ok", 0, nil)); code != 409 { + t.Errorf("disabled: %d", code) + } + + // Every answer was counted by status code. + counted := map[string]float64{} + mfs, _ := reg.Gather() + for _, mf := range mfs { + if mf.GetName() != "ingest_http_responses_total" { + continue + } + for _, m := range mf.GetMetric() { + counted[labelValue(m, "code")] = m.GetCounter().GetValue() + } + } + if counted["401"] != 3 || counted["409"] != 1 || counted["400"] != 1 || counted["202"] < 10 { + t.Errorf("answers by code: %v", counted) + } +} + +func labelValue(m *dto.Metric, name string) string { + for _, l := range m.GetLabel() { + if l.GetName() == name { + return l.GetValue() + } + } + return "" +} diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 3711284..77c3200 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -101,7 +101,7 @@ func Default() Config { Enabled: true, Listen: ":9443", MaxBody: 64 << 10, - RateLimit: Rate{PerSecond: 1, Burst: 10}, + RateLimit: Rate{PerSecond: 2, Burst: 20}, // box.agent beats once a second }, MQTT: IngestMQTT{ Enabled: true, @@ -147,6 +147,7 @@ func Default() Config { "snmp.pdu": 2, "snmp.interfaces": 3, "redfish.health": 2, "xapi.pool": 0, "xapi.pool:host": 3, "camera.snapshot": 2, "rtsp.stream": 2, // each pool host bills, not the pool (2026-10-07) "monitor.self": 0, // the engine's own health check, platform tenant + "box.agent": 2, // each monitored LLM box (TBD with billing) "vm.agent": 2, // each monitored PlusClouds VM (confirmed 2026-10-07) "mqtt.connection": 0, // an MQTT sensor bills once, through its push.mqtt data check (confirmed 2026-10-07) }, DefaultWeight: 1, WhoopsyMultiplier: 5}, diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index a8ce61d..c8b2600 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -61,6 +61,7 @@ type Server struct { messages *prometheus.CounterVec stale prometheus.Counter + statuses *prometheus.CounterVec } type limiter struct { @@ -78,6 +79,9 @@ func New(o Options) (*Server, error) { messages: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "ingest_http_messages_total", Help: "Pushed HTTP messages by outcome.", }, []string{"outcome"}), + statuses: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "ingest_http_responses_total", Help: "Answers of the push endpoint by HTTP status code.", + }, []string{"code"}), stale: prometheus.NewCounter(prometheus.CounterOpts{ Name: "ingest_stale_results_total", Help: "Last-seen results sent for silent push checks.", }), @@ -86,7 +90,7 @@ func New(o Options) (*Server, error) { s.log = slog.New(slog.DiscardHandler) } if o.Registry != nil { - for _, c := range []prometheus.Collector{s.messages, s.stale} { + for _, c := range []prometheus.Collector{s.messages, s.stale, s.statuses} { if err := o.Registry.Register(c); err != nil { return nil, err } @@ -98,7 +102,7 @@ func New(o Options) (*Server, error) { // Handler serves the ingest endpoints. func (s *Server) Handler() http.Handler { mux := http.NewServeMux() - mux.HandleFunc("POST /ingest/v1/{check_id}", s.push) + mux.HandleFunc("POST /ingest/v1/{check_id}", s.counted(s.push)) mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { _, _ = io.WriteString(w, "ok\n") }) return mux } @@ -322,13 +326,41 @@ func (s *Server) fail(w http.ResponseWriter, status int, typ, title, detail stri }) } +// statusWriter remembers the status a handler wrote. +type statusWriter struct { + http.ResponseWriter + code int +} + +func (w *statusWriter) WriteHeader(code int) { + if w.code == 0 { + w.code = code + } + w.ResponseWriter.WriteHeader(code) +} + +// counted counts the handler's answers by status code. +func (s *Server) counted(h http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + sw := &statusWriter{ResponseWriter: w} + h(sw, r) + if sw.code == 0 { + sw.code = http.StatusOK + } + s.statuses.WithLabelValues(strconv.Itoa(sw.code)).Inc() + } +} + // missedCount is a push check's missed_count (push.http and push.mqtt // share the field), as the sweep query reads it. -func missedCount(cfg json.RawMessage) int { +func missedCount(kind string, cfg json.RawMessage) int { var c struct { MissedCount int `json:"missed_count"` } if json.Unmarshal(cfg, &c) != nil || c.MissedCount < 1 { + if kind == "box.agent" { + return 5 + } return 3 } return c.MissedCount @@ -344,7 +376,7 @@ func (s *Server) Sweep(ctx context.Context) (int, error) { FROM checks c, tenants t WHERE c.id = ps.check_id AND t.id = c.tenant_id AND c.enabled AND t.status = 'active' AND greatest(ps.last_push_at, ps.created_at) < - now() - make_interval(secs => c.interval_seconds * coalesce((c.config->>'missed_count')::int, 3)) + now() - make_interval(secs => c.interval_seconds * coalesce((c.config->>'missed_count')::int, CASE c.plugin WHEN 'box.agent' THEN 5 ELSE 3 END)) AND (ps.stale_reported_at IS NULL OR ps.stale_reported_at < now() - make_interval(secs => c.interval_seconds)) RETURNING ps.check_id, c.tenant_id, c.device_id, c.plugin, c.config, c.interval_seconds, ps.last_push_at`) if err != nil { @@ -371,7 +403,7 @@ func (s *Server) Sweep(ctx context.Context) (int, error) { if p, ok := plugin.Lookup(x.plugin); ok { layout = plugin.MetricsOf(p, x.config) } - missed := missedCount(x.config) + missed := missedCount(x.plugin, x.config) out := "no data received yet" if x.last != nil { out = "no data since " + x.last.UTC().Format(time.RFC3339) + " (" + strconv.Itoa(missed) + " intervals of " + diff --git a/internal/inventory/checks.go b/internal/inventory/checks.go index 7964a80..159bd35 100644 --- a/internal/inventory/checks.go +++ b/internal/inventory/checks.go @@ -170,6 +170,7 @@ var checkConstraints = map[string]string{ "checks_device_id_name_key": "the device already has a check with this name", "checks_one_host_check": "the device already has a host check", "checks_vm_agent_uuid": "another check already receives this VM's telemetry (vm_uuid)", + "checks_box_agent_id": "another check already receives this box's heartbeats (box_id)", } func scanCheck(row pgx.Row) (Check, error) { @@ -326,7 +327,7 @@ func createCheck(ctx context.Context, tx pgx.Tx, actor audit.Actor, device uuid. } var token string switch in.Plugin { - case "push.http": + case "push.http", "box.agent": // own token, minted for monitoring alone if token, err = setPushToken(ctx, tx, d.TenantID, id); err != nil { return Check{}, err } diff --git a/internal/inventory/push.go b/internal/inventory/push.go index de4bf71..2deb767 100644 --- a/internal/inventory/push.go +++ b/internal/inventory/push.go @@ -60,8 +60,8 @@ func RotatePushToken(ctx context.Context, tx pgx.Tx, actor audit.Actor, id uuid. if err != nil { return Check{}, err } - if c.Plugin != "push.http" { - return Check{}, errs.Conflictf("not-push", "%s has no ingest token; only push.http checks do", c.Plugin) + if c.Plugin != "push.http" && c.Plugin != "box.agent" { + return Check{}, errs.Conflictf("not-push", "%s has no ingest token; only push.http and box.agent checks do", c.Plugin) } token, err := setPushToken(ctx, tx, c.TenantID, id) if err != nil { diff --git a/migrations/00024_box_agent.sql b/migrations/00024_box_agent.sql new file mode 100644 index 0000000..887153e --- /dev/null +++ b/migrations/00024_box_agent.sql @@ -0,0 +1,8 @@ +-- v0.13.0: box.agent checks (LLM box heartbeats over HTTPS, own push token). A +-- box's ID names at most one check on the server: its heartbeats have one home. + +-- +goose Up +CREATE UNIQUE INDEX checks_box_agent_id ON checks ((lower(config->>'box_id'))) WHERE plugin = 'box.agent'; + +-- +goose Down +DROP INDEX checks_box_agent_id; diff --git a/plugins/all/all.go b/plugins/all/all.go index ac2c947..e9c530d 100644 --- a/plugins/all/all.go +++ b/plugins/all/all.go @@ -3,6 +3,7 @@ package all import ( + _ "github.com/plusclouds/monitoring.server/plugins/boxagent" _ "github.com/plusclouds/monitoring.server/plugins/camera" _ "github.com/plusclouds/monitoring.server/plugins/http" _ "github.com/plusclouds/monitoring.server/plugins/icmp" diff --git a/plugins/boxagent/boxagent.go b/plugins/boxagent/boxagent.go new file mode 100644 index 0000000..1b2a0fe --- /dev/null +++ b/plugins/boxagent/boxagent.go @@ -0,0 +1,414 @@ +// Package boxagent implements box.agent: the heartbeat the box.agent service +// (a WebSocket proxy in front of vLLM on an LLM host) POSTs to the ingest +// listener about once a second, with a token minted for monitoring alone +// (push token, F08). A box is one check whose objects are the host, the +// vLLM backend and each GPU, so each has its own thresholds, state and graphs. +// +// The agent reports facts and a severity; the server decides. Silence is +// CRITICAL through the last-seen rule (missed_count intervals), so a box that +// is booting, downloading a model or restarting vLLM keeps sending heartbeats +// with that state and is not an incident. +package boxagent + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "math" + "regexp" + "strings" + "time" + + "github.com/plusclouds/monitoring.server/pkg/plugin" +) + +func init() { plugin.Register(&Agent{}) } + +// Metric slots, shared by every object; an object fills its own. +const ( + Load1 = iota + MemUsage + DiskUsage + OOMKills + VLLMUp + BackendHealthy + VLLMRestarts + QueueRunning + QueueWaiting + RouterConnected + GPUTemp + GPUUtil + GPUMemUsed + GPUMemTotal + GPUMemUsage + GPUECCDbe + GPUXid + GPUThrottle + Resp2xx + Resp429 + Resp4xx + Resp5xx + RespErrors + NumMetrics +) + +var defs = []plugin.MetricDef{ + g("load1", "", "host: 1-minute load average"), + g("memory_usage_pct", "percent", "host: memory used"), + g("disk_usage_pct", "percent", "host: fullest disk"), + g("oom_kills", "count", "host: OOM kills since boot"), + g("vllm_up", "", "vllm: 1 process running, 0 not"), + g("backend_healthy", "", "vllm: 1 health probe passes, 0 fails"), + g("vllm_restarts", "count", "vllm: supervisor restarts since the agent started"), + g("queue_running", "count", "vllm: requests running"), + g("queue_waiting", "count", "vllm: requests waiting"), + g("router_connected", "", "host: 1 the router WebSocket is up, 0 down"), + g("gpu_temp_c", "celsius", "gpu: temperature"), + g("gpu_util_pct", "percent", "gpu: busy"), + g("gpu_mem_used_bytes", "bytes", "gpu: memory used"), + g("gpu_mem_total_bytes", "bytes", "gpu: memory installed"), + g("gpu_mem_usage_pct", "percent", "gpu: memory used"), + g("gpu_ecc_dbe", "count", "gpu: uncorrectable (double-bit) ECC errors"), + g("gpu_xid_last", "", "gpu: last XID error code (0 none)"), + g("gpu_throttle", "", "gpu: 1 throttled by heat or power, 0 not"), + g("monitor_responses_2xx", "count", "host: heartbeats this server accepted, since the agent started"), + g("monitor_responses_429", "count", "host: heartbeats refused by the rate limit (429)"), + g("monitor_responses_4xx", "count", "host: other 4xx answers (bad token, disabled check, invalid message)"), + g("monitor_responses_5xx", "count", "host: 5xx answers (server busy or failing)"), + g("monitor_send_errors", "count", "host: heartbeats that got no answer (network, timeout)"), +} + +func g(name, unit, desc string) plugin.MetricDef { + // The 1 s stream is for liveness and fast thresholds; the standard + // class keeps its storage and downsampling in line with other plugins. + return plugin.MetricDef{Name: name, Unit: unit, Description: desc, Kind: "gauge", RetentionClass: plugin.RetentionStandard} +} + +// Config of a box.agent check: which box's heartbeats it receives. +type Config struct { + BoxID string `json:"box_id" jsonschema:"required,maxLength=128,description=The box's ID (the agent instance ID llmocean.api uses). Unique on the server; every heartbeat must carry it"` + MissedCount int `json:"missed_count,omitempty" jsonschema:"minimum=1,maximum=100,default=5,description=CRITICAL after this many intervals without a heartbeat"` +} + +var defaults = Config{MissedCount: 5} + +var boxID = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`) + +// Agent implements box.agent. +type Agent struct{} + +func (*Agent) Manifest() plugin.Manifest { + return plugin.Manifest{ + Type: "box.agent", + Kind: plugin.KindIngester, + Description: "An LLM box's own heartbeat from box.agent (POST to the ingest URL with the check's token, about every second): " + + "host, vLLM backend and each GPU as their own objects. CRITICAL when no heartbeat arrives for missed_count intervals, " + + "or on a hard GPU fault (XID 79, uncorrectable ECC). Boot, model download and vLLM restarts are not incidents.", + ConfigSchema: plugin.SchemaFor[Config](), + DefaultInterval: time.Second, + MinInterval: time.Second, + BillingClass: plugin.BillingStandard, + Objects: true, + } +} + +func (*Agent) Validate(raw json.RawMessage) error { + cfg, err := plugin.DecodeConfig(raw, defaults) + if err != nil { + return err + } + if !boxID.MatchString(cfg.BoxID) { + return errors.New("box_id: letters, digits, . _ -, starting with a letter or digit, at most 128 characters") + } + if cfg.MissedCount < 1 || cfg.MissedCount > 100 { + return errors.New("missed_count must be between 1 and 100") + } + return nil +} + +func (*Agent) MetricsFor(json.RawMessage) ([]plugin.MetricDef, error) { return defs, nil } + +// Metrics is the layout. +func Metrics() []plugin.MetricDef { return defs } + +// MissedCount is the number of silent intervals before CRITICAL. +func MissedCount(raw json.RawMessage) int { + cfg, err := plugin.DecodeConfig(raw, defaults) + if err != nil || cfg.MissedCount < 1 { + return defaults.MissedCount + } + return cfg.MissedCount +} + +// Heartbeat is the agent's message. +type Heartbeat struct { + V int `json:"v"` + BoxID string `json:"box_id"` + Seq uint64 `json:"seq"` + Timestamp float64 `json:"ts"` // Unix seconds + AgentVersion string `json:"agent_version"` + State string `json:"state"` // downloading, booting, ready, unhealthy, restarting, stopped + Severity string `json:"severity"` // ok, degraded, failed + BackendHealthy *bool `json:"backend_healthy"` + RouterUp *bool `json:"router_connected"` + VLLM *struct { + Up *bool `json:"up"` + Restarts *float64 `json:"restart_count"` + Reason string `json:"last_restart_reason"` + } `json:"vllm"` + GPU []struct { + Idx *int `json:"idx"` + Temp *float64 `json:"temp"` + Util *float64 `json:"util"` + MemUsed *float64 `json:"mem_used"` + MemTotal *float64 `json:"mem_total"` + ECCDbe *float64 `json:"ecc_dbe"` + XID *float64 `json:"xid_last"` + Throttle *bool `json:"throttle"` + } `json:"gpu"` + // Responses are the agent's own count of this server's answers since it + // started; they reach the server on a later beat. + Responses *struct { + R2xx *float64 `json:"2xx"` + R429 *float64 `json:"429"` + R4xx *float64 `json:"4xx"` + R5xx *float64 `json:"5xx"` + Errors *float64 `json:"errors"` + } `json:"monitor_responses"` + Queue *struct { + Running *float64 `json:"running"` + Waiting *float64 `json:"waiting"` + } `json:"queue"` + Host *struct { + Load *float64 `json:"load"` + MemPct *float64 `json:"mem_pct"` + DiskPct *float64 `json:"disk_pct"` + OOM *float64 `json:"oom_kills"` + } `json:"host"` +} + +// Message is one decoded heartbeat. +type Message struct { + Box string // box_id + Time time.Time + HB Heartbeat +} + +// Limits against a broken or hostile agent. +const maxGPUs = 64 + +// Decode reads a heartbeat. +func Decode(body []byte) (Message, error) { + var hb Heartbeat + if err := json.NewDecoder(bytes.NewReader(body)).Decode(&hb); err != nil { + return Message{}, fmt.Errorf("not a box.agent heartbeat: %w", err) + } + if !boxID.MatchString(hb.BoxID) { + return Message{}, errors.New("box_id is missing or malformed") + } + switch hb.Severity { + case "", "ok", "degraded", "failed": + default: + return Message{}, fmt.Errorf("severity %q is not ok, degraded or failed", hb.Severity) + } + if len(hb.GPU) > maxGPUs { + return Message{}, fmt.Errorf("at most %d GPUs", maxGPUs) + } + m := Message{Box: hb.BoxID, HB: hb} + if ts := hb.Timestamp; ts > 0 && !math.IsInf(ts, 0) && ts < 1e11 { + sec, frac := math.Modf(ts) + m.Time = time.Unix(int64(sec), int64(frac*1e9)).UTC() + } + return m, nil +} + +// transient states are expected to look unhealthy: a box in one of them is +// not an incident while the heartbeats keep coming. "stopped" is the last +// beat of a graceful shutdown; the silence after it is still CRITICAL unless +// the check is disabled or deleted. +func transient(state string) bool { + switch state { + case "downloading", "booting", "restarting", "stopped": + return true + } + return false +} + +// hardXID are the XID codes that mean the GPU is gone or its memory is +// broken: 79 fell off the bus, 48 double-bit ECC, 64 ECC row-remap failure. +func hardXID(x float64) bool { return x == 79 || x == 48 || x == 64 } + +// HardFault names the first GPU fault that justifies replacing the box at +// once, or "". +func HardFault(m Message) string { + for i, gpu := range m.HB.GPU { + idx := i + if gpu.Idx != nil { + idx = *gpu.Idx + } + if gpu.XID != nil && hardXID(*gpu.XID) { + return fmt.Sprintf("GPU %d XID %d", idx, int(*gpu.XID)) + } + if gpu.ECCDbe != nil && *gpu.ECCDbe > 0 { + return fmt.Sprintf("GPU %d uncorrectable ECC errors (%d)", idx, int(*gpu.ECCDbe)) + } + } + return "" +} + +// Objects turns a heartbeat into objects: host, vllm and one per GPU. +func Objects(m Message) []plugin.Object { + hb := m.HB + host := plugin.Object{Key: "host", Name: "host", Labels: map[string]string{"kind": "host"}, Metrics: plugin.NaNs(NumMetrics)} + if h := hb.Host; h != nil { + set(host.Metrics, Load1, h.Load) + set(host.Metrics, MemUsage, h.MemPct) + set(host.Metrics, DiskUsage, h.DiskPct) + set(host.Metrics, OOMKills, h.OOM) + } + setBool(host.Metrics, RouterConnected, hb.RouterUp) + if r := hb.Responses; r != nil { + set(host.Metrics, Resp2xx, r.R2xx) + set(host.Metrics, Resp429, r.R429) + set(host.Metrics, Resp4xx, r.R4xx) + set(host.Metrics, Resp5xx, r.R5xx) + set(host.Metrics, RespErrors, r.Errors) + } + + vllm := plugin.Object{Key: "vllm", Name: "vllm", Labels: map[string]string{"kind": "backend", "state": hb.State}, + Metrics: plugin.NaNs(NumMetrics)} + setBool(vllm.Metrics, BackendHealthy, hb.BackendHealthy) + if v := hb.VLLM; v != nil { + setBool(vllm.Metrics, VLLMUp, v.Up) + set(vllm.Metrics, VLLMRestarts, v.Restarts) + if v.Reason != "" { + vllm.Labels["last_restart_reason"] = truncate(v.Reason) + } + } + if q := hb.Queue; q != nil { + set(vllm.Metrics, QueueRunning, q.Running) + set(vllm.Metrics, QueueWaiting, q.Waiting) + } + if !transient(hb.State) { + switch { + case hb.VLLM != nil && hb.VLLM.Up != nil && !*hb.VLLM.Up: + vllm.Status, vllm.Output = plugin.Critical, "vLLM is not running" + case hb.BackendHealthy != nil && !*hb.BackendHealthy: + vllm.Status, vllm.Output = plugin.Critical, "vLLM health probe fails" + } + } + + out := []plugin.Object{host, vllm} + for i, gpu := range hb.GPU { + idx := i + if gpu.Idx != nil && *gpu.Idx >= 0 && *gpu.Idx < 1<<16 { + idx = *gpu.Idx + } + o := plugin.Object{Key: fmt.Sprintf("gpu:%d", idx), Name: fmt.Sprintf("GPU %d", idx), + Labels: map[string]string{"kind": "gpu"}, Metrics: plugin.NaNs(NumMetrics)} + set(o.Metrics, GPUTemp, gpu.Temp) + set(o.Metrics, GPUUtil, gpu.Util) + set(o.Metrics, GPUMemUsed, gpu.MemUsed) + set(o.Metrics, GPUMemTotal, gpu.MemTotal) + set(o.Metrics, GPUECCDbe, gpu.ECCDbe) + set(o.Metrics, GPUXid, gpu.XID) + setBool(o.Metrics, GPUThrottle, gpu.Throttle) + if gpu.MemUsed != nil && gpu.MemTotal != nil && finite(*gpu.MemUsed) && finite(*gpu.MemTotal) && *gpu.MemTotal > 0 { + o.Metrics[GPUMemUsage] = *gpu.MemUsed / *gpu.MemTotal * 100 + } + if gpu.XID != nil && hardXID(*gpu.XID) { + o.Status, o.Output = plugin.Critical, fmt.Sprintf("XID %d", int(*gpu.XID)) + } else if gpu.ECCDbe != nil && *gpu.ECCDbe > 0 { + o.Status, o.Output = plugin.Critical, "uncorrectable ECC errors" + } + out = append(out, o) + } + return out +} + +// Status is the check's state for a heartbeat: a hard GPU fault is CRITICAL +// whatever the state; transient states are OK; otherwise the agent's +// severity, with a lost router connection on a ready box a WARNING. +func Status(m Message) (plugin.Status, string) { + if f := HardFault(m); f != "" { + return plugin.Critical, f + } + hb := m.HB + if transient(hb.State) { + return plugin.OK, "box is " + hb.State + } + switch hb.Severity { + case "failed": + return plugin.Critical, "agent reports failed (state " + hb.State + ")" + case "degraded": + return plugin.Warning, "agent reports degraded (state " + hb.State + ")" + } + if hb.State == "unhealthy" { + return plugin.Critical, "box is unhealthy" + } + if hb.RouterUp != nil && !*hb.RouterUp { + return plugin.Warning, "router connection is down" + } + return plugin.OK, "" +} + +// Summary is the check's output for a message. +func Summary(m Message, why string) string { + parts := []string{fmt.Sprintf("state %s", orDash(m.HB.State))} + if why != "" { + parts = append(parts, why) + } + parts = append(parts, fmt.Sprintf("%d GPUs", len(m.HB.GPU))) + return strings.Join(parts, ", ") +} + +// Parse makes box.agent an Ingester. The heartbeat's box_id must be the +// check's, so one box's token cannot report for another. +func (*Agent) Parse(raw json.RawMessage, body []byte) ([]plugin.Result, error) { + cfg, err := plugin.DecodeConfig(raw, defaults) + if err != nil { + return nil, err + } + m, err := Decode(body) + if err != nil { + return nil, err + } + if !strings.EqualFold(m.Box, cfg.BoxID) { + return nil, errors.New("box_id is not this check's box") + } + status, why := Status(m) + return []plugin.Result{{Status: status, Time: m.Time, Objects: Objects(m), Output: Summary(m, why)}}, nil +} + +func orDash(s string) string { + if s == "" { + return "-" + } + return s +} + +func set(dst []float64, slot int, v *float64) { + if v != nil && finite(*v) { + dst[slot] = *v + } +} + +func setBool(dst []float64, slot int, v *bool) { + if v == nil { + return + } + dst[slot] = 0 + if *v { + dst[slot] = 1 + } +} + +func finite(v float64) bool { return !math.IsNaN(v) && !math.IsInf(v, 0) } + +func truncate(s string) string { + if len(s) > 200 { + return s[:200] + } + return s +} diff --git a/plugins/boxagent/boxagent_test.go b/plugins/boxagent/boxagent_test.go new file mode 100644 index 0000000..cde6c7d --- /dev/null +++ b/plugins/boxagent/boxagent_test.go @@ -0,0 +1,163 @@ +package boxagent + +import ( + "encoding/json" + "math" + "testing" + "time" + + "github.com/plusclouds/monitoring.server/pkg/plugin" +) + +const box = "box-7f3a" + +// Sample is a heartbeat as box.agent encodes it. +func Sample(id, state, severity string, xid float64) []byte { + b, _ := json.Marshal(map[string]any{ + "v": 1, "box_id": id, "seq": 42, "ts": 1790000000.5, "agent_version": "1.4.0", + "state": state, "severity": severity, "backend_healthy": true, "router_connected": true, + "vllm": map[string]any{"up": true, "restart_count": 1, "last_restart_reason": "watchdog"}, + "queue": map[string]any{"running": 3, "waiting": 7}, + "host": map[string]any{"load": 1.5, "mem_pct": 40.2, "disk_pct": 61, "oom_kills": 0}, + "gpu": []any{ + map[string]any{"idx": 0, "temp": 71, "util": 98, "mem_used": 40e9, "mem_total": 80e9, "ecc_dbe": 0, "xid_last": 0, "throttle": false}, + map[string]any{"idx": 1, "temp": 69, "util": 97, "mem_used": 41e9, "mem_total": 80e9, "ecc_dbe": 0, "xid_last": xid, "throttle": true}, + }, + }) + return b +} + +func cfg() json.RawMessage { return json.RawMessage(`{"box_id":"` + box + `"}`) } + +func TestParseOK(t *testing.T) { + res, err := (&Agent{}).Parse(cfg(), Sample(box, "ready", "ok", 0)) + if err != nil || len(res) != 1 { + t.Fatalf("parse: %v %v", res, err) + } + r := res[0] + if r.Status != plugin.OK || !r.Time.Equal(time.Unix(1790000000, 5e8)) { + t.Errorf("result: %+v", r) + } + if len(r.Objects) != 4 || r.Objects[0].Key != "host" || r.Objects[1].Key != "vllm" || r.Objects[3].Key != "gpu:1" { + t.Fatalf("objects: %+v", r.Objects) + } + if h := r.Objects[0].Metrics; h[Load1] != 1.5 || h[MemUsage] != 40.2 || h[RouterConnected] != 1 || !math.IsNaN(h[GPUTemp]) { + t.Errorf("host: %v", h) + } + if v := r.Objects[1].Metrics; v[VLLMUp] != 1 || v[QueueWaiting] != 7 || v[VLLMRestarts] != 1 { + t.Errorf("vllm: %v", v) + } + if g := r.Objects[2].Metrics; g[GPUTemp] != 71 || g[GPUMemUsage] != 50 || g[GPUThrottle] != 0 { + t.Errorf("gpu0: %v", g) + } + if len(r.Objects[0].Metrics) != len(Metrics()) { + t.Error("object metrics do not match the layout") + } +} + +func TestStatus(t *testing.T) { + for _, c := range []struct { + name string + state, severity string + xid float64 + want plugin.Status + }{ + {"ready", "ready", "ok", 0, plugin.OK}, + {"degraded", "ready", "degraded", 0, plugin.Warning}, + {"failed", "ready", "failed", 0, plugin.Critical}, + {"unhealthy", "unhealthy", "", 0, plugin.Critical}, + {"booting is exempt", "booting", "failed", 0, plugin.OK}, + {"downloading is exempt", "downloading", "failed", 0, plugin.OK}, + {"restarting is exempt", "restarting", "failed", 0, plugin.OK}, + {"stopped is exempt", "stopped", "failed", 0, plugin.OK}, + {"xid 79 is immediate", "booting", "ok", 79, plugin.Critical}, + {"xid 31 is not a hard fault", "ready", "ok", 31, plugin.OK}, + } { + res, err := (&Agent{}).Parse(cfg(), Sample(box, c.state, c.severity, c.xid)) + if err != nil || res[0].Status != c.want { + t.Errorf("%s: %v %v", c.name, res, err) + } + } + res, _ := (&Agent{}).Parse(cfg(), Sample(box, "ready", "ok", 79)) + if g := res[0].Objects[3]; g.Status != plugin.Critical || g.Output != "XID 79" { + t.Errorf("gpu object: %+v", g) + } +} + +func TestMonitorResponses(t *testing.T) { + var m map[string]any + _ = json.Unmarshal(Sample(box, "ready", "ok", 0), &m) + m["monitor_responses"] = map[string]any{"2xx": 100, "429": 3, "5xx": 2, "errors": 1} + b, _ := json.Marshal(m) + res, err := (&Agent{}).Parse(cfg(), b) + if err != nil { + t.Fatal(err) + } + h := res[0].Objects[0].Metrics + if h[Resp2xx] != 100 || h[Resp429] != 3 || h[Resp5xx] != 2 || h[RespErrors] != 1 || !math.IsNaN(h[Resp4xx]) { + t.Errorf("responses: %v", h) + } +} + +func TestRouterDown(t *testing.T) { + var m map[string]any + _ = json.Unmarshal(Sample(box, "ready", "ok", 0), &m) + m["router_connected"] = false + b, _ := json.Marshal(m) + res, err := (&Agent{}).Parse(cfg(), b) + if err != nil || res[0].Status != plugin.Warning { + t.Errorf("router down: %v %v", res, err) + } +} + +func TestParseErrors(t *testing.T) { + a := &Agent{} + if _, err := a.Parse(cfg(), Sample("other-box", "ready", "ok", 0)); err == nil { + t.Error("a token must not report for another box") + } + for _, body := range []string{``, `{`, `[]`, `{"box_id":""}`, `{"box_id":"a b"}`, `{"box_id":"x","severity":"bad"}`} { + if _, err := a.Parse(json.RawMessage(`{"box_id":"x"}`), []byte(body)); err == nil { + t.Errorf("accepted %q", body) + } + } +} + +func TestValidate(t *testing.T) { + a := &Agent{} + for raw, ok := range map[string]bool{ + `{"box_id":"box-1"}`: true, + `{"box_id":"box-1","missed_count":10}`: true, + `{}`: false, + `{"box_id":"bad id"}`: false, + `{"box_id":"box-1","missed_count":0}`: false, + `{"box_id":"box-1","missed_count":101}`: false, + `{"box_id":"box-1","extra":1}`: false, + } { + if err := a.Validate(json.RawMessage(raw)); (err == nil) != ok { + t.Errorf("%s: %v", raw, err) + } + } +} + +func TestMissedCount(t *testing.T) { + if MissedCount(cfg()) != 5 || MissedCount(json.RawMessage(`{"box_id":"x","missed_count":8}`)) != 8 { + t.Error("missed_count") + } +} + +func FuzzDecode(f *testing.F) { + f.Add(Sample(box, "ready", "ok", 0)) + f.Add([]byte(`{"box_id":"x","gpu":[{}]}`)) + f.Fuzz(func(t *testing.T, body []byte) { + m, err := Decode(body) + if err != nil { + return + } + for _, o := range Objects(m) { + if len(o.Metrics) != NumMetrics { + t.Fatalf("object %s has %d metrics", o.Key, len(o.Metrics)) + } + } + Status(m) + }) +}