Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions deploy/config.example.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -271,9 +271,9 @@ ingest:
key_file: /etc/monitor/tls/ingest.key
trusted_proxies: []
max_body: 64KiB
rate_limit: # per ingest token (box.agent beats once a second)
per_second: 2
burst: 20
rate_limit: # per ingest token
per_second: 1
burst: 10

mqtt:
# Embedded broker. External-broker mode is configured per tenant through
Expand Down
8 changes: 4 additions & 4 deletions docs/features/F08-push-ingestion.md
Original file line number Diff line number Diff line change
Expand Up @@ -91,12 +91,12 @@ 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`)
## As built: box.agent (`v0.13.0`, 5 s from `v0.13.1`)

- **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.
- **Source.** `box.agent` (Go, on servers that run vLLM) POSTs a JSON heartbeat every 5 seconds 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": 3}`, 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: 5`, 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 plugin's own 5 s floor holds in every tenant (`Manifest.TenantFloorExempt`): the tenant's `min_check_interval_seconds` does not apply to `box.agent`, so no tenant setting is needed.
- **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).
- **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 (3 x 5 s = 15 s) 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.
Expand Down
6 changes: 5 additions & 1 deletion docs/progress.md
Original file line number Diff line number Diff line change
Expand Up @@ -167,9 +167,13 @@ 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.1 — box.agent every 5 s in every tenant

`box.agent` beats every 5 s (was planned at 1 s): `DefaultInterval` and `MinInterval` are 5 s, `missed_count` defaults to 3 (silence after 15 s), and `Manifest.TenantFloorExempt` makes the 5 s floor hold in every tenant, so the tenant's 30 s or 60 s `min_check_interval_seconds` no longer blocks box provisioning. The ingest rate limit stays at 1/s. Verified by `TestBoxAgent`.

### 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`.
`box.agent` checks receive an LLM box's heartbeat (1 s planned, 5 s from v0.13.1) 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 missed 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

Expand Down
20 changes: 11 additions & 9 deletions internal/api/boxagent_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ func boxBeat(box, state, severity string, xid float64, extra map[string]any) str

// 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
// heartbeats (every 5 s) 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) {
Expand All @@ -48,15 +48,17 @@ func TestBoxAgent(t *testing.T) {
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,
body := map[string]any{"name": "box", "plugin": "box.agent", "is_host_check": true, "interval_seconds": 5,
"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 {
// The plugin's own 5 s floor holds in every tenant; the tenant's minimum
// (30 s here) does not apply to box.agent.
if _, err := db.System.Exec(ctx, `UPDATE tenants SET min_check_interval_seconds = 30 WHERE id = $1`, x.tenant); err != nil {
t.Fatal(err)
}
faster := map[string]any{"name": "box", "plugin": "box.agent", "interval_seconds": 1, "config": map[string]any{"box_id": box}}
x.must(x.do("POST", "/v1/devices/"+dev+"/checks", x.key, faster), 422)
created := x.must(x.do("POST", "/v1/devices/"+dev+"/checks", x.key, body), 201)
chk := x.id(created)
token, _ := created.body["push_token"].(string)
Expand All @@ -69,9 +71,9 @@ func TestBoxAgent(t *testing.T) {
// 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)
"interval_seconds": 5, "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)
"interval_seconds": 5, "config": map[string]any{"box_id": "bad id"}}), 422)

results := make(chan runner.Result, 100)
reg := prometheus.NewRegistry()
Expand Down Expand Up @@ -196,7 +198,7 @@ func TestBoxAgent(t *testing.T) {
t.Errorf("after recovery: %v", st)
}

// Silence: 5 intervals without a beat is CRITICAL, once per interval.
// Silence: 3 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)
Expand All @@ -205,7 +207,7 @@ func TestBoxAgent(t *testing.T) {
t.Fatalf("sweep: %d %v", n, err)
}
apply()
if st := state(); st["phase"] != "PROBLEM" || !strings.Contains(st["last_output"].(string), "5 intervals") {
if st := state(); st["phase"] != "PROBLEM" || !strings.Contains(st["last_output"].(string), "3 intervals") {
t.Errorf("silent: %v", st)
}
if code := post(chk, token, boxBeat(box, "ready", "ok", 0, nil)); code != 202 {
Expand Down
2 changes: 1 addition & 1 deletion internal/config/defaults.go
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,7 @@ func Default() Config {
Enabled: true,
Listen: ":9443",
MaxBody: 64 << 10,
RateLimit: Rate{PerSecond: 2, Burst: 20}, // box.agent beats once a second
RateLimit: Rate{PerSecond: 1, Burst: 10},
},
MQTT: IngestMQTT{
Enabled: true,
Expand Down
9 changes: 3 additions & 6 deletions internal/ingest/ingest.go
Original file line number Diff line number Diff line change
Expand Up @@ -353,14 +353,11 @@ func (s *Server) counted(h http.HandlerFunc) http.HandlerFunc {

// missedCount is a push check's missed_count (push.http and push.mqtt
// share the field), as the sweep query reads it.
func missedCount(kind string, cfg json.RawMessage) int {
func missedCount(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
Expand All @@ -376,7 +373,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, CASE c.plugin WHEN 'box.agent' THEN 5 ELSE 3 END))
now() - make_interval(secs => c.interval_seconds * coalesce((c.config->>'missed_count')::int, 3))
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 {
Expand All @@ -403,7 +400,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.plugin, x.config)
missed := missedCount(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 " +
Expand Down
13 changes: 11 additions & 2 deletions internal/inventory/checks.go
Original file line number Diff line number Diff line change
Expand Up @@ -93,9 +93,9 @@ func prepare(ctx context.Context, tx pgx.Tx, in CheckInput, lim Limits) (CheckIn
return in, errs.Invalidf("config", "%s", err.Error())
}
if in.IntervalSeconds == 0 {
in.IntervalSeconds = max(int(m.DefaultInterval/time.Second), lim.MinCheckIntervalSeconds)
in.IntervalSeconds = max(int(m.DefaultInterval/time.Second), tenantFloor(m, lim))
}
minInterval := max(int(m.MinInterval/time.Second), lim.MinCheckIntervalSeconds)
minInterval := max(int(m.MinInterval/time.Second), tenantFloor(m, lim))
if in.IntervalSeconds < minInterval {
return in, errs.Invalidf("interval_seconds", "must be at least %d for %s in this tenant", minInterval, m.Type)
}
Expand Down Expand Up @@ -166,6 +166,15 @@ const checkCols = `c.id, c.tenant_id, c.device_id, c.name, c.plugin, c.config, c
(SELECT jsonb_build_object('token_prefix', ps.token_prefix, 'token_created_at', ps.token_created_at,
'last_push_at', ps.last_push_at) FROM push_sources ps WHERE ps.check_id = c.id)`

// tenantFloor is the tenant's minimum check interval that applies to a plugin:
// none for a plugin that is exempt (its own MinInterval is the floor).
func tenantFloor(m plugin.Manifest, lim Limits) int {
if m.TenantFloorExempt {
return 0
}
return lim.MinCheckIntervalSeconds
}

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",
Expand Down
15 changes: 10 additions & 5 deletions pkg/plugin/plugin.go
Original file line number Diff line number Diff line change
Expand Up @@ -87,11 +87,16 @@ type Manifest struct {
Metrics []MetricDef `json:"metrics"` // fixed layout; Result.Metrics aligns with it
DefaultInterval time.Duration `json:"-"`
MinInterval time.Duration `json:"-"`
NeedsRawSocket bool `json:"needs_raw_socket"`
MaxConcurrency int `json:"-"` // per runner; 0 = runner default
PerTargetLimit int `json:"-"` // concurrent runs against one address; 0 = runner default
Slow bool `json:"-"` // runs in the slow pool (F04)
BillingClass string `json:"billing_class"`
// TenantFloorExempt: the plugin's own MinInterval is the floor in every
// tenant, not the tenant's min_check_interval_seconds. For ingesters whose
// agents beat at a fixed short rate (box.agent) and whose interval only
// sets the silence window.
TenantFloorExempt bool `json:"-"`
NeedsRawSocket bool `json:"needs_raw_socket"`
MaxConcurrency int `json:"-"` // per runner; 0 = runner default
PerTargetLimit int `json:"-"` // concurrent runs against one address; 0 = runner default
Slow bool `json:"-"` // runs in the slow pool (F04)
BillingClass string `json:"billing_class"`
// WhoopsyMetric is the metric Whoopsy! (band alerting) watches unless
// the check names another: response time for http, round-trip time
// for icmp. Empty: the check must name one.
Expand Down
23 changes: 12 additions & 11 deletions plugins/boxagent/boxagent.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
// 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
// listener every 5 seconds, 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.
//
Expand Down Expand Up @@ -80,18 +80,18 @@ var defs = []plugin.MetricDef{
}

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.
// A 5 s stream for liveness and 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"`
MissedCount int `json:"missed_count,omitempty" jsonschema:"minimum=1,maximum=100,default=3,description=CRITICAL after this many intervals without a heartbeat"`
}

var defaults = Config{MissedCount: 5}
var defaults = Config{MissedCount: 3}

var boxID = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`)

Expand All @@ -102,14 +102,15 @@ 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): " +
Description: "An LLM box's own heartbeat from box.agent (POST to the ingest URL with the check's token, every 5 seconds): " +
"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,
ConfigSchema: plugin.SchemaFor[Config](),
DefaultInterval: 5 * time.Second,
MinInterval: 5 * time.Second,
TenantFloorExempt: true, // 5 s in every tenant: the agent beats at this rate
BillingClass: plugin.BillingStandard,
Objects: true,
}
}

Expand Down
2 changes: 1 addition & 1 deletion plugins/boxagent/boxagent_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,7 @@ func TestValidate(t *testing.T) {
}

func TestMissedCount(t *testing.T) {
if MissedCount(cfg()) != 5 || MissedCount(json.RawMessage(`{"box_id":"x","missed_count":8}`)) != 8 {
if MissedCount(cfg()) != 3 || MissedCount(json.RawMessage(`{"box_id":"x","missed_count":8}`)) != 8 {
t.Error("missed_count")
}
}
Expand Down
Loading