feat: box.agent heartbeat plugin with its own push token (v0.13.0)

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 ./...

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 "<plugin>+whoopsy"

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.<uuid>.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": "<id>", "missed_count": 5}`, unique on the server (409). The provisioner (llmocean.api) creates the device (external `{source: "greenference", type: "llmbox", id: <box 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:<idx>` (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.

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`).

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" {

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 ""
+}

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},