From 8c29a6936658705294372d0e6c229ea060e6f9ba Mon Sep 17 00:00:00 2001 From: Darkroom4364 Date: Sat, 25 Jul 2026 22:15:40 +0200 Subject: [PATCH] feat(server): add operator health, readiness, and telemetry endpoints Machine-readable liveness (/api/health), readiness (/api/ready), and runtime telemetry (/api/telemetry) on the operator mux behind the existing operator guard. Per-listener status, sanitized last-error class, active agent count, queue depth, and recent task/build failure indicators with documented healthy/degraded/failing semantics. Refs #102 --- docs/architecture.md | 3 + docs/health-telemetry.md | 155 +++++ server/cmd/server.go | 4 + server/internal/behaviour/http_polling.go | 47 ++ .../internal/handlers/api/health_handlers.go | 357 +++++++++++ .../handlers/api/health_handlers_test.go | 574 ++++++++++++++++++ server/internal/tasks/durable_store.go | 27 + server/internal/tasks/store.go | 34 ++ 8 files changed, 1201 insertions(+) create mode 100644 docs/health-telemetry.md create mode 100644 server/internal/handlers/api/health_handlers.go create mode 100644 server/internal/handlers/api/health_handlers_test.go diff --git a/docs/architecture.md b/docs/architecture.md index 7457d93..df854b6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -124,6 +124,9 @@ are visible during development. | `/api/agents/{id}/command` | `internal/handlers/api` | Deprecated adapter from a raw command to a v1 shell task. | | `/api/agents/{id}/results` | `internal/handlers/api` | Deprecated bare-array result adapter with strict `limit`/`offset` pagination and a 16 MiB encoded-page ceiling; response headers expose count, total, and continuation offset. | | `/api/audit/events` | `internal/handlers/api` | Retrieve a bounded, newest-first page of Audit Event v1 records. | +| `/api/health` | `internal/handlers/api` | Liveness probe; see [health-telemetry.md](health-telemetry.md). | +| `/api/ready` | `internal/handlers/api` | Readiness probe with per-dependency checks. | +| `/api/telemetry` | `internal/handlers/api` | Listener runtime status, sanitized error classes, agent counts, queue depth, and recent task/build failure indicators. | | `/api/file_drop/upload` | `internal/handlers/api` | Upload operator files into the server file store with a 32 MiB total request cap and same-directory atomic staging. | | `/api/file_drop/list` | `internal/handlers/api` | List operator file-drop contents. | | `/api/file_drop/download/{name}` | `internal/handlers/api` | Download a verified regular file with causal request/completion audit events. | diff --git a/docs/health-telemetry.md b/docs/health-telemetry.md new file mode 100644 index 0000000..1f0ae62 --- /dev/null +++ b/docs/health-telemetry.md @@ -0,0 +1,155 @@ +# Health, Readiness, and Runtime Telemetry + +MicroC2 exposes a small machine-readable health surface on the operator +server so deployments and CI can detect degraded state without tailing logs. +All three endpoints are GET-only JSON routes on the operator mux, so they sit +behind the same operator guard as every other operator route: loopback +clients always pass, and non-loopback clients must present the configured +`X-Operator-Token` header. Local deployment checks and CI smoke tests run +over loopback and need no token. + +The endpoints are deterministic and inexpensive by contract. They never spawn +builds, never make network calls, and never scan task payloads or agent +history; readiness is a single `SELECT 1`, and telemetry reads runtime +counters plus two constant-time `COUNT` aggregates per listener. + +## Endpoints + +### `GET /api/health` — liveness + +Answers whenever the process can serve HTTP. Performs no I/O beyond encoding +the response. + +```json +{"status": "ok", "uptime_seconds": 42} +``` + +Always returns `200` with `"status": "ok"`. If the process is wedged, the +request itself fails — that is the liveness signal. + +### `GET /api/ready` — readiness + +Reports whether the server's configured dependencies answer a trivial check. +Today the only check is durable storage. + +```json +{"status": "ready", "checks": {"storage": "ok"}} +``` + +- `200` with `"status": "ready"` when every check passes. +- `503` with `"status": "not_ready"` when storage is `"unavailable"`. +- Storage is `"disabled"` when the server runs without a configured database; + that is a valid, ready state for non-durable lab runs. + +### `GET /api/telemetry` — runtime summary + +Reports operator-server health plus per-listener runtime telemetry. + +```json +{ + "status": "degraded", + "generated_at": "2026-07-25T12:00:00Z", + "uptime_seconds": 42, + "storage": "ok", + "listeners": [ + { + "id": "…", + "name": "…", + "protocol": "https", + "port": 8443, + "status": "ACTIVE", + "health": "degraded", + "active_agents": 1, + "queue_depth": 2, + "recent_task_failures": 1, + "recent_build_failures": 0 + } + ], + "builds": {"recent_failures": 1} +} +``` + +Always returns `200`; the `status` field carries the verdict. Per listener: + +- `status` is the listener lifecycle state (`ACTIVE`, `STOPPED`, `ERROR`). +- `active_agents` counts agents that completed a heartbeat during this server + process lifetime. Durable agent history loaded at startup does not count + until the agent heartbeats again. +- `queue_depth` counts tasks `queued` or `dispatched` but not yet completed. +- `recent_task_failures` counts tasks that reached the failed terminal state + inside the failure window (one hour, trailing). +- `recent_build_failures` counts payload builds in `failed` or `interrupted` + state created inside the failure window. Completed builds whose artifacts + later fail reconciliation (`missing`/`corrupt`) are not build failures. +- `last_error_class` appears only when the listener recorded an error and is + one of `bind_failed`, `tls_config_invalid`, `runtime_error`, or + `telemetry_unavailable`. Raw error strings are never exposed because they + can embed certificate paths, bind addresses, and TLS library internals. + +Responses never contain agent identifiers, filesystem paths, secrets, +tokens, or raw error internals — only listener identities (already visible on +`/api/listeners/list`), counts, and classified categories. + +## Health semantics + +Each listener and the server as a whole report one of three states: + +- `healthy` — working as configured, no recent failure indicators. +- `degraded` — still serving, but recent task or build failures were recorded + inside the failure window. +- `failing` — cannot serve its purpose right now. + +Listener rules, evaluated in order: + +1. Status `ERROR` → `failing`. +2. Listener runtime counters unreadable → `failing` with + `last_error_class: "telemetry_unavailable"` (counters are not fabricated). +3. Status `ACTIVE` with `recent_task_failures > 0` or + `recent_build_failures > 0` → `degraded`. +4. Otherwise → `healthy`. A `STOPPED` listener is a deliberate operator + action and does not degrade the server, though its counters are still + reported. + +Server rules: + +1. Storage `unavailable` → `failing`. +2. Otherwise the worst health across all listeners. + +Because the failure window is trailing, a resolved incident stops degrading +the server once its failures age out — no operator action is required. + +## Local deployment checks + +With the server running locally on the default port: + +```sh +curl -sk https://127.0.0.1:8443/api/health +curl -sk https://127.0.0.1:8443/api/ready +curl -sk https://127.0.0.1:8443/api/telemetry +``` + +From a non-loopback host, add the operator token: + +```sh +curl -sk -H "X-Operator-Token: $MICROC2_OPERATOR_TOKEN" \ + https://operator-host:8443/api/ready +``` + +## CI smoke checks + +A minimal smoke check starts the server with a test config and polls +readiness until storage answers, then asserts the telemetry contract: + +```sh +# Wait for readiness (loopback needs no token). +until curl -skf https://127.0.0.1:8443/api/ready > /dev/null; do sleep 1; done + +# Assert the telemetry envelope parses and reports a known state. +curl -skf https://127.0.0.1:8443/api/telemetry \ + | grep -q '"status":"healthy"' +``` + +Use the HTTP status code of `/api/ready` as the gate (it returns `503` until +ready) and the `status` field of `/api/telemetry` to assert healthy, +degraded, or failing semantics. `/api/health` is the right probe when only +process liveness matters, for example as a container `HEALTHCHECK`. diff --git a/server/cmd/server.go b/server/cmd/server.go index a8ef66d..bc8b162 100644 --- a/server/cmd/server.go +++ b/server/cmd/server.go @@ -198,6 +198,10 @@ func main() { // Set up listener management routes listenerHandlers.RegisterRoutes(operatorMux) + // Set up health, readiness, and runtime telemetry routes + healthHandlers := api.NewHealthHandlers(serverManager.GetListenerManager(), stateDatabase) + healthHandlers.RegisterRoutes(operatorMux) + // Set up payload generator routes payloadHandler.RegisterRoutes(operatorMux) diff --git a/server/internal/behaviour/http_polling.go b/server/internal/behaviour/http_polling.go index ba37036..a93b10c 100644 --- a/server/internal/behaviour/http_polling.go +++ b/server/internal/behaviour/http_polling.go @@ -62,6 +62,10 @@ type durableLegacyResultPager interface { ) ([]tasks.LegacyResult, int, bool, error) } +type taskQueueStatsStore interface { + QueueStats(failedSince time.Time) (tasks.QueueStats, error) +} + type taskSummaryPager interface { ListTaskSummariesPage( agentID string, @@ -914,6 +918,49 @@ func (p *HTTPPollingProtocol) GetAllAgents() map[string]interface{} { return result } +// RuntimeStats summarizes live agent presence and task backlog for operator +// telemetry. It reads the same runtime agent registry and task store that +// heartbeats and dispatch mutate; it is never a separate source of truth. +type RuntimeStats struct { + // ActiveAgents counts agents that completed a heartbeat during this + // server process lifetime. Durable agent history loaded at startup does + // not count until the agent heartbeats again. + ActiveAgents int + // PendingTasks counts tasks waiting for an agent or dispatched but not + // yet completed. + PendingTasks int + // RecentFailedTasks counts tasks that failed at or after the + // caller-supplied window start. + RecentFailedTasks int +} + +// RuntimeStats reports current listener runtime counters. failedSince bounds +// the recent-failure window; agent presence and pending counts are current +// point-in-time values. Stores without aggregate support report zero task +// counters rather than an error. +func (p *HTTPPollingProtocol) RuntimeStats(failedSince time.Time) (RuntimeStats, error) { + var stats RuntimeStats + p.agents.Lock() + for agentID := range p.agents.list { + if p.agents.activeThisBoot[agentID] { + stats.ActiveAgents++ + } + } + p.agents.Unlock() + + counter, ok := p.taskStore.(taskQueueStatsStore) + if !ok { + return stats, nil + } + queue, err := counter.QueueStats(failedSince) + if err != nil { + return RuntimeStats{}, err + } + stats.PendingTasks = queue.Pending + stats.RecentFailedTasks = queue.RecentFailed + return stats, nil +} + func (p *HTTPPollingProtocol) AgentLastSeen(agentID string) (time.Time, bool) { p.agents.Lock() defer p.agents.Unlock() diff --git a/server/internal/handlers/api/health_handlers.go b/server/internal/handlers/api/health_handlers.go new file mode 100644 index 0000000..ef6d99e --- /dev/null +++ b/server/internal/handlers/api/health_handlers.go @@ -0,0 +1,357 @@ +package api + +import ( + "encoding/json" + "log" + "net/http" + "strings" + "time" + + "microc2/server/internal/behaviour" + "microc2/server/internal/listeners" + "microc2/server/internal/persistence" +) + +// Health states reported by the operator telemetry endpoints. The semantics +// are deliberately small and deterministic so external tooling can alert on +// them without parsing logs: +// +// - healthy: the component works as configured and shows no recent failure +// indicators. +// - degraded: the component still serves, but recent task or payload build +// failures were recorded inside the telemetry failure window. +// - failing: the component cannot serve its purpose right now (listener in +// ERROR state, storage unavailable, or telemetry itself unreadable). +const ( + HealthStateHealthy = "healthy" + HealthStateDegraded = "degraded" + HealthStateFailing = "failing" +) + +// Listener error classes. Raw listener error strings can contain filesystem +// paths, bind addresses, and TLS library internals, so telemetry only ever +// exposes these fixed categories. +const ( + listenerErrorClassBindFailed = "bind_failed" + listenerErrorClassTLSConfigInvalid = "tls_config_invalid" + listenerErrorClassRuntimeError = "runtime_error" + listenerErrorClassTelemetryUnavailable = "telemetry_unavailable" +) + +// Storage states shared by the readiness and telemetry responses. +const ( + storageStateOK = "ok" + storageStateDisabled = "disabled" + storageStateUnavailable = "unavailable" +) + +// telemetryFailureWindow bounds "recent" task and build failure indicators. +// Counters inside the window restart from zero once failures age out, so a +// resolved incident stops degrading the server without operator action. +const telemetryFailureWindow = time.Hour + +// buildFailureWindowLayout truncates the recent-build-failure comparison to +// whole seconds. payload_builds.created_at values are UTC RFC3339Nano +// strings, so a second-precision lower bound compares correctly against them +// while a nanosecond bound could mismatch variable-width fractions. +const buildFailureWindowLayout = "2006-01-02T15:04:05" + +// HealthHandlers serves the machine-readable operator health surface: +// liveness, readiness, and runtime telemetry. All routes are registered on +// the operator mux, so they sit behind the same loopback-or-operator-token +// guard as every other operator API. +type HealthHandlers struct { + manager *listeners.ListenerManager + database *persistence.Database + startedAt time.Time + now func() time.Time +} + +// NewHealthHandlers creates the health handler set over the existing +// listener manager and durable state handle. Both may be nil: a nil manager +// reports zero listeners, and a nil database reports storage as disabled. +func NewHealthHandlers( + manager *listeners.ListenerManager, + database *persistence.Database, +) *HealthHandlers { + return newHealthHandlersWithClock(manager, database, time.Now) +} + +func newHealthHandlersWithClock( + manager *listeners.ListenerManager, + database *persistence.Database, + now func() time.Time, +) *HealthHandlers { + if now == nil { + now = time.Now + } + return &HealthHandlers{ + manager: manager, + database: database, + startedAt: now().UTC(), + now: now, + } +} + +// RegisterRoutes registers all health routes on the provided mux. +func (h *HealthHandlers) RegisterRoutes(mux *http.ServeMux) { + mux.HandleFunc("/api/health", h.HandleHealth) + mux.HandleFunc("/api/ready", h.HandleReady) + mux.HandleFunc("/api/telemetry", h.HandleTelemetry) +} + +// HandleHealth is the liveness probe. It answers 200 whenever the process +// can serve HTTP; it performs no I/O beyond encoding the response, so it +// stays deterministic and inexpensive under load. +func (h *HealthHandlers) HandleHealth(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + sendJSONResponse(w, map[string]interface{}{ + "status": "ok", + "uptime_seconds": h.uptimeSeconds(), + }) +} + +// HandleReady is the readiness probe. It reports 200 only when every +// configured dependency answers a trivial check; today that is a single +// `SELECT 1` against durable storage. A server without configured storage is +// ready with storage reported as disabled. +func (h *HealthHandlers) HandleReady(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + storage := h.storageState() + status := "ready" + code := http.StatusOK + if storage == storageStateUnavailable { + status = "not_ready" + code = http.StatusServiceUnavailable + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + if err := json.NewEncoder(w).Encode(map[string]interface{}{ + "status": status, + "checks": map[string]string{"storage": storage}, + }); err != nil { + log.Printf("[ERROR] Failed to encode readiness response: %v", err) + } +} + +type listenerTelemetry struct { + ID string `json:"id"` + Name string `json:"name"` + Protocol string `json:"protocol"` + Port int `json:"port"` + Status string `json:"status"` + Health string `json:"health"` + LastErrorClass string `json:"last_error_class,omitempty"` + ActiveAgents int `json:"active_agents"` + QueueDepth int `json:"queue_depth"` + RecentTaskFailures int `json:"recent_task_failures"` + RecentBuildFailures int `json:"recent_build_failures"` +} + +type buildTelemetry struct { + RecentFailures int `json:"recent_failures"` +} + +type telemetryResponse struct { + Status string `json:"status"` + GeneratedAt string `json:"generated_at"` + UptimeSeconds int64 `json:"uptime_seconds"` + Storage string `json:"storage"` + Listeners []listenerTelemetry `json:"listeners"` + Builds buildTelemetry `json:"builds"` +} + +// HandleTelemetry reports the operator server runtime summary: per-listener +// status, sanitized last-error class, active agent count, queue depth, and +// recent task/build failure indicators, plus the aggregate server health. +// Responses never contain agent identifiers, filesystem paths, secrets, or +// raw error strings. +func (h *HealthHandlers) HandleTelemetry(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + now := h.now().UTC() + windowStart := now.Add(-telemetryFailureWindow) + + storage := h.storageState() + buildFailures, totalBuildFailures, buildsErr := h.recentBuildFailures(windowStart) + if buildsErr != nil { + // The build table lives in the same database as the readiness probe; + // surface its failure as a storage problem instead of fabricating + // plausible-looking zero counters. + log.Printf("[ERROR] Failed to load recent build failures: %v", buildsErr) + storage = storageStateUnavailable + } + + response := telemetryResponse{ + Status: HealthStateHealthy, + GeneratedAt: now.Format(time.RFC3339), + UptimeSeconds: h.uptimeSeconds(), + Storage: storage, + Listeners: []listenerTelemetry{}, + Builds: buildTelemetry{RecentFailures: totalBuildFailures}, + } + if storage == storageStateUnavailable { + response.Status = HealthStateFailing + } + if h.manager != nil { + for _, listener := range h.manager.ListListeners() { + entry := h.summarizeListener(listener, windowStart, buildFailures) + response.Listeners = append(response.Listeners, entry) + response.Status = worstHealth(response.Status, entry.Health) + } + } + sendJSONResponse(w, response) +} + +func (h *HealthHandlers) summarizeListener( + listener *listeners.Listener, + windowStart time.Time, + buildFailures map[string]int, +) listenerTelemetry { + snapshot := listener.Snapshot() + entry := listenerTelemetry{ + ID: snapshot.Config.ID, + Name: snapshot.Config.Name, + Protocol: snapshot.Config.Protocol, + Port: snapshot.Config.Port, + Status: string(snapshot.Status), + Health: HealthStateHealthy, + LastErrorClass: classifyListenerError(snapshot.Error), + } + + statsAvailable := false + provider, ok := listener.Protocol.(interface { + RuntimeStats(failedSince time.Time) (behaviour.RuntimeStats, error) + }) + if ok { + stats, err := provider.RuntimeStats(windowStart) + if err != nil { + log.Printf("[ERROR] Failed to load listener runtime stats: %v", err) + entry.LastErrorClass = listenerErrorClassTelemetryUnavailable + } else { + statsAvailable = true + entry.ActiveAgents = stats.ActiveAgents + entry.QueueDepth = stats.PendingTasks + entry.RecentTaskFailures = stats.RecentFailedTasks + } + } + entry.RecentBuildFailures = buildFailures[snapshot.Config.ID] + + switch { + case snapshot.Status == listeners.StatusError: + entry.Health = HealthStateFailing + case ok && !statsAvailable: + // Telemetry for this listener could not be read; treat it as failing + // rather than reporting unverifiable zeros. + entry.Health = HealthStateFailing + case snapshot.Status == listeners.StatusActive && + (entry.RecentTaskFailures > 0 || entry.RecentBuildFailures > 0): + entry.Health = HealthStateDegraded + } + return entry +} + +// recentBuildFailures counts failed and interrupted payload builds per +// listener since windowStart. Completed builds whose artifacts later fail +// reconciliation (missing/corrupt) are not build failures and stay out of +// this indicator. +func (h *HealthHandlers) recentBuildFailures( + windowStart time.Time, +) (map[string]int, int, error) { + counts := make(map[string]int) + if h.database == nil || h.database.SQL() == nil { + return counts, 0, nil + } + rows, err := h.database.SQL().Query( + `SELECT listener_id, COUNT(*) + FROM payload_builds + WHERE state IN ('failed', 'interrupted') AND created_at >= ? + GROUP BY listener_id`, + windowStart.UTC().Format(buildFailureWindowLayout), + ) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + total := 0 + for rows.Next() { + var listenerID string + var count int + if err := rows.Scan(&listenerID, &count); err != nil { + return nil, 0, err + } + counts[listenerID] = count + total += count + } + if err := rows.Err(); err != nil { + return nil, 0, err + } + return counts, total, nil +} + +// storageState performs the cheapest possible durable-storage check. It runs +// on every readiness and telemetry request, so it must stay a trivial +// constant-time probe rather than a schema or content scan. +func (h *HealthHandlers) storageState() string { + if h.database == nil || h.database.SQL() == nil { + return storageStateDisabled + } + var one int + if err := h.database.SQL().QueryRow("SELECT 1").Scan(&one); err != nil { + log.Printf("[ERROR] Durable storage readiness probe failed: %v", err) + return storageStateUnavailable + } + return storageStateOK +} + +func (h *HealthHandlers) uptimeSeconds() int64 { + uptime := h.now().UTC().Sub(h.startedAt) + if uptime < 0 { + return 0 + } + return int64(uptime / time.Second) +} + +// classifyListenerError maps a raw listener error string to a fixed category. +// Raw errors can embed certificate paths, bind addresses, and TLS library +// internals, none of which may leave the process through telemetry. +func classifyListenerError(raw string) string { + if raw == "" { + return "" + } + lowered := strings.ToLower(raw) + switch { + case strings.Contains(lowered, "bind"), + strings.Contains(lowered, "address already in use"), + strings.Contains(lowered, "listen tcp"): + return listenerErrorClassBindFailed + case strings.Contains(lowered, "certificate"), + strings.Contains(lowered, "tls"), + strings.Contains(lowered, "x509"): + return listenerErrorClassTLSConfigInvalid + default: + return listenerErrorClassRuntimeError + } +} + +func worstHealth(current, next string) string { + rank := map[string]int{ + HealthStateHealthy: 0, + HealthStateDegraded: 1, + HealthStateFailing: 2, + } + if rank[next] > rank[current] { + return next + } + return current +} diff --git a/server/internal/handlers/api/health_handlers_test.go b/server/internal/handlers/api/health_handlers_test.go new file mode 100644 index 0000000..14acb39 --- /dev/null +++ b/server/internal/handlers/api/health_handlers_test.go @@ -0,0 +1,574 @@ +package api + +import ( + "bytes" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "microc2/server/internal/behaviour" + "microc2/server/internal/listeners" + "microc2/server/internal/persistence" + "microc2/server/internal/tasks" +) + +func TestHealthEndpointReportsLiveness(t *testing.T) { + clock := healthTestClock() + handler := newHealthHandlersWithClock(nil, nil, clock) + mux := http.NewServeMux() + handler.RegisterRoutes(mux) + + response := serveHealthRequest(mux, http.MethodGet, "/api/health") + if response.Code != http.StatusOK { + t.Fatalf("health status = %d: %s", response.Code, response.Body.String()) + } + var body map[string]interface{} + if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil { + t.Fatalf("decode health response: %v", err) + } + if body["status"] != "ok" { + t.Fatalf("health status field = %#v, want ok", body["status"]) + } + if body["uptime_seconds"] != float64(0) { + t.Fatalf("uptime_seconds = %#v, want 0 with a frozen clock", body["uptime_seconds"]) + } + + response = serveHealthRequest(mux, http.MethodPost, "/api/health") + if response.Code != http.StatusMethodNotAllowed { + t.Fatalf("POST /api/health status = %d, want 405", response.Code) + } +} + +func TestReadyEndpointStorageStates(t *testing.T) { + t.Run("no storage configured", func(t *testing.T) { + handler := newHealthHandlersWithClock(nil, nil, time.Now) + response := serveHealthRequest(healthTestMux(handler), http.MethodGet, "/api/ready") + assertReadyResponse(t, response, http.StatusOK, "ready", storageStateDisabled) + }) + + t.Run("healthy storage", func(t *testing.T) { + database := openHealthTestDatabase(t) + handler := newHealthHandlersWithClock(nil, database, time.Now) + response := serveHealthRequest(healthTestMux(handler), http.MethodGet, "/api/ready") + assertReadyResponse(t, response, http.StatusOK, "ready", storageStateOK) + }) + + t.Run("unavailable storage", func(t *testing.T) { + database := openHealthTestDatabase(t) + if err := database.Close(); err != nil { + t.Fatalf("close database: %v", err) + } + handler := newHealthHandlersWithClock(nil, database, time.Now) + response := serveHealthRequest(healthTestMux(handler), http.MethodGet, "/api/ready") + assertReadyResponse( + t, + response, + http.StatusServiceUnavailable, + "not_ready", + storageStateUnavailable, + ) + }) + + t.Run("rejects non-GET methods", func(t *testing.T) { + handler := newHealthHandlersWithClock(nil, nil, time.Now) + response := serveHealthRequest(healthTestMux(handler), http.MethodPost, "/api/ready") + if response.Code != http.StatusMethodNotAllowed { + t.Fatalf("POST /api/ready status = %d, want 405", response.Code) + } + }) +} + +func TestTelemetryHealthyActiveListener(t *testing.T) { + manager := newHealthTestManager(t) + listener, proto := addHealthTestListener(t, manager, "listener-one", 49101) + heartbeatHealthTestAgent(t, proto, "agent-health-one") + listener.Status = listeners.StatusActive + createHealthTestTask(t, proto, "agent-health-one", "whoami") + + handler := newHealthHandlersWithClock(manager, nil, time.Now) + response := serveHealthRequest(healthTestMux(handler), http.MethodGet, "/api/telemetry") + if response.Code != http.StatusOK { + t.Fatalf("telemetry status = %d: %s", response.Code, response.Body.String()) + } + if strings.Contains(response.Body.String(), "agent-health-one") { + t.Fatalf("telemetry leaked an agent identifier: %s", response.Body.String()) + } + + body := decodeTelemetryResponse(t, response) + if body.Status != HealthStateHealthy { + t.Fatalf("server health = %q, want healthy", body.Status) + } + if body.Storage != storageStateDisabled { + t.Fatalf("storage = %q, want disabled", body.Storage) + } + if body.Builds.RecentFailures != 0 { + t.Fatalf("recent build failures = %d, want 0", body.Builds.RecentFailures) + } + if len(body.Listeners) != 1 { + t.Fatalf("listeners = %#v, want exactly one", body.Listeners) + } + entry := body.Listeners[0] + if entry.ID != "listener-one" || entry.Status != string(listeners.StatusActive) { + t.Fatalf("unexpected listener entry: %#v", entry) + } + if entry.Health != HealthStateHealthy { + t.Fatalf("listener health = %q, want healthy", entry.Health) + } + if entry.ActiveAgents != 1 { + t.Fatalf("active agents = %d, want 1", entry.ActiveAgents) + } + if entry.QueueDepth != 1 { + t.Fatalf("queue depth = %d, want 1 queued task", entry.QueueDepth) + } + if entry.RecentTaskFailures != 0 || entry.RecentBuildFailures != 0 { + t.Fatalf("unexpected failure counters: %#v", entry) + } + if entry.LastErrorClass != "" { + t.Fatalf("last error class = %q, want empty", entry.LastErrorClass) + } +} + +func TestTelemetryErroredListenerFailsAndSanitizesError(t *testing.T) { + manager := newHealthTestManager(t) + listener, _ := addHealthTestListener(t, manager, "listener-two", 49102) + listener.SetError(errors.New( + "failed to load TLS certificate for listener listener-two: " + + "open /etc/private/certs/listener.pem: no such file or directory", + )) + + handler := newHealthHandlersWithClock(manager, nil, time.Now) + response := serveHealthRequest(healthTestMux(handler), http.MethodGet, "/api/telemetry") + if response.Code != http.StatusOK { + t.Fatalf("telemetry status = %d: %s", response.Code, response.Body.String()) + } + for _, leaked := range []string{ + "/etc/private/certs/listener.pem", + "no such file or directory", + } { + if strings.Contains(response.Body.String(), leaked) { + t.Fatalf("telemetry leaked raw error internals %q: %s", leaked, response.Body.String()) + } + } + + body := decodeTelemetryResponse(t, response) + if body.Status != HealthStateFailing { + t.Fatalf("server health = %q, want failing", body.Status) + } + if len(body.Listeners) != 1 { + t.Fatalf("listeners = %#v, want exactly one", body.Listeners) + } + entry := body.Listeners[0] + if entry.Health != HealthStateFailing { + t.Fatalf("listener health = %q, want failing", entry.Health) + } + if entry.Status != string(listeners.StatusError) { + t.Fatalf("listener status = %q, want ERROR", entry.Status) + } + if entry.LastErrorClass != listenerErrorClassTLSConfigInvalid { + t.Fatalf("last error class = %q, want tls_config_invalid", entry.LastErrorClass) + } +} + +func TestTelemetryDegradedListenerWithQueueAndRecentTaskFailure(t *testing.T) { + manager := newHealthTestManager(t) + listener, proto := addHealthTestListener(t, manager, "listener-three", 49103) + heartbeatHealthTestAgent(t, proto, "agent-health-three") + listener.Status = listeners.StatusActive + + failedTask := createHealthTestTask(t, proto, "agent-health-three", "id") + createHealthTestTask(t, proto, "agent-health-three", "uname -a") + failHealthTestTask(t, proto, failedTask) + + handler := newHealthHandlersWithClock(manager, nil, time.Now) + response := serveHealthRequest(healthTestMux(handler), http.MethodGet, "/api/telemetry") + if response.Code != http.StatusOK { + t.Fatalf("telemetry status = %d: %s", response.Code, response.Body.String()) + } + + body := decodeTelemetryResponse(t, response) + if body.Status != HealthStateDegraded { + t.Fatalf("server health = %q, want degraded", body.Status) + } + if len(body.Listeners) != 1 { + t.Fatalf("listeners = %#v, want exactly one", body.Listeners) + } + entry := body.Listeners[0] + if entry.Health != HealthStateDegraded { + t.Fatalf("listener health = %q, want degraded", entry.Health) + } + if entry.QueueDepth != 1 { + t.Fatalf("queue depth = %d, want 1 remaining queued task", entry.QueueDepth) + } + if entry.RecentTaskFailures != 1 { + t.Fatalf("recent task failures = %d, want 1", entry.RecentTaskFailures) + } +} + +func TestTelemetryRecentBuildFailures(t *testing.T) { + database := openHealthTestDatabase(t) + manager := newHealthTestManager(t) + listener, proto := addHealthTestListener(t, manager, "listener-builds", 49104) + heartbeatHealthTestAgent(t, proto, "agent-health-builds") + listener.Status = listeners.StatusActive + + now := time.Now().UTC() + insertHealthTestPayloadBuild(t, database, "build-recent-failed", "listener-builds", + "failed", now.Add(-30*time.Minute)) + insertHealthTestPayloadBuild(t, database, "build-old-failed", "listener-builds", + "failed", now.Add(-2*time.Hour)) + insertHealthTestPayloadBuild(t, database, "build-recent-completed", "listener-builds", + "completed", now.Add(-10*time.Minute)) + + handler := newHealthHandlersWithClock(manager, database, time.Now) + response := serveHealthRequest(healthTestMux(handler), http.MethodGet, "/api/telemetry") + if response.Code != http.StatusOK { + t.Fatalf("telemetry status = %d: %s", response.Code, response.Body.String()) + } + + body := decodeTelemetryResponse(t, response) + if body.Status != HealthStateDegraded { + t.Fatalf("server health = %q, want degraded", body.Status) + } + if body.Storage != storageStateOK { + t.Fatalf("storage = %q, want ok", body.Storage) + } + if body.Builds.RecentFailures != 1 { + t.Fatalf("recent build failures = %d, want 1 inside the window", body.Builds.RecentFailures) + } + if len(body.Listeners) != 1 { + t.Fatalf("listeners = %#v, want exactly one", body.Listeners) + } + entry := body.Listeners[0] + if entry.RecentBuildFailures != 1 { + t.Fatalf("listener recent build failures = %d, want 1", entry.RecentBuildFailures) + } + if entry.Health != HealthStateDegraded { + t.Fatalf("listener health = %q, want degraded", entry.Health) + } +} + +func TestTelemetryUnavailableStorageMarksServerFailing(t *testing.T) { + database := openHealthTestDatabase(t) + if err := database.Close(); err != nil { + t.Fatalf("close database: %v", err) + } + handler := newHealthHandlersWithClock(nil, database, time.Now) + response := serveHealthRequest(healthTestMux(handler), http.MethodGet, "/api/telemetry") + if response.Code != http.StatusOK { + t.Fatalf("telemetry status = %d: %s", response.Code, response.Body.String()) + } + body := decodeTelemetryResponse(t, response) + if body.Status != HealthStateFailing { + t.Fatalf("server health = %q, want failing", body.Status) + } + if body.Storage != storageStateUnavailable { + t.Fatalf("storage = %q, want unavailable", body.Storage) + } +} + +func TestTelemetryRejectsNonGetMethods(t *testing.T) { + handler := newHealthHandlersWithClock(nil, nil, time.Now) + response := serveHealthRequest(healthTestMux(handler), http.MethodPost, "/api/telemetry") + if response.Code != http.StatusMethodNotAllowed { + t.Fatalf("POST /api/telemetry status = %d, want 405", response.Code) + } +} + +func TestClassifyListenerError(t *testing.T) { + cases := []struct { + name string + raw string + want string + }{ + {"empty", "", ""}, + { + "bind conflict", + "failed to bind listener lab on 0.0.0.0:8080: listen tcp 0.0.0.0:8080: bind: address already in use", + listenerErrorClassBindFailed, + }, + { + "tls certificate path", + "failed to load TLS certificate for listener lab: open /srv/certs/lab.pem: no such file or directory", + listenerErrorClassTLSConfigInvalid, + }, + { + "tls x509 internals", + "tls: failed to parse certificate: x509: malformed certificate", + listenerErrorClassTLSConfigInvalid, + }, + { + "unknown runtime error", + "HTTP server error: unexpected EOF", + listenerErrorClassRuntimeError, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := classifyListenerError(tc.raw); got != tc.want { + t.Fatalf("classifyListenerError(%q) = %q, want %q", tc.raw, got, tc.want) + } + }) + } +} + +func healthTestClock() func() time.Time { + frozen := time.Date(2026, 7, 25, 12, 0, 0, 0, time.UTC) + return func() time.Time { return frozen } +} + +func healthTestMux(handler *HealthHandlers) *http.ServeMux { + mux := http.NewServeMux() + handler.RegisterRoutes(mux) + return mux +} + +func serveHealthRequest(mux *http.ServeMux, method, path string) *httptest.ResponseRecorder { + recorder := httptest.NewRecorder() + mux.ServeHTTP(recorder, httptest.NewRequest(method, path, nil)) + return recorder +} + +func assertReadyResponse( + t *testing.T, + response *httptest.ResponseRecorder, + wantCode int, + wantStatus string, + wantStorage string, +) { + t.Helper() + if response.Code != wantCode { + t.Fatalf("ready status = %d, want %d: %s", response.Code, wantCode, response.Body.String()) + } + var body struct { + Status string `json:"status"` + Checks map[string]string `json:"checks"` + } + if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil { + t.Fatalf("decode readiness response: %v", err) + } + if body.Status != wantStatus { + t.Fatalf("readiness status = %q, want %q", body.Status, wantStatus) + } + if body.Checks["storage"] != wantStorage { + t.Fatalf("storage check = %q, want %q", body.Checks["storage"], wantStorage) + } +} + +func decodeTelemetryResponse( + t *testing.T, + response *httptest.ResponseRecorder, +) telemetryResponse { + t.Helper() + var body telemetryResponse + if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil { + t.Fatalf("decode telemetry response: %v", err) + } + return body +} + +func openHealthTestDatabase(t *testing.T) *persistence.Database { + t.Helper() + database, err := persistence.Open(filepath.Join(t.TempDir(), "microc2.db")) + if err != nil { + t.Fatalf("open database: %v", err) + } + t.Cleanup(func() { + _ = database.Close() + }) + return database +} + +// newHealthTestManager creates a non-durable isolated-lab listener manager. +// The compatibility constructors resolve static/listeners relative to the +// working directory, so tests run inside a temporary directory. +func newHealthTestManager(t *testing.T) *listeners.ListenerManager { + t.Helper() + tempDir := t.TempDir() + oldCwd, err := os.Getwd() + if err != nil { + t.Fatalf("get cwd: %v", err) + } + if err := os.Chdir(tempDir); err != nil { + t.Fatalf("switch to temp cwd: %v", err) + } + t.Cleanup(func() { + if err := os.Chdir(oldCwd); err != nil { + t.Errorf("restore cwd: %v", err) + } + }) + return listeners.NewListenerManagerForIsolatedLab(nil) +} + +func addHealthTestListener( + t *testing.T, + manager *listeners.ListenerManager, + listenerID string, + port int, +) (*listeners.Listener, *behaviour.HTTPPollingProtocol) { + t.Helper() + listener, err := listeners.NewListenerForIsolatedLab(listeners.ListenerConfig{ + ID: listenerID, + Name: listenerID, + Protocol: "http", + BindHost: "127.0.0.1", + Port: port, + }) + if err != nil { + t.Fatalf("create listener %s: %v", listenerID, err) + } + proto, ok := listener.Protocol.(*behaviour.HTTPPollingProtocol) + if !ok { + t.Fatalf("listener %s protocol = %T, want HTTP polling", listenerID, listener.Protocol) + } + if err := manager.AddListener(listener); err != nil { + t.Fatalf("add listener %s: %v", listenerID, err) + } + return listener, proto +} + +func heartbeatHealthTestAgent( + t *testing.T, + proto *behaviour.HTTPPollingProtocol, + agentID string, +) { + t.Helper() + heartbeat := []byte( + `{"id":"` + agentID + `","os":"linux","hostname":"workstation","ip":"127.0.0.1"}`, + ) + if err := proto.HandleAgentHeartbeat(heartbeat); err != nil { + t.Fatalf("register agent heartbeat: %v", err) + } +} + +func createHealthTestTask( + t *testing.T, + proto *behaviour.HTTPPollingProtocol, + agentID string, + command string, +) tasks.Task { + t.Helper() + expiresIn := 300 + task, err := proto.CreateTask(agentID, tasks.CreateRequest{ + SchemaVersion: tasks.SchemaVersion, + Type: tasks.TypeShell, + Arguments: tasks.ShellArguments{Command: command}, + TimeoutSeconds: 30, + ExpiresInSeconds: &expiresIn, + }) + if err != nil { + t.Fatalf("create task %q: %v", command, err) + } + return task +} + +// failHealthTestTask drives one task through dispatch and running into the +// failed terminal state, mirroring completeTypedTask with a failed outcome. +func failHealthTestTask( + t *testing.T, + proto *behaviour.HTTPPollingProtocol, + task tasks.Task, +) { + t.Helper() + handler := proto.GetHTTPHandler() + + recorder := httptest.NewRecorder() + request := httptest.NewRequest( + http.MethodGet, + "/api/agent/"+task.AgentID+"/tasks", + nil, + ) + handler.ServeHTTP(recorder, request) + if recorder.Code != http.StatusOK { + t.Fatalf("dispatch task: %d: %s", recorder.Code, recorder.Body.String()) + } + var dispatched tasks.Task + if err := json.Unmarshal(recorder.Body.Bytes(), &dispatched); err != nil { + t.Fatalf("decode dispatched task: %v", err) + } + if dispatched.ID != task.ID || dispatched.DispatchedAt == nil { + t.Fatalf("dispatched task = %#v, want %q", dispatched, task.ID) + } + + startedAt := dispatched.DispatchedAt.Add(time.Second) + statusBody, err := json.Marshal(tasks.StatusUpdate{ + SchemaVersion: tasks.SchemaVersion, + TaskID: task.ID, + AgentID: task.AgentID, + Status: tasks.StatusRunning, + Timestamp: startedAt, + }) + if err != nil { + t.Fatalf("marshal running update: %v", err) + } + recorder = httptest.NewRecorder() + request = httptest.NewRequest( + http.MethodPost, + "/api/agent/"+task.AgentID+"/tasks/"+task.ID+"/status", + bytes.NewReader(statusBody), + ) + handler.ServeHTTP(recorder, request) + if recorder.Code != http.StatusOK { + t.Fatalf("mark task running: %d: %s", recorder.Code, recorder.Body.String()) + } + + resultBody, err := json.Marshal(tasks.Result{ + SchemaVersion: tasks.SchemaVersion, + TaskID: task.ID, + AgentID: task.AgentID, + Outcome: tasks.OutcomeFailed, + StartedAt: startedAt, + CompletedAt: startedAt.Add(time.Second), + Error: "simulated task failure", + }) + if err != nil { + t.Fatalf("marshal failed result: %v", err) + } + recorder = httptest.NewRecorder() + request = httptest.NewRequest( + http.MethodPost, + "/api/agent/"+task.AgentID+"/results", + bytes.NewReader(resultBody), + ) + handler.ServeHTTP(recorder, request) + if recorder.Code != http.StatusOK { + t.Fatalf("fail task: %d: %s", recorder.Code, recorder.Body.String()) + } +} + +func insertHealthTestPayloadBuild( + t *testing.T, + database *persistence.Database, + payloadID string, + listenerID string, + state string, + createdAt time.Time, +) { + t.Helper() + if _, err := database.SQL().Exec( + `INSERT INTO payload_builds ( + id, payload_id, listener_id, mutation_seed, filename, + relative_path, size, sha256, created_at, state, + state_detail, provenance_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + payloadID, + payloadID, + listenerID, + "0123456789abcdef", + "agent", + "release/"+payloadID+"/agent", + 0, + "", + createdAt.UTC().Format(time.RFC3339Nano), + state, + "", + []byte("{}"), + ); err != nil { + t.Fatalf("insert payload build %s: %v", payloadID, err) + } +} diff --git a/server/internal/tasks/durable_store.go b/server/internal/tasks/durable_store.go index 42597a4..be0fa55 100644 --- a/server/internal/tasks/durable_store.go +++ b/server/internal/tasks/durable_store.go @@ -300,6 +300,33 @@ func (s *DurableStore) List(agentID string) ([]Task, error) { return tasks, nil } +// QueueStats aggregates pending and recently failed tasks for this listener +// scope with two constant-time COUNT aggregates over the durable tasks table. +// It never scans task payloads, so it stays cheap enough to serve on every +// telemetry request. failedSince bounds the recent-failure window; pending +// counts are not windowed. +func (s *DurableStore) QueueStats(failedSince time.Time) (QueueStats, error) { + row := s.db.SQL().QueryRow( + `SELECT + COALESCE(SUM(CASE WHEN status IN (?, ?) THEN 1 ELSE 0 END), 0), + COALESCE(SUM( + CASE WHEN status = ? AND server_completed_at >= ? THEN 1 ELSE 0 END + ), 0) + FROM tasks + WHERE listener_id = ?`, + string(StatusQueued), + string(StatusDispatched), + string(StatusFailed), + formatDurableTime(failedSince), + s.listenerID, + ) + var stats QueueStats + if err := row.Scan(&stats.Pending, &stats.RecentFailed); err != nil { + return QueueStats{}, fmt.Errorf("summarize durable task queue: %w", err) + } + return stats, nil +} + // ListTaskSummariesPage returns newest-first task metadata without selecting // stdout or stderr. Keeping the collection query separate from Get prevents a // bounded operator page from materializing the full retained result corpus. diff --git a/server/internal/tasks/store.go b/server/internal/tasks/store.go index 2002ace..21346e5 100644 --- a/server/internal/tasks/store.go +++ b/server/internal/tasks/store.go @@ -298,6 +298,40 @@ func (s *Store) CompleteWithInfo(agentID string, result Result) (Task, Completio return cloneTask(task), info, nil } +// QueueStats summarizes outstanding and recently failed work in a store. +// It is the machine-readable telemetry view over the same task lifecycle +// state that dispatch and completion mutate; it never guesses from logs. +type QueueStats struct { + // Pending counts tasks waiting for an agent (queued) or delivered to an + // agent but not yet completed (dispatched). + Pending int + // RecentFailed counts tasks that reached the failed terminal state at or + // after the caller-supplied window start. + RecentFailed int +} + +// QueueStats aggregates pending and recently failed tasks across all agents +// in the store. failedSince bounds the recent-failure window; pending counts +// are not windowed. +func (s *Store) QueueStats(failedSince time.Time) (QueueStats, error) { + failedSince = normalizeTime(failedSince) + + s.mu.RLock() + defer s.mu.RUnlock() + var stats QueueStats + for _, task := range s.byID { + switch task.Status { + case StatusQueued, StatusDispatched: + stats.Pending++ + case StatusFailed: + if task.CompletedAt != nil && !task.CompletedAt.Before(failedSince) { + stats.RecentFailed++ + } + } + } + return stats, nil +} + func (s *Store) Cancel(agentID, taskID string) (Task, error) { if err := validateIDs(agentID, taskID); err != nil { return Task{}, err