From 637c92c762388850465a7a70c51079e05887b9e4 Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Thu, 29 Oct 2026 12:30:36 +0100 Subject: [PATCH] =?UTF-8?q?Operator=20actions=20for=20the=20agent:=20resta?= =?UTF-8?q?rt=5Fcontroller=20from=20the=20hub=20(closed=20list,=20the=20su?= =?UTF-8?q?pervisor's=20guards,=20the=20brake=20counts=20operator=20restar?= =?UTF-8?q?ts);=20R-1344:=20a=20running-but-unhealthy=20controller=20raise?= =?UTF-8?q?s=20unresponsive=20=E2=80=94=20alarm=20only;=20tests=20red=20fi?= =?UTF-8?q?rst?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 23 ++ REUSE.md | 1 + cmd/felhom-agent/main.go | 11 + cmd/felhom-agent/opactions_wiring_test.go | 51 ++++ internal/hub/collect.go | 10 + internal/hub/report.go | 26 ++ internal/localapi/controllersupervisor.go | 67 +++++ internal/localapi/opactions.go | 251 ++++++++++++++++++ internal/localapi/opactions_test.go | 222 ++++++++++++++++ internal/localapi/server.go | 7 + internal/localapi/stalelock_inflight_test.go | 6 +- .../localapi/supervisor_unresponsive_test.go | 74 ++++++ 12 files changed, 747 insertions(+), 2 deletions(-) create mode 100644 cmd/felhom-agent/opactions_wiring_test.go create mode 100644 internal/localapi/opactions.go create mode 100644 internal/localapi/opactions_test.go create mode 100644 internal/localapi/supervisor_unresponsive_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b1c2ff..f83a561 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,26 @@ +## Unreleased (2026-10-29, afternoon) — „Restart controller" from the hub (the agent's own closed list); the unresponsive controller ALARM (R-1344) — commit only; the next agent release (needs the operator's vouch) + +- **Operator actions for the agent (operator brief 2026-10-29, Part 1; felhom.eu `04` §6.1).** The host-report reply may + carry `operator_actions: [{id, action, arg}]` (`hub.ControlEnvelope.OperatorActions`); the agent acts ONCE per id and + answers on the next report, `operator_action_results: [{id, outcome, message}]` (`hub.HostReport`), which it requests + at once through the cap-1 report trigger. CLOSED list: `restart_controller ` only — anything else refused, nothing + called. The act is the supervisor's (`systemctl restart felhom-controller-bootstrap.service` in the guest, existing + grants — no sudoers change, no config bundle) with its guards: provisioned guest, guest running, the guest list and the + vzdump state readable, not parked, no swap, not locked, no vzdump. The thrash brake counts operator restarts with the + supervisor's: 3 in 15 minutes → the press is refused, and presses stay refused for 30 minutes. The result says how long + the controller took to answer (Docker's health check: healthy), or `failed` when it did not within 3 minutes. One + `audit: operator action` line per decision. Works with the controller stopped. Tests `TestOpActions_*` (closed list, + unknown/bad arg refused, restart + answer time + report trigger + once per id + re-send until dropped, stopped + controller, every guard, the brake across supervisor and operator restarts, no healthy answer = failed) and + `TestMainWiresTheAgentOperatorActions` (AST) — each seen failing by a mutation. +- **R-1344 — a controller that RUNS but does not answer: ALARM ONLY.** Each sweep also reads Docker's health status for a + running controller (the swap's `docker inspect -f` grant); 6 consecutive „unhealthy" sweeps (3 minutes, on top of + Docker's own ~90 s) start an episode: one ERROR line, the stanza's `unresponsive` / `unresponsive_since` (the hub raises + `controller_unresponsive`, operator-only). NOT restarted for it: the brief allows the restart only after a measurement + shows a busy night never reads like this, and that measurement runs through tonight (felhom.eu R-1344). Not counted + during a swap, a parked or locked guest, or a vzdump; healthy / starting / no health check / an unreadable status reset + the count. Tests `TestSupervisor_Unresponsive*` (red by mutation: the vzdump guard, the threshold). + ## v0.168.0 (2026-10-29) — R-528: the host's per-container OOM counter (`GET /guest/oom-kills`); R-1254: a lost or late escrow answer is asked again before it is worded — RELEASED (release-agent.sh from a clean clone of `91bd96c`; agent sha256 1e70196c51346cf10e3f04f51975de5899b5dc6baa5a45ec1d17bc248830d2ca, bundle 3415c034955d46c59ca3f7b284a648abf852b1257b77cc49bd6ea9994a7b8cb8; no configs/ change since v0.167.0 → binary-only signed updates to demo-felhom, Tester 1, demo-hp) — **VOUCHED by the operator 2026-10-29 10:21:56 CET**; Day-0 pin set (D50) ### R-528 diff --git a/REUSE.md b/REUSE.md index d33180a..759a17a 100644 --- a/REUSE.md +++ b/REUSE.md @@ -89,6 +89,7 @@ | `Server.RecoverStaleLockedGuests` | internal/localapi/stalelock.go | `RecoverStaleLockedGuests(ctx)` | startup stale vzdump-lock heal (F2-b) | Clears ONLY `backup`/`snapshot-delete`, only when no vzdump in-flight; A1 RESOLVED (v0.62.0): scan is pool-intersected (`ListLXC` ∩ `Client.Pool`), fail-safe skip on pool-read failure | | `ControllerSwapper.Swap` + `ValidControllerImage` | internal/localapi/controllerswap.go | `Swap(ctx, vmid, target) *ControllerSwapState` | agent-owned controller image swap + rollback | Strict image regex (repo + 3-part semver); state file written BEFORE swap; no-healthcheck images need `verifyDwell` | | `Server.ControllerSupervisorTick` + `ControllerParkedMarker` | internal/localapi/controllersupervisor.go | `ControllerSupervisorTick(ctx)` | R-523: restart a provisioned guest's not-running controller via its bootstrap unit | Two-sweep confirm; honours swapInFlight, the host-side park marker, guest lock + vzdump; 3 restarts/15 min → 30 min pause; record rides the report as `controller_supervisor` (the hub mints the events — the agent has no event channel) | +| `Server.OnEnvelope` / `OperatorActionResults` (operator actions) | internal/localapi/opactions.go | `OnEnvelope(ctx, env)`, `OperatorActionResults() []hub.OperatorActionResult` | 2026-10-29: the operator's actions for THIS agent from the host-report reply — a CLOSED list (`restart_controller` only, `TestOpActions_AgentClosedList`), once per id, the result re-sent until the reply stops listing it, the report requested at once (`SetOperatorActionTrigger`). The restart is the supervisor's act with its guards; the 3-in-15-min brake counts operator restarts too (30-min refusal, `opBrakeUntil`). A new verb is an operator decision AND a hub `store.hostOperatorActionNames` entry in the same release | | `MemoryOps` + `Server.readMemoryBounds` | internal/localapi/guestmemory.go | `readMemoryBounds(ctx, vmid) (memoryBounds, err)` | guest RAM resize (v0.90.0, R-24): GET/POST /guest/memory | NEW narrow seam (never extend `GuestAPI` — it breaks every fake); the AGENT is the boundary — bounds recomputed FRESH per request (min 2048 / max host_total−2048 / shrink floor max(2048, usage+512)); §8 UNITS TRAP (config `memory`=MB, status/node=bytes); verify maxmem==target after `SetConfig` before claiming success; SetConfig NEVER called on a refusal path | ### Proxmox client / hub / PBS / provisioning diff --git a/cmd/felhom-agent/main.go b/cmd/felhom-agent/main.go index 3c9e0cf..d5254be 100644 --- a/cmd/felhom-agent/main.go +++ b/cmd/felhom-agent/main.go @@ -1267,6 +1267,17 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int } localSrv := buildLocalAPIServer(cfg, px, backupStore, heavyOps, observer, driveKnown, hostOps, gate, collector, client, intentRec, guestBindStore, formatJobStore, logRing, escrowCeremonyCfg, logger, &localTokens) if localSrv != nil { + // 2026-10-29 (operator brief, Part 1): the operator's actions for THIS agent ride the host report's reply — the + // closed list (localapi/opactions.go). The server joins the observer chain; its results ride the next report, + // which it requests at once through the same cap-1 trigger the storage watchdog uses. + localSrv.SetOperatorActionTrigger(func() { + select { + case storageTrigger <- struct{}{}: + default: + } + }) + collector.SetOperatorActionResults(localSrv.OperatorActionResults) + loop.SetEnvelopeObserver(hub.MultiObserver(desiredSyncer, jobsRunner, localSrv)) backupCapRefusal = localSrv.BackupCapRefusal // R-973 internalPlaceExec.SetEnsure(localSrv.EnsureInternalPlace) // R-998 (doc 14): POST /host/poweroff — guest shutdown, then host power-off; guards in hostpoweroff.go. diff --git a/cmd/felhom-agent/opactions_wiring_test.go b/cmd/felhom-agent/opactions_wiring_test.go new file mode 100644 index 0000000..2335bbd --- /dev/null +++ b/cmd/felhom-agent/opactions_wiring_test.go @@ -0,0 +1,51 @@ +package main + +import ( + "go/ast" + "go/parser" + "go/token" + "testing" +) + +// 2026-10-29 (operator brief, Part 1): the agent's operator actions are unit-green in localapi; main.go must also put +// the server in the report loop's observer chain, give the collector its results and give the server the report +// trigger — an inert seam otherwise (the „seam built but never wired" class). Asserted from the AST. +// RED-PROOF: seen failing before the wiring existed (all three false). +func TestMainWiresTheAgentOperatorActions(t *testing.T) { + fset := token.NewFileSet() + f, err := parser.ParseFile(fset, "main.go", nil, 0) + if err != nil { + t.Fatalf("parse main.go: %v", err) + } + var observed, results, trigger bool + ast.Inspect(f, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + fn, ok := call.Fun.(*ast.SelectorExpr) + if !ok { + return true + } + switch fn.Sel.Name { + case "MultiObserver": + for _, a := range call.Args { + if id, ok := a.(*ast.Ident); ok && id.Name == "localSrv" { + observed = true + } + } + case "SetOperatorActionResults": + if len(call.Args) == 1 { + if sel, ok := call.Args[0].(*ast.SelectorExpr); ok && sel.Sel.Name == "OperatorActionResults" { + results = true + } + } + case "SetOperatorActionTrigger": + trigger = true + } + return true + }) + if !observed || !results || !trigger { + t.Fatalf("operator actions wiring: observer=%v results=%v trigger=%v — all three must be wired in main.go", observed, results, trigger) + } +} diff --git a/internal/hub/collect.go b/internal/hub/collect.go index 325c2ba..69286ca 100644 --- a/internal/hub/collect.go +++ b/internal/hub/collect.go @@ -128,6 +128,7 @@ type Collector struct { wg WireguardReporter // S3: offsite-tunnel status (nil → stanza omitted) pbsdr PBSDRReporter // slice 2: PBS DR tier bridge state (nil → stanza omitted) ctrlSup ControllerSupervisorReporter // R-523: in-guest controller supervisor (nil → stanza omitted) + opResults func() []OperatorActionResult // 2026-10-29: the operator actions' answers (nil → none) guestNet GuestNetReporter // R-54: per-guest network watchdog (nil → stanza omitted) diskTrim GuestDiskTrimReporter // R-444: weekly guest disk trim (nil → stanza omitted) lineTest LineTestReporter // R-1247: weekly upload line test (nil → stanza omitted) @@ -235,6 +236,12 @@ type ControllerSupervisorReporter interface { ControllerSupervisorStatus(ctx context.Context) *ControllerSupervisorStatus } +// SetOperatorActionResults wires the source of the operator actions' answers (2026-10-29; nil → none sent). +func (c *Collector) SetOperatorActionResults(f func() []OperatorActionResult) *Collector { + c.opResults = f + return c +} + // SetControllerSupervisorReporter wires the R-523 controller supervisor as a report source (nil-safe). func (c *Collector) SetControllerSupervisorReporter(r ControllerSupervisorReporter) *Collector { c.ctrlSup = r @@ -432,6 +439,9 @@ func (c *Collector) Collect(ctx context.Context) (*HostReport, error) { if c.ctrlSup != nil { report.ControllerSupervisor = c.ctrlSup.ControllerSupervisorStatus(ctx) } + if c.opResults != nil { + report.OperatorActionResults = c.opResults() + } // R-54: guest-network watchdog state (nil reporter = feature not wired → stanza omitted). if c.guestNet != nil { report.GuestNet = c.guestNet.GuestNetStatus(ctx) diff --git a/internal/hub/report.go b/internal/hub/report.go index 4344ca0..58df779 100644 --- a/internal/hub/report.go +++ b/internal/hub/report.go @@ -172,6 +172,24 @@ type HostReport struct { // byte-stable. The hub parser is pinned by hub/internal/monitor/controller_supervisor_test.go // against the JSON TestControllerSupervisorStanza_WireShape pins here. ControllerSupervisor *ControllerSupervisorStatus `json:"controller_supervisor,omitempty"` + // OperatorActionResults (2026-10-29, the operator brief's Part 1): the answers to the operator actions the hub's + // replies listed — the controller's shape (`04` §6.1). Re-sent until the reply stops listing the id. omitempty: absent + // when there is nothing to answer, so the golden stays byte-stable. + OperatorActionResults []OperatorActionResult `json:"operator_action_results,omitempty"` +} + +// OperatorAction is ONE entry of the reply's operator_actions (hub store.OperatorActionDirective, field-exact). +type OperatorAction struct { + ID int64 `json:"id"` + Action string `json:"action"` + Arg string `json:"arg,omitempty"` +} + +// OperatorActionResult is one answer (hub api.hostOperatorActionResult, field-exact): outcome done | refused | failed. +type OperatorActionResult struct { + ID int64 `json:"id"` + Outcome string `json:"outcome"` + Message string `json:"message"` } // ControllerSupervisorStatus is the R-523 stanza. Carries no secret. @@ -195,6 +213,11 @@ type ControllerSupervisorGuest struct { Restarts24h int `json:"restarts_24h"` SlowCrashloop bool `json:"slow_crashloop"` SlowCrashloopSince string `json:"slow_crashloop_since,omitempty"` // RFC3339 + // R-1344 (2026-10-29) — the controller RUNS but Docker's health check has read „unhealthy" on several consecutive + // sweeps. ALARM ONLY: the agent does not restart it for this (the measurement must first show a busy night never + // looks like it). UnresponsiveSince is the episode's start; the hub keys on it MOVING (controller_unresponsive). + Unresponsive bool `json:"unresponsive"` + UnresponsiveSince string `json:"unresponsive_since,omitempty"` // RFC3339 } // PBSDRStatus is the per-heartbeat PBS-DR-tier bridge state (slice 2). States: @@ -631,6 +654,9 @@ type ControlEnvelope struct { // NEXT heartbeat carries it in log_tail (the report-channel log_tail_requests mirror). // Absent/false on an old hub → nothing happens. LogTailRequested bool `json:"log_tail_requested"` + // OperatorActions (2026-10-29): the operator's pending actions for THIS host's agent — a CLOSED list on both sides + // (`04` §6.1). Absent on an older hub. + OperatorActions []OperatorAction `json:"operator_actions,omitempty"` } // LogTail is the heartbeat's on-demand agent log tail (see HostReport.LogTail). diff --git a/internal/localapi/controllersupervisor.go b/internal/localapi/controllersupervisor.go index 6649137..d5c52c9 100644 --- a/internal/localapi/controllersupervisor.go +++ b/internal/localapi/controllersupervisor.go @@ -83,6 +83,14 @@ const ( // controllerSlowCounterFile holds the 24-hour restart times and the last raise, per guest, beside // the parked marker. controllerSlowCounterFile = "controller-restarts-24h.json" + + // R-1344 (operator brief 2026-10-29, Part 2) — a controller that RUNS but does not answer. Docker's own HEALTHCHECK + // (`curl -f localhost:8080/api/health`, every 30 s, timeout 5 s) reads „unhealthy" only after 3 failed probes (~90 s); + // controllerUnresponsiveConfirm more sweeps (3 minutes) make it an episode: logged, carried by the stanza, mailed by + // the hub (controller_unresponsive). ALARM ONLY — it is NOT restarted for this until the measurement (R-1344) shows a + // busy night (backups, updates) never reads unhealthy that long. Not counted during a swap, a parked guest, a locked + // guest or a vzdump. + controllerUnresponsiveConfirm = 6 ) // controllerSupState is one guest's supervisor record. In-memory on purpose (the guest-power @@ -103,6 +111,14 @@ type controllerSupState struct { // on the other hand, would hide exactly the box it exists for (one whose agent restarts too). restarts24h []time.Time slowCrashloopSince time.Time // the last raise; kept after it ages out, the hub keys on it MOVING + + // opBrakeUntil (2026-10-29): an operator restart refused by the brake refuses every press until then (30 min). + opBrakeUntil time.Time + + // R-1344: consecutive running-but-unhealthy sweeps, and the current episode (since kept after it ends). + unhealthySeen int + unresponsive bool + unresponsiveSince time.Time } type controllerSupervisor struct { @@ -293,6 +309,7 @@ func (s *Server) superviseOneController(ctx context.Context, vmid int, guestStat if running { st.notRunningSeen = 0 s.ctrlSup.mu.Unlock() + s.observeControllerHealth(ctx, vmid, parked) return true } st.notRunningSeen++ @@ -392,6 +409,52 @@ func (s *Server) superviseOneController(ctx context.Context, vmid int, guestStat return false } +// observeControllerHealth is R-1344's count for a RUNNING controller: Docker's health status, one read per sweep over +// the swap's `docker inspect -f` grant. Anything but „unhealthy" — healthy, starting, no health check, a read that +// fails — resets the count; so do a swap, a parked guest, a locked guest and a vzdump (each is someone else's moment). +func (s *Server) observeControllerHealth(ctx context.Context, vmid int, parked bool) { + s.swapMu.Lock() + swapping := s.swapInFlight[vmid] + s.swapMu.Unlock() + unhealthy := false + if !parked && !swapping { + if out, err := s.guestExec.GuestExec(ctx, vmid, "docker", "inspect", "-f", opHealthTemplate, controllerContainer); err == nil { + f := strings.SplitN(strings.TrimSpace(out), "|", 2) + unhealthy = len(f) == 2 && f[0] == "true" && f[1] == "unhealthy" + } + } + if unhealthy { + if lock, _, err := s.staleLock.Lock(ctx, vmid); err != nil || lock != "" { + unhealthy = false + } else if busy, berr := s.staleLock.BackupRunning(ctx, vmid); berr != nil || busy { + unhealthy = false + } + } + now := s.clock() + s.ctrlSup.mu.Lock() + st := s.supState(vmid) + if !unhealthy { + was := st.unresponsive + st.unhealthySeen, st.unresponsive = 0, false + s.ctrlSup.mu.Unlock() + if was { + s.logger.Info("controller-supervisor: the controller answers its health check again", "vmid", vmid) + } + return + } + st.unhealthySeen++ + seen := st.unhealthySeen + start := seen >= controllerUnresponsiveConfirm && !st.unresponsive + if start { + st.unresponsive, st.unresponsiveSince = true, now + } + s.ctrlSup.mu.Unlock() + if start { + s.logger.Error("controller-supervisor: the controller RUNS but does not answer its health check — raising controller_unresponsive; NOT restarting it (R-1344: alarm only until measured)", + "vmid", vmid, "unhealthy_sweeps", seen, "interval", controllerSupervisorInterval.String()) + } +} + func (s *Server) resetNotRunning(vmid int) { s.ctrlSup.mu.Lock() defer s.ctrlSup.mu.Unlock() @@ -446,6 +509,10 @@ func (s *Server) ControllerSupervisorStatus(_ context.Context) *hub.ControllerSu g.SlowCrashloopSince = st.slowCrashloopSince.UTC().Format(time.RFC3339) g.SlowCrashloop = now.Sub(st.slowCrashloopSince) < controllerSlowCrashloopWindow } + g.Unresponsive = st.unresponsive + if !st.unresponsiveSince.IsZero() { + g.UnresponsiveSince = st.unresponsiveSince.UTC().Format(time.RFC3339) + } out.Guests = append(out.Guests, g) } sort.Slice(out.Guests, func(i, j int) bool { return out.Guests[i].VMID < out.Guests[j].VMID }) diff --git a/internal/localapi/opactions.go b/internal/localapi/opactions.go new file mode 100644 index 0000000..9eba28f --- /dev/null +++ b/internal/localapi/opactions.go @@ -0,0 +1,251 @@ +package localapi + +import ( + "context" + "fmt" + "sort" + "strconv" + "strings" + "sync" + "time" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" +) + +// The operator's actions for THIS agent (operator brief 2026-10-29, Part 1; felhom.eu `04` §6.1). +// +// THE DOOR IS THE HOST REPORT'S REPLY, the controller's shape: the operator presses a button on the hub's host page, the +// hub stores a row and pokes the box (the poke's out-of-cycle report gets the reply in seconds; without the tunnel the +// 15-minute cycle carries it), the reply lists `operator_actions: [{id, action, arg}]`, this agent acts ONCE per id, and +// the next report answers `operator_action_results: [{id, outcome, message}]` — the agent asks for that report at once. +// The result is re-sent until the reply stops listing the id. +// +// THE LIST IS CLOSED: one verb, restart_controller . Anything else — a verb, a vmid that does not parse, a guest +// this agent did not provision — is refused and NOTHING is called. Pinned by TestOpActions_AgentClosedList. +// +// THE ACT is the supervisor's own (controllersupervisor.go): `systemctl restart felhom-controller-bootstrap.service` in +// the guest over the existing grants, with the supervisor's guards — not during a controller swap, not parked, not on a +// stopped, locked or vzdump-busy guest, not when the guest list or the vzdump state cannot be read — and the thrash brake +// counts these restarts too: 3 restarts (the supervisor's or the operator's) inside 15 minutes refuse the press, and +// presses stay refused for 30 minutes. It works with the controller stopped or stuck: the agent does it, not the +// controller. Every decision is one `audit:` line (the customer-visible audit form of reconcile.SlogAudit). + +const ( + // OpRestartController is the agent's one operator verb. + OpRestartController = "restart_controller" + + opOutcomeDone = "done" + opOutcomeRefused = "refused" + opOutcomeFailed = "failed" + + // opBrakeRefusal is how long presses stay refused after the brake refused one. + opBrakeRefusal = 30 * time.Minute + // opForgetAfter drops a started id the reply no longer lists. + opForgetAfter = 24 * time.Hour + + defaultOpHealthPoll = 2 * time.Second + // defaultOpHealthWait covers the bootstrap's pull-less rm/run (seconds), Docker's start period (10 s) and its first + // probe (30 s interval) with room: the controller's own HEALTHCHECK is the answer. + defaultOpHealthWait = 3 * time.Minute + // opHealthTemplate reads running + health over the swap's `docker inspect -f` grant. + opHealthTemplate = "{{.State.Running}}|{{if .State.Health}}{{.State.Health.Status}}{{else}}none{{end}}" +) + +// operatorActionNames is THE agent's closed list. Adding an entry fails TestOpActions_AgentClosedList by design. +var operatorActionNames = []string{OpRestartController} + +// OperatorActionNames returns the closed list (sorted copy). +func OperatorActionNames() []string { + out := append([]string(nil), operatorActionNames...) + sort.Strings(out) + return out +} + +type opActionState struct { + mu sync.Mutex + started map[int64]time.Time + results map[int64]hub.OperatorActionResult +} + +// SetOperatorActionTrigger wires the report loop's out-of-band trigger (the result goes at once, not in 15 minutes). +func (s *Server) SetOperatorActionTrigger(f func()) { s.opTrigger = f } + +// OnEnvelope is the report loop's hook (hub.EnvelopeObserver): it starts each newly listed action once and forgets +// the ids the reply no longer lists (the hub has their results). +func (s *Server) OnEnvelope(_ context.Context, env *hub.ControlEnvelope) { + if env == nil { + return + } + listed := map[int64]bool{} + var start []hub.OperatorAction + now := time.Now() + s.opActs.mu.Lock() + if s.opActs.started == nil { + s.opActs.started = map[int64]time.Time{} + s.opActs.results = map[int64]hub.OperatorActionResult{} + } + for _, a := range env.OperatorActions { + listed[a.ID] = true + if _, ok := s.opActs.started[a.ID]; ok { + continue + } + s.opActs.started[a.ID] = now + start = append(start, a) + } + for id, at := range s.opActs.started { + if listed[id] { + continue + } + if _, answered := s.opActs.results[id]; answered || now.Sub(at) > opForgetAfter { + delete(s.opActs.started, id) + delete(s.opActs.results, id) + } + } + s.opActs.mu.Unlock() + for _, a := range start { + go s.runOperatorAction(a) // a restart waits for the health answer — never block the report loop + } +} + +// OperatorActionResults is the report's answer list, oldest id first. +func (s *Server) OperatorActionResults() []hub.OperatorActionResult { + s.opActs.mu.Lock() + defer s.opActs.mu.Unlock() + out := make([]hub.OperatorActionResult, 0, len(s.opActs.results)) + for _, r := range s.opActs.results { + out = append(out, r) + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out +} + +func (s *Server) runOperatorAction(a hub.OperatorAction) { + ctx, cancel := context.WithTimeout(context.Background(), s.opWait()+time.Minute) + defer cancel() + outcome, msg := opOutcomeRefused, fmt.Sprintf("unknown action %q — the agent's list holds restart_controller only", a.Action) + if a.Action == OpRestartController { + outcome, msg = s.operatorRestartController(ctx, a.ID, a.Arg) + } + s.logger.Warn("audit: operator action", "id", a.ID, "action", a.Action, "arg", a.Arg, "by", "the operator via the hub", + "outcome", outcome, "message", msg) + s.opActs.mu.Lock() + if _, still := s.opActs.started[a.ID]; still { + s.opActs.results[a.ID] = hub.OperatorActionResult{ID: a.ID, Outcome: outcome, Message: msg} + } + s.opActs.mu.Unlock() + if s.opTrigger != nil { + s.opTrigger() + } +} + +func (s *Server) opPoll() time.Duration { + if s.opHealthPoll > 0 { + return s.opHealthPoll + } + return defaultOpHealthPoll +} + +func (s *Server) opWait() time.Duration { + if s.opHealthWait > 0 { + return s.opHealthWait + } + return defaultOpHealthWait +} + +// operatorRestartController is the press: the supervisor's guards, the brake, the restart, then the wait for the +// controller's healthy answer. Returns the outcome and the sentence the host page shows. +func (s *Server) operatorRestartController(ctx context.Context, id int64, arg string) (string, string) { + vmid, err := strconv.Atoi(arg) + if err != nil || strconv.Itoa(vmid) != arg { + return opOutcomeRefused, fmt.Sprintf("refused: %q is not a guest id", arg) + } + if s.staleLock == nil || s.guestExec == nil { + return opOutcomeRefused, "refused: the controller supervisor is not wired on this agent" + } + if !s.provisionedGuest(vmid) { + return opOutcomeRefused, fmt.Sprintf("refused: guest %d is not a guest this agent provisioned", vmid) + } + guests, err := s.staleLock.Guests(ctx) + if err != nil { + return opOutcomeRefused, "refused: the guest list cannot be read — ownership unproven, nothing touched" + } + status := "" + for _, g := range guests { + if g.VMID == vmid { + status = g.Status + } + } + if status != "running" { + return opOutcomeRefused, fmt.Sprintf("refused: guest %d is not running (%s) — the guest-power watchdog owns it", vmid, orDash(status)) + } + if s.controllerParked(vmid) { + return opOutcomeRefused, fmt.Sprintf("refused: guest %d's controller is parked (remove the controller-parked marker first)", vmid) + } + s.swapMu.Lock() + swapping := s.swapInFlight[vmid] + s.swapMu.Unlock() + if swapping { + return opOutcomeRefused, "refused: a controller swap (update) is in flight — it owns the controller" + } + if lock, _, err := s.staleLock.Lock(ctx, vmid); err != nil { + return opOutcomeRefused, "refused: the guest lock cannot be read — nothing touched" + } else if lock != "" { + return opOutcomeRefused, fmt.Sprintf("refused: guest %d is locked (%s) — another operation owns it", vmid, lock) + } + if busy, berr := s.staleLock.BackupRunning(ctx, vmid); berr != nil || busy { + return opOutcomeRefused, "refused: a whole-guest backup (vzdump) may be in flight for the guest" + } + + // The brake: every restart counts — the supervisor's and the operator's. + now := s.clock() + s.ctrlSup.mu.Lock() + st := s.supState(vmid) + if now.Before(st.opBrakeUntil) { + until := st.opBrakeUntil + s.ctrlSup.mu.Unlock() + return opOutcomeRefused, fmt.Sprintf("refused: 3 restarts in 15 minutes — presses are refused until %s UTC", until.UTC().Format("15:04")) + } + st.restarts = pruneBefore(st.restarts, now.Add(-controllerCrashloopWindow)) + if len(st.restarts) >= controllerCrashloopMax { + st.opBrakeUntil = now.Add(opBrakeRefusal) + until := st.opBrakeUntil + s.ctrlSup.mu.Unlock() + return opOutcomeRefused, fmt.Sprintf("refused: 3 restarts in 15 minutes — presses are refused until %s UTC", until.UTC().Format("15:04")) + } + reason := fmt.Sprintf("the operator via the hub (action #%d)", id) + st.restarts = append(st.restarts, now) + st.restartsTotal++ + st.lastRestartAt = now + st.lastReason = reason + st.notRunningSeen = 0 + s.ctrlSup.mu.Unlock() + + s.logger.Warn("controller-supervisor: RESTARTING the controller on the operator's press", "vmid", vmid, "action_id", id, "unit", bootstrapUnit) + began := time.Now() + if _, err := s.guestExec.GuestExec(ctx, vmid, "systemctl", "restart", bootstrapUnit); err != nil { + return opOutcomeFailed, fmt.Sprintf("the restart of guest %d's controller failed: %v", vmid, err) + } + for time.Since(began) < s.opWait() { + out, err := s.guestExec.GuestExec(ctx, vmid, "docker", "inspect", "-f", opHealthTemplate, controllerContainer) + if err == nil { + f := strings.SplitN(strings.TrimSpace(out), "|", 2) + if len(f) == 2 && f[0] == "true" && (f[1] == "healthy" || f[1] == "none") { + secs := int(time.Since(began).Round(time.Second) / time.Second) + return opOutcomeDone, fmt.Sprintf("restarted; the controller answered after %d s (Docker's health check: healthy)", secs) + } + } + select { + case <-ctx.Done(): + return opOutcomeFailed, "restarted, but the wait for its answer was cut" + case <-time.After(s.opPoll()): + } + } + return opOutcomeFailed, fmt.Sprintf("restarted, but no healthy answer within %s — look at the box", s.opWait().Round(time.Second)) +} + +func orDash(s string) string { + if s == "" { + return "not listed" + } + return s +} diff --git a/internal/localapi/opactions_test.go b/internal/localapi/opactions_test.go new file mode 100644 index 0000000..0930a15 --- /dev/null +++ b/internal/localapi/opactions_test.go @@ -0,0 +1,222 @@ +package localapi + +import ( + "context" + "errors" + "os" + "path/filepath" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" +) + +// Operator brief 2026-10-29, Part 1: „Restart controller" from the hub. The agent's closed list holds one verb; the act +// is the supervisor's own restart with the supervisor's guards, and the brake counts operator restarts too. + +// opExec answers the supervisor's fake plus the health read the operator restart waits on. +type opExec struct { + *supExec + health string // what {{.State.Health.Status}} reads after the restart +} + +func (f *opExec) GuestExec(ctx context.Context, vmid int, args ...string) (string, error) { + if len(args) >= 5 && args[0] == "docker" && args[1] == "inspect" && args[3] == opHealthTemplate { + f.mu.Lock() + st := f.status[vmid] + f.mu.Unlock() + if st == "" { + return "", errors.New("pct exec: exit status 1: Error: No such object: felhom-controller") + } + return map[bool]string{true: "true", false: "false"}[st == "running"] + "|" + f.health + "\n", nil + } + return f.supExec.GuestExec(ctx, vmid, args...) +} + +func opServer(t *testing.T, status string) (*Server, *opExec, *fakeGuestPowerCtl, *supClock, string, *atomic.Int32) { + t.Helper() + ex := &opExec{supExec: &supExec{status: map[int]string{9201: status}, onRestart: "running"}, health: "healthy"} + ctl := runningGuest(9201) + s, clk, dir := supServer(t, nil, ctl, 9201) + s.guestExec = ex + s.opHealthPoll = time.Millisecond + s.opHealthWait = 2 * time.Second + var triggered atomic.Int32 + s.SetOperatorActionTrigger(func() { triggered.Add(1) }) + return s, ex, ctl, clk, dir, &triggered +} + +func env(acts ...hub.OperatorAction) *hub.ControlEnvelope { + return &hub.ControlEnvelope{OperatorActions: acts} +} + +// waitResult waits for the action's result (the act runs in the background). +func waitResult(t *testing.T, s *Server, id int64) hub.OperatorActionResult { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + for _, r := range s.OperatorActionResults() { + if r.ID == id { + return r + } + } + time.Sleep(2 * time.Millisecond) + } + t.Fatalf("no result for #%d", id) + return hub.OperatorActionResult{} +} + +// RED-PROOF: seen failing before the list existed (build failure, then the names). +func TestOpActions_AgentClosedList(t *testing.T) { + if got, want := OperatorActionNames(), []string{"restart_controller"}; !reflect.DeepEqual(got, want) { + t.Fatalf("the agent's operator actions = %v, want exactly %v — a new verb needs the operator's word", got, want) + } +} + +// Anything outside the list — and a bad argument — is refused and NOTHING is called. +func TestOpActions_UnknownOrBadArgIsRefused(t *testing.T) { + s, ex, _, _, _, _ := opServer(t, "running") + s.OnEnvelope(context.Background(), env( + hub.OperatorAction{ID: 1, Action: "reboot_guest", Arg: "9201"}, + hub.OperatorAction{ID: 2, Action: "restart_controller", Arg: "abc"}, + hub.OperatorAction{ID: 3, Action: "restart_controller", Arg: "9202"}, // not provisioned here + )) + for _, id := range []int64{1, 2, 3} { + if r := waitResult(t, s, id); r.Outcome != "refused" { + t.Errorf("#%d: %+v, want refused", id, r) + } + } + if n := ex.count(9201); n != 0 { + t.Fatalf("a refused action restarted the controller (%d)", n) + } +} + +// The press restarts ONCE, says how long the controller took to answer, asks for a report at once, and re-sends the +// result until the hub stops listing the id. RED-PROOF: drop the GuestExec restart → restarts 0 → FAILS. +func TestOpActions_RestartReportsTheAnswerTime(t *testing.T) { + s, ex, _, _, _, trig := opServer(t, "running") + a := hub.OperatorAction{ID: 7, Action: "restart_controller", Arg: "9201"} + s.OnEnvelope(context.Background(), env(a)) + r := waitResult(t, s, 7) + if r.Outcome != "done" || !strings.Contains(r.Message, "answered after") { + t.Fatalf("result %+v", r) + } + if n := ex.count(9201); n != 1 { + t.Fatalf("restarts = %d, want 1", n) + } + if trig.Load() == 0 { + t.Fatal("the agent did not ask for a report — the result would wait 15 minutes") + } + s.OnEnvelope(context.Background(), env(a)) // the hub still lists it: re-send, never act again + time.Sleep(100 * time.Millisecond) // a wrong second act runs in the background — give it time to show + if n := ex.count(9201); n != 1 { + t.Fatalf("a re-listed id acted again (%d)", n) + } + if len(s.OperatorActionResults()) != 1 { + t.Fatal("the result must be re-sent while listed") + } + s.OnEnvelope(context.Background(), env()) // the hub recorded it + if left := s.OperatorActionResults(); len(left) != 0 { + t.Fatalf("a recorded result is still sent: %+v", left) + } +} + +// The point of the agent doing it: a STOPPED controller is restarted too (the controller cannot restart itself). +func TestOpActions_WorksWithTheControllerStopped(t *testing.T) { + s, ex, _, _, _, _ := opServer(t, "exited") + s.OnEnvelope(context.Background(), env(hub.OperatorAction{ID: 8, Action: "restart_controller", Arg: "9201"})) + if r := waitResult(t, s, 8); r.Outcome != "done" || ex.count(9201) != 1 { + t.Fatalf("result %+v, restarts %d", r, ex.count(9201)) + } +} + +// Each supervisor guard refuses the press, with the reason. RED-PROOF: drop the lock check → the locked case +// restarts → FAILS (each guard seen failing by its own mutation). +func TestOpActions_GuardsRefuse(t *testing.T) { + cases := []struct { + name, want string + setup func(s *Server, ctl *fakeGuestPowerCtl, dir string) + }{ + {"parked", "parked", func(_ *Server, _ *fakeGuestPowerCtl, dir string) { + if err := os.WriteFile(filepath.Join(dir, "9201", ControllerParkedMarker), nil, 0o600); err != nil { + panic(err) + } + }}, + {"swap", "swap", func(s *Server, _ *fakeGuestPowerCtl, _ string) { s.swapInFlight[9201] = true }}, + {"locked", "locked", func(_ *Server, ctl *fakeGuestPowerCtl, _ string) { ctl.locks[9201] = "backup" }}, + {"vzdump", "backup", func(_ *Server, ctl *fakeGuestPowerCtl, _ string) { ctl.backupRun = map[int]bool{9201: true} }}, + {"vzdump unknown", "backup", func(_ *Server, ctl *fakeGuestPowerCtl, _ string) { ctl.backupErr = errors.New("tasks unreadable") }}, + {"guest stopped", "not running", func(_ *Server, ctl *fakeGuestPowerCtl, _ string) { ctl.guests[0].Status = "stopped" }}, + {"guest list", "guest list", func(_ *Server, ctl *fakeGuestPowerCtl, _ string) { ctl.guestsErr = errors.New("api down") }}, + } + for i, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + s, ex, ctl, _, dir, _ := opServer(t, "running") + tc.setup(s, ctl, dir) + id := int64(100 + i) + s.OnEnvelope(context.Background(), env(hub.OperatorAction{ID: id, Action: "restart_controller", Arg: "9201"})) + r := waitResult(t, s, id) + if r.Outcome != "refused" || !strings.Contains(strings.ToLower(r.Message), tc.want) { + t.Fatalf("result %+v, want refused naming %q", r, tc.want) + } + if n := ex.count(9201); n != 0 { + t.Fatalf("guard %q did not hold: restarted %d", tc.name, n) + } + }) + } +} + +// The thrash brake counts operator restarts AND the supervisor's: 3 in 15 minutes → the next press is refused, and +// presses stay refused for 30 minutes. RED-PROOF: drop the brake → the 4th press restarts → FAILS. +func TestOpActions_TheBrakeCountsEveryRestart(t *testing.T) { + s, ex, _, clk, _, _ := opServer(t, "running") + // one restart by the supervisor itself + ex.mu.Lock() + ex.status[9201] = "exited" + ex.mu.Unlock() + s.ControllerSupervisorTick(context.Background()) + s.ControllerSupervisorTick(context.Background()) + if ex.count(9201) != 1 { + t.Fatalf("setup: the supervisor did not restart (%d)", ex.count(9201)) + } + for id := int64(1); id <= 2; id++ { + clk.t = clk.t.Add(time.Minute) + s.OnEnvelope(context.Background(), env(hub.OperatorAction{ID: id, Action: "restart_controller", Arg: "9201"})) + if r := waitResult(t, s, id); r.Outcome != "done" { + t.Fatalf("press %d: %+v", id, r) + } + } + clk.t = clk.t.Add(time.Minute) + s.OnEnvelope(context.Background(), env(hub.OperatorAction{ID: 3, Action: "restart_controller", Arg: "9201"})) + r := waitResult(t, s, 3) + if r.Outcome != "refused" || !strings.Contains(r.Message, "3 restarts in 15 minutes") { + t.Fatalf("the 4th restart in 15 minutes: %+v", r) + } + if n := ex.count(9201); n != 3 { + t.Fatalf("restarts = %d, want 3", n) + } + clk.t = clk.t.Add(20 * time.Minute) // the window has passed, but the 30-minute refusal has not + s.OnEnvelope(context.Background(), env(hub.OperatorAction{ID: 4, Action: "restart_controller", Arg: "9201"})) + if r := waitResult(t, s, 4); r.Outcome != "refused" { + t.Fatalf("inside the 30-minute refusal: %+v", r) + } + clk.t = clk.t.Add(11 * time.Minute) + s.OnEnvelope(context.Background(), env(hub.OperatorAction{ID: 5, Action: "restart_controller", Arg: "9201"})) + if r := waitResult(t, s, 5); r.Outcome != "done" { + t.Fatalf("after the refusal: %+v", r) + } +} + +// A controller that does not come back healthy is a FAILED press, said plainly. +func TestOpActions_NoHealthyAnswerIsFailed(t *testing.T) { + s, ex, _, _, _, _ := opServer(t, "running") + ex.health = "unhealthy" + s.opHealthWait = 30 * time.Millisecond + s.OnEnvelope(context.Background(), env(hub.OperatorAction{ID: 9, Action: "restart_controller", Arg: "9201"})) + if r := waitResult(t, s, 9); r.Outcome != "failed" || !strings.Contains(r.Message, "no healthy answer") { + t.Fatalf("result %+v", r) + } +} diff --git a/internal/localapi/server.go b/internal/localapi/server.go index d2421be..bf3fcb3 100644 --- a/internal/localapi/server.go +++ b/internal/localapi/server.go @@ -411,6 +411,13 @@ type Server struct { guestsDir string ctrlSup controllerSupervisor + // 2026-10-29: the operator's actions from the hub's reply (opactions.go). opHealthPoll/opHealthWait bound the wait + // for a restarted controller's healthy answer (tests shorten them; zero → the defaults). + opActs opActionState + opTrigger func() + opHealthPoll time.Duration + opHealthWait time.Duration + // Network-storage verify job (SPIKE-nas-verify): the IN-MEMORY single slot + the seams the // detached pipeline runs through (tests inject; production defaults set in NewServer). netVerifyMu sync.Mutex diff --git a/internal/localapi/stalelock_inflight_test.go b/internal/localapi/stalelock_inflight_test.go index 54111c8..9e4ef2d 100644 --- a/internal/localapi/stalelock_inflight_test.go +++ b/internal/localapi/stalelock_inflight_test.go @@ -256,8 +256,10 @@ func TestBackupRunning_AllCallersUseTheOneHelper(t *testing.T) { } } } - if callers != 5 { - t.Errorf("expected the five known callers (stale-lock, guest-power, power-off, supervisor, the R-1006 cleanup), found %d", callers) + // 2026-10-29: +2 deliberate callers — the operator's restart_controller (opactions.go) and R-1344's unresponsive count + // (controllersupervisor.go observeControllerHealth). Both through s.staleLock, the one helper. + if callers != 7 { + t.Errorf("expected the seven known callers (stale-lock, guest-power, power-off, supervisor, the R-1006 cleanup, the operator restart, the unresponsive count), found %d", callers) } b, _ := os.ReadFile("stalelock.go") if !strings.Contains(string(b), "vzdumpInFlight(tasks, vmid, proc)") { diff --git a/internal/localapi/supervisor_unresponsive_test.go b/internal/localapi/supervisor_unresponsive_test.go new file mode 100644 index 0000000..84cae17 --- /dev/null +++ b/internal/localapi/supervisor_unresponsive_test.go @@ -0,0 +1,74 @@ +package localapi + +import ( + "context" + "testing" +) + +// R-1344 (operator brief 2026-10-29, Part 2): a controller that RUNS but whose Docker health check reads „unhealthy" on +// controllerUnresponsiveConfirm consecutive sweeps is reported unresponsive (the stanza; the hub raises +// controller_unresponsive). ALARM ONLY: it is never restarted for it until the measurement licenses that. +// RED-PROOF: seen failing before the stanza carried it. + +func unrespServer(t *testing.T, health string) (*Server, *opExec, *fakeGuestPowerCtl) { + t.Helper() + s, ex, ctl, _, _, _ := opServer(t, "running") + ex.health = health + return s, ex, ctl +} + +func sweep(s *Server, n int) { + for i := 0; i < n; i++ { + s.ControllerSupervisorTick(context.Background()) + } +} + +func TestSupervisor_UnresponsiveIsAlarmedNeverRestarted(t *testing.T) { + s, ex, _ := unrespServer(t, "unhealthy") + sweep(s, controllerUnresponsiveConfirm-1) + if g := s.ControllerSupervisorStatus(context.Background()).Guests; len(g) != 1 || g[0].Unresponsive { + t.Fatalf("unresponsive before %d sweeps: %+v", controllerUnresponsiveConfirm, g) + } + sweep(s, 1) + g := s.ControllerSupervisorStatus(context.Background()).Guests + if len(g) != 1 || !g[0].Unresponsive || g[0].UnresponsiveSince == "" { + t.Fatalf("after %d unhealthy sweeps: %+v", controllerUnresponsiveConfirm, g) + } + since := g[0].UnresponsiveSince + sweep(s, 10) + if n := ex.count(9201); n != 0 { + t.Fatalf("an unresponsive controller was RESTARTED (%d) — the restart waits for the measurement", n) + } + if g := s.ControllerSupervisorStatus(context.Background()).Guests; g[0].UnresponsiveSince != since { + t.Fatal("unresponsive_since moved while it stayed unresponsive — the hub would mail again") + } + // it answers again + ex.health = "healthy" + sweep(s, 1) + if g := s.ControllerSupervisorStatus(context.Background()).Guests; g[0].Unresponsive || g[0].UnresponsiveSince != since { + t.Fatalf("after it answered: %+v (unresponsive must clear; since stays as the record)", g[0]) + } +} + +// Slow but answering (healthy / starting), a swap, a vzdump or a parked guest never counts. +func TestSupervisor_UnresponsiveGuards(t *testing.T) { + for _, tc := range []struct { + name string + setup func(s *Server, ex *opExec, ctl *fakeGuestPowerCtl) + }{ + {"healthy", func(_ *Server, ex *opExec, _ *fakeGuestPowerCtl) { ex.health = "healthy" }}, + {"starting", func(_ *Server, ex *opExec, _ *fakeGuestPowerCtl) { ex.health = "starting" }}, + {"swap", func(s *Server, _ *opExec, _ *fakeGuestPowerCtl) { s.swapInFlight[9201] = true }}, + {"vzdump", func(_ *Server, _ *opExec, ctl *fakeGuestPowerCtl) { ctl.backupRun = map[int]bool{9201: true} }}, + {"locked", func(_ *Server, _ *opExec, ctl *fakeGuestPowerCtl) { ctl.locks[9201] = "backup" }}, + } { + t.Run(tc.name, func(t *testing.T) { + s, ex, ctl := unrespServer(t, "unhealthy") + tc.setup(s, ex, ctl) + sweep(s, controllerUnresponsiveConfirm+2) + if g := s.ControllerSupervisorStatus(context.Background()).Guests; len(g) == 1 && g[0].Unresponsive { + t.Fatalf("%s counted as unresponsive: %+v", tc.name, g[0]) + } + }) + } +}