diff --git a/CHANGELOG.md b/CHANGELOG.md index c9803a0..1e40639 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,58 @@ All notable changes to **felhom-agent** are recorded here. Update on every code change that gets pushed. +## v0.5.0-rc1 — slice 5 Phase A: storage observe + report + watchdog (read-only, live) (2026-06-09) + +Phase A of the storage slice (doc 03 §7). Read-only and live: the agent now observes every +host storage target, reports it into the host-report's `storage_targets` (previously an empty +stub), and runs a fast-poll watchdog that pushes a disconnect to the hub in seconds. No +host-root writes this phase (mounts/SMART/grow/destructive-gate are Phase B). The hub-owned +desired manifest (class/role/policy/creds) is not served until slice 10, so reconcile against +it is built-but-unfed — this phase ships only the genuinely-useful read-only footprint. + +### Added +- **`internal/storage` package** (new): + - **`StorageTarget` wire contract** (`internal/hub/report.go`) — filled the slice-3 stub: + `name`/`type`/`durable_id`/`state`/`reachable`, usage (`total`/`used`/`avail`/ + `used_fraction`), `content`, `mount_path`/`backing_device`, `class_hint` (rotational HINT + — never authoritative; class is hub-owned), `role` (empty until slice 10), a `thin_pool` + sub-object (lvmthin data fill; metadata fill is Phase B/`lvs`), and a `smart` sub-object + (`UNKNOWN` until Phase B). Cross-repo golden kept byte-identical with `felhom.eu/hub` and + guarded by the bidirectional key-set test (`contract_test.go`). + - **`durable_id` derivation** (`durableid.go`) — deterministic per type (the DR-load-bearing + re-attach key): fs-UUID (usb/local-dir), `server:export` (nfs/cifs), `repo+fingerprint` + (pbs), `vg/pool` (lvmthin); never empty (falls back to a stable store id). + - **`HostReader` seam + `ProcHostReader`** (`hostread.go`) — non-privileged `/proc/mounts`, + `/dev/disk/by-uuid`, `/sys/.../rotational` + `removable` reads. Root-free by construction. + - **`Observer`** (`observe.go`) — builds `[]hub.StorageTarget` from `ListStorage`/`NodeStorage` + joined with host reads; surfaces the lvmthin thin-pool data fill prominently (warns ≥85%). + - **Storage watchdog** (`watchdog.go`) — a third daemon goroutine fast-polling the *known* + target set (a defined Proxmox storage and/or a previously-seen one) for + `attached↔disconnected` transitions; on a transition it triggers an immediate, **debounced** + out-of-band host-report. Only flags a *known* target's change (never a never-attached + device); coalesces flaps within the debounce window (leading + trailing edge). + `CachingKnownTargets` rate-limits the Proxmox-derived known set; `HostLiveness` probes + device/mount presence (local) + a reachability dial (network), all non-privileged. +- **Proxmox `Storage` type** (`internal/proxmox/types.go`) — additive parse-only config fields + (`server`/`export`/`share`/`datastore`/`fingerprint`/`vgname`/`thinpool`) feeding durable_id. +- **Collector `StorageObserver` seam** (`internal/hub/collect.go`) — populates `storage_targets` + via the observer; a nil observer or an observe error degrades to empty (never sinks the + heartbeat). Hub does not import storage (storage imports hub for the wire type). +- **Out-of-band report trigger** (`internal/hub/loop.go`) — `Loop.SetTrigger`: a watchdog + signal runs one extra collect→report immediately without disturbing the regular cadence. +- **`StorageConfig`** (`internal/config`) — watchdog interval / debounce / known-refresh knobs + (all optional; package defaults otherwise). +- **Hub ingest** (`felhom.eu/hub`) — `hostReportPayload` now parses `storage_targets` + (full mirror struct), persists them via `report_json`, counts + warns on disconnected + targets, and has its own half of the bidirectional golden key-set test. + +### Notes +- The daemon still runs cleanly with no removable storage, no signers, and no hub manifest — + the watchdog finds nothing to flag; storage reporting is best-effort. +- `proxmox`/`hub`/`authz`/`reconcile` exported surfaces + their golden/adversarial tests are + intact. No host-root writes, no destructive paths, no SMART this phase (all Phase B). +- Version: **v0.5.0-rc1** at the Phase-A checkpoint; **v0.5.0** when Phase B lands. + ## v0.4.0 — slice 4 Phase B: reversibility gate + signed-op consuming layer (2026-06-08) The security core of slice 4: hub-supplied intent stops being trusted for destructive diff --git a/CLAUDE.md b/CLAUDE.md index de67d46..1254cab 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,7 +15,7 @@ - Module `gitea.dooplex.hu/admin/felhom-agent`; binary `felhom-agent` (`cmd/felhom-agent/`). - **Pure Go stdlib + `golang.org/x/crypto` only** — no web frameworks. - `go.mod` directive **go 1.25.0**; dep `golang.org/x/crypto v0.52.0` (declares go 1.25, will NOT build on Go 1.24). The **build server (192.168.0.180) runs go1.26.0** (upstream Go on PATH, backward-compatible). Build/run the agent there for live tests (same LAN as the demo host). -- Version: `version` var in `cmd/felhom-agent/main.go`, overridable via `-ldflags "-X main.version="`; `--version` flag. **Current: v0.4.0** (slice 4 complete: reconcile engine + reversibility gate). Bump on meaningful changes + add a CHANGELOG entry. +- Version: `version` var in `cmd/felhom-agent/main.go`, overridable via `-ldflags "-X main.version="`; `--version` flag. **Current: v0.5.0-rc1** (slice 5 Phase A: storage observe/report + watchdog, read-only; Phase B = the host-root surface, pending). Bump on meaningful changes + add a CHANGELOG entry. ## Layout @@ -26,6 +26,8 @@ internal/log/ slog setup internal/proxmox/ API-first Client + fenced root-CLI Privileged + UPID WaitTask internal/authz/ operator signed-op verifier (SSHSIG); durable FileNonceStore internal/hub/ daemon: HostReport collector + Bearer client + resilient Loop +internal/reconcile/ reconcile engine + reversibility gate + op journal + crash recovery +internal/storage/ storage-target observer + durable_id + fast-poll watchdog (slice 5) ``` ## Proxmox model (the load-bearing rules) @@ -50,7 +52,8 @@ Built in slices, all on `main`: - **v0.3.2** — slice-4 pre-check: reversible `SetConfig` step added to `--selftest=task`; passed live on guest 9999. Findings: LXC `description` write is **synchronous** (empty UPID — dual-mode modeling confirmed); PVE appends a trailing `\n` to `description` on read (reconcile must normalize). First live `VM.Config.*` exercise. - **v0.4.0-rc1** — slice-4 **Phase A** (structural): `internal/reconcile` — engine, per-guest serializer (§10), desired-state model + `DesiredProvider` seam, normalization layer (`NormDescription` promoted out of main.go), plan/diff engine (benign Start/Stop/SetConfig set), durable op journal + idempotency store. Wired into `runDaemon` sharing the queue. Runs **live but unfed** (EmptyProvider → zero mutations until slice 10). - **v0.4.0** — slice-4 **Phase B** (security core): the benign/destructive **classifier** (provenance + data-bearing, not by verb; scratch/same-txn provenance is agent-internal, never hub-sourced), the **reversibility gate** (destructive → `pending_signature` unless a verified, role-scoped, action-bound operator signature), the **signed-op consuming layer** over `internal/authz` (role-scoping per doc 04 §4, op-to-action binding, idempotency-by-nonce, audit), and the **crash-recovery consumer** (`Recover` over `InFlight()`, resume-or-rollback). The gate fronts the queue's executor (every mutation passes it). **Inert this slice** — no destructive deltas served until slice 10; the destructive path is classified, gated, and adversarially tested but not wired to live execution. `authz` surface untouched. -- **Next: slice 5/6 (storage manifest, backup/restore)** — the slices that fill the host-report's empty storage/backup collections and add the destructive executors the gate already guards. +- **v0.5.0-rc1** — slice-5 **Phase A** (read-only, live): `internal/storage` — the `StorageTarget` wire contract (filled the slice-3 stub), `durable_id` derivation per type, the `Observer` (builds `storage_targets` from `ListStorage`/`NodeStorage` + non-privileged host reads), and the **storage watchdog** (third daemon goroutine; fast-poll → debounced out-of-band report on a known target's attach/disconnect). No host-root writes. Hub ingest extended to accept/persist `storage_targets`; cross-repo golden kept byte-identical. **Phase B (pending)** = the host-root surface: systemd `.mount` units + sudoers allowlist, benign re-mount, SMART, disk-grow executor, destructive-storage gating. +- **Next: slice 5 Phase B (host-root surface), then slice 6 (backup/restore)** — the destructive executors the gate already guards. ## Demo host (for live tests) diff --git a/README.md b/README.md index e27b353..74ca4a6 100644 --- a/README.md +++ b/README.md @@ -15,13 +15,38 @@ Module: `gitea.dooplex.hu/admin/felhom-agent` · binary: `felhom-agent` · Go 1. ## Layout ``` -cmd/felhom-agent/ # entry point + --selftest (wiring only; no daemon loop yet) +cmd/felhom-agent/ # entry point + --selftest modes + the daemon (poll loop + reconcile + watchdog) internal/proxmox/ # the Proxmox interaction layer (API-first + fenced root-CLI) internal/config/ # JSON config + env overrides (secrets never logged) internal/log/ # slog setup +internal/authz/ # operator signed-op verifier (SSHSIG); durable nonce store +internal/hub/ # daemon: host-report collector + Bearer client + resilient poll loop +internal/reconcile/ # reconcile engine + reversibility gate + op journal + crash recovery +internal/storage/ # storage-target observer + durable_id + fast-poll watchdog (slice 5) configs/agent.example.json ``` +## The `storage` package — observe + watchdog (slice 5) + +Read-only this slice (no hub desired-state until slice 10): + +- **Observer** builds the host-report's `storage_targets` from `ListStorage`/`NodeStorage` + joined with non-privileged host reads (`/proc/mounts`, `/dev/disk/by-uuid`, + `/sys/.../rotational`). It reports each target's `durable_id` (the DR-load-bearing + re-attach key: fs-UUID for usb/local-dir, `server:export` for nfs/cifs, + `repo+fingerprint` for pbs, `vg/pool` for lvmthin), state, usage, a rotational + **class hint** (never authoritative — class is hub-owned), and the lvmthin **thin-pool + data fill** (a full pool corrupts every guest on it). SMART is a Phase-B privileged read. +- **Watchdog** is the third daemon goroutine: a fast poll (seconds) over the *known* + target set that detects an `attached↔disconnected` transition and fires a **debounced, + out-of-band** host-report so the hub learns of a USB drop in seconds rather than at the + ~15-minute cycle. It mutates nothing (the benign re-mount-by-UUID response lands in + Phase B). The `HostReader` seam keeps it root-free and unit-testable with no real devices. + +The reported `StorageTarget` shape is a cross-repo contract duplicated in `felhom.eu/hub`; +`internal/hub/testdata/host-report.golden.json` is byte-identical with the hub's copy and a +bidirectional key-set test guards drift. + ## The `proxmox` package — model Two backends, one fixed routing policy (the fence is structural — `Client` never shells out, diff --git a/REPORT.md b/REPORT.md index 2562299..f6294b2 100644 --- a/REPORT.md +++ b/REPORT.md @@ -1,95 +1,81 @@ -# REPORT — Slice 4: reconcile engine + the reversibility gate (v0.4.0) (2026-06-08) +# REPORT — Slice 5 Phase A: storage observe + report + watchdog (v0.5.0-rc1) (2026-06-09) > Overwrite-latest report (most recent significant work only). Cumulative history lives in [CHANGELOG.md](CHANGELOG.md). ## Outcome -**Slice 4 is complete and pushed as `v0.4.0`.** Both phases landed: +**Slice 5 Phase A is complete and pushed as `v0.5.0-rc1`** — the read-only, live half of the +storage slice (doc 03 §7). The agent now observes every host storage target, reports it into +the host-report (the slice-3 `storage_targets` stub is filled), and runs a fast-poll storage +watchdog that pushes a disconnect to the hub out-of-band in seconds. **No host-root writes** +this phase — mounts/SMART/disk-grow/destructive-gate are Phase B, which earns isolated review +of the new privileged surface. This is the hard checkpoint: Phase A awaits validation before +Phase B begins. -- **Phase A** (structural, pushed earlier as `v0.4.0-rc1`): the reconcile engine, the - per-guest serializer (doc 03 §10), the desired-state model + provider seam, the - field-normalization layer, the plan/diff engine, and the durable op journal + - idempotency store. Runs **live but unfed** — `EmptyProvider` → zero mutations until - slice 10 serves desired state. -- **Phase B** (this push, the security core): the benign/destructive **classifier**, - the **reversibility gate**, and the **signed-op consuming layer** over `internal/authz` - — with role-scoping, op-to-action binding, idempotency/journaling, audit, and the - crash-recovery consumer. The gate sits in front of the per-guest queue's executor, so - **every mutation passes it**. +## What landed -The whole module is **race-clean and vet-clean** on the Linux build server; 62 reconcile -tests pass (the adversarial matrix runs against the real `authz.Verifier`). +New package **`internal/storage`**: -## The security model (Phase B) +- **`StorageTarget` wire contract** (`internal/hub/report.go`) — the slice-3 `struct{}` stub + is now the full reported shape: identity (`name`/`type`/`durable_id`), `state` + (`attached`/`disconnected`/`decommissioned`) + `reachable`, usage (`total`/`used`/`avail`/ + `used_fraction`), `content`, `mount_path`/`backing_device`, a `class_hint` (a rotational + HINT — **never** authoritative; class is hub-owned), `role` (empty until slice 10), a + `thin_pool` sub-object (lvmthin DATA fill), and a `smart` sub-object (`UNKNOWN` until Phase B). +- **`durable_id` derivation** — deterministic per type, the DR-load-bearing re-attach key: + fs-UUID (usb/local-dir), `server:export` (nfs/cifs), `repo+fingerprint` (pbs), `vg/pool` + (lvmthin). Never empty — falls back to a stable store id so the hub's re-attach lookup + always has something. +- **`Observer`** — joins `ListStorage` (config) + `NodeStorage` (usage/active) with + non-privileged host reads (`/proc/mounts`, `/dev/disk/by-uuid`, `/sys/.../rotational` + + `removable`) behind a `HostReader` seam. Surfaces the lvmthin **thin-pool data fill** + prominently (a full pool corrupts every guest on it) and warns at ≥85%. +- **Storage watchdog** — the third daemon goroutine (alongside the hub loop + reconcile + engine). Fast-polls (default 8s) the *known* target set for `attached↔disconnected` + transitions and fires a **debounced** (default 30s) out-of-band host-report. Flags only a + *known* target's change (never a never-attached device), coalesces flaps (leading + + trailing edge). `CachingKnownTargets` rate-limits the Proxmox-derived known set; + `HostLiveness` does device/mount-presence (local) + a reachability dial (network). -Hub-supplied intent is no longer trusted for destructive change — **by provenance + -data-bearing-ness, not by verb** (doc 03 §4): +Wiring + supporting changes: -- **Benign** (unsigned): start/stop/restart/create, and destroying a resource the agent - created in the **same journaled transaction** (compensating rollback) or **tagged - scratch**. That scratch/same-txn provenance is **agent-internal, journal-recorded, and - never accepted from the hub** — a compromised hub cannot relabel a data-bearing guest - as scratch to walk the gate. -- **Destructive** (signature required): destroy/overwrite of the only/primary copy of - customer data — **regardless of whether it arrives as a job or a desired-state delta**. - Absent/invalid signature → refused **`pending_signature`**, never executed. +- `proxmox.Storage` gained additive parse-only config fields (server/export/share/datastore/ + fingerprint/vgname/thinpool) — the durable_id sources. The API/root fence is untouched. +- The collector gained a `StorageObserver` seam (hub does **not** import storage); `Loop` + gained `SetTrigger` for the out-of-band report; the daemon runs the watchdog as a third + goroutine; `StorageConfig` exposes the watchdog knobs. +- **Hub** (`felhom.eu/hub`): `hostReportPayload` parses `storage_targets`, persists them via + `report_json`, counts/warns on disconnected targets, and carries its half of the + bidirectional golden key-set test. Both repos' goldens are byte-identical. -The signed-op consuming layer calls `authz.Verifier.Verify` (the locked -namespace→allow-list→crypto→target→time→nonce pipeline, untouched) and then enforces -the slice-4 policy on the `VerifiedOp`: **role-scoping** (recovery key = key-rotation -only; operational key = ordinary destructive + planned rotation, doc 04 §4) and -**op-to-action binding** (the verified op + host + guest + params must name the exact -gated action). Idempotency keys the journal by the op nonce; every decision is audited -(a signal, never the guard). +## Tests -## Inert by design (slice-4 scope) +`go test ./...` is green across both repos. New tests: observer build (incl. lvmthin +thin-pool fill, USB-unplugged→disconnected, Proxmox-error fatal, mount-read-failure +degrade), `durable_id` per-type table, watchdog transitions + debounce coalescing + +never-attached suppression + caching TTL + stale-on-error + HostLiveness mount/network, +collector seam (populate + error-degrade), loop out-of-band trigger, and the cross-repo +golden contract on both sides. -There is **no live destructive execution** this slice: nothing serves destructive deltas -until slice 10, and the guest-destroy/storage-wipe/restore-overwrite executors land in -6/7. So the destructive path is fully **classified, gated, and adversarially tested**, -but `RunSignedJob`'s executor is nil in production — an authorized destructive op is -journaled as authorized-but-not-executed. Reconcile itself only produces the benign -Start/Stop/SetConfig set, all allowed through the gate unsigned. +**`go test -race`** must be run on the build server (192.168.0.180, has cgo/gcc); the local +Windows toolchain lacks gcc. To run after pull: `CGO_ENABLED=1 go test -race ./...`. -## Adversarial proof (each case independently rejected) +## Out of scope (deferred) -Run against the **real** `authz.Verifier` with in-test-minted SSHSIGs (the ~40-line -framing is replicated in reconcile's test binary — production `authz` is untouched and -gains no signing capability; live minting is required because the verifier's clock is -not cross-package injectable): +- **Phase B** (next, after this checkpoint validates): the host-root surface — systemd + `.mount` units + sudoers allowlist behind a `HostOps` seam, benign re-mount-by-UUID, + SMART (SATA + NVMe), the disk-grow executor (`pct resize`, grow-only), and routing + destructive storage ops (detach/wipe/resize-shrink) through the slice-4 gate. Built + + tested but inert live until slice 10 serves destructive deltas. +- Reconcile against the hub manifest (attach new targets, enforce class/role/policy, + `decommissioned`) → slice 10. Backup/restore orchestration → slice 6. -unsigned destructive **job** → pending_signature · unsigned destructive **desired-state -delta** → pending_signature (distrusts hub desired state, not just jobs) · forged / -unknown signer → `ErrUnknownSigner` · expired → `ErrExpired` · **replayed nonce across an -agent restart** (durable `FileNonceStore`) → `ErrReplay` · wrong host → `ErrTarget` · -wrong guest / wrong op / wrong params → binding_mismatch · **recovery key on ordinary -destructive** → role_denied · **hub-supplied "scratch" tag** on a data-bearing guest → -ignored, still destructive → refused · **valid + correct role + correct target + fresh -nonce → accepted**, and a second presentation → `ErrReplay`. +## Validation notes for the reviewer -## The two forward-looking notes - -- **Note 1 (carried in)** — the `InFlight()` **resume-or-rollback** startup consumer - (`Engine.Recover`) landed **together with** the signed-op executor, as required. An op - that crashed after the Proxmox POST but before its terminal record (`OpTaskRunning`, - nonce already consumed) is not covered by idempotency dedupe — only this consumer - resolves it (re-read the task via the new `TaskStatusOnce`, record the real outcome; a - no-task-id op is abandoned fail-safe). Wired into daemon startup and tested. -- **Note 2 (addressed)** — the memory comparison is canonicalized (`desiredMemoryMiB`): - desired and actual compare in the same MiB unit that is then written, so a - non-MiB-aligned `MemoryBytes` converges in one pass rather than re-issuing SetConfig - every cycle. A test proves convergence. Recommendation stands that slice 10 serve - MiB-aligned specs at the source. - -## Verification - -- `go test -race -count=1 ./...` and `go vet ./...` clean on the Linux build server - (go1.26); all tests green locally and there. -- No live Proxmox needed — Phase A is unfed and Phase B's destructive path is inert this - slice. The gate's crypto path is proven end-to-end against the real verifier. - -## Conventions - -Version → **v0.4.0**. CHANGELOG has a per-phase entry (newest on top). No secrets in any -committed file. Pushed to `main`. Per the task, I stop at this checkpoint and await the -validation pass. +- The reachability heuristic for dir storages: a Felhom usb/local-dir target is realized as + its **own** mountpoint, so reachability = it is currently an exact mount + its device node + exists (we deliberately do not fall through to PVE's `active` flag, which reads stale- + attached because the mountpoint directory survives on the root fs after an unplug). Builtin + `local` and network/block targets use the `active` flag. Worth confirming against the demo + host's actual storage set during live validation. +- A live `--selftest=hub` against the demo host will print the populated `storage_targets`. diff --git a/cmd/felhom-agent/main.go b/cmd/felhom-agent/main.go index 7487013..e3b0571 100644 --- a/cmd/felhom-agent/main.go +++ b/cmd/felhom-agent/main.go @@ -25,11 +25,12 @@ import ( applog "gitea.dooplex.hu/admin/felhom-agent/internal/log" "gitea.dooplex.hu/admin/felhom-agent/internal/proxmox" "gitea.dooplex.hu/admin/felhom-agent/internal/reconcile" + "gitea.dooplex.hu/admin/felhom-agent/internal/storage" ) // version is the agent version. Overridable at build time with // -ldflags "-X main.version="; defaults to the in-repo CHANGELOG version. -var version = "0.4.0" +var version = "0.5.0-rc1" func main() { var ( @@ -109,10 +110,34 @@ func runDaemon(cfg config.Config, logger *slog.Logger) int { return 1 } hcfg := cfg.Hub.WithDefaults() - collector := hub.NewCollector(px, hub.SystemctlProber{}, cfg.Hub.HostID, version, logger) + // Storage observer (slice 5): read-only, builds the report's storage_targets from + // Proxmox + non-privileged host reads. Wired into the collector via the StorageObserver + // seam (so hub does not import storage). + hostReader := storage.NewProcHostReader() + observer := storage.NewObserver(px, hostReader, logger) + collector := hub.NewCollector(px, hub.SystemctlProber{}, observer, cfg.Hub.HostID, version, logger) loop := hub.NewLoop(collector, client, time.Duration(hcfg.PollSeconds)*time.Second, logger) interval := time.Duration(hcfg.PollSeconds) * time.Second + // Storage watchdog (slice 5): a third daemon goroutine fast-polling the known target + // set for attached↔disconnected transitions and triggering an immediate, debounced + // out-of-band report. With no removable/network storage it simply finds nothing to flag. + storageTrigger := make(chan struct{}, 1) + loop.SetTrigger(storageTrigger) + watchdog := storage.NewWatchdog(storage.WatchdogOptions{ + Targets: storage.NewCachingKnownTargets(observer, cfg.Storage.KnownRefresh()), + Liveness: storage.NewHostLiveness(hostReader, 0), + Trigger: func() { + select { + case storageTrigger <- struct{}{}: + default: + } + }, + Interval: cfg.Storage.WatchdogInterval(), + Debounce: cfg.Storage.WatchdogDebounce(), + Logger: logger, + }) + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() logger.Info("felhom-agent daemon starting", @@ -167,14 +192,17 @@ func runDaemon(cfg config.Config, logger *slog.Logger) int { // last died BEFORE issuing new mutations. With an empty journal this is a no-op. engine.Recover(ctx) - // Run reconcile and the hub loop concurrently; either returning ends the daemon. - errc := make(chan error, 2) + // Run reconcile, the hub loop, and the storage watchdog concurrently; any one + // returning ends the daemon (then ctx cancellation tears the others down). + errc := make(chan error, 3) go func() { errc <- engine.Run(ctx, interval) }() go func() { errc <- loop.Run(ctx) }() + go func() { errc <- watchdog.Run(ctx) }() err = <-errc - stop() // tear down the sibling on the first exit - <-errc // wait for it + stop() // tear down the siblings on the first exit + <-errc // wait for the second + <-errc // wait for the third if err != nil && err != context.Canceled { logger.Error("daemon: exited with error", "err", err) return 1 @@ -242,7 +270,8 @@ func runSelftestHub(ctx context.Context, cfg config.Config, logger *slog.Logger) fmt.Fprintln(os.Stderr, "selftest: hub client:", err) return 1 } - collector := hub.NewCollector(px, hub.SystemctlProber{}, cfg.Hub.HostID, version, logger) + observer := storage.NewObserver(px, storage.NewProcHostReader(), logger) + collector := hub.NewCollector(px, hub.SystemctlProber{}, observer, cfg.Hub.HostID, version, logger) ctx, cancel := context.WithTimeout(ctx, 60*time.Second) defer cancel() diff --git a/internal/config/config.go b/internal/config/config.go index d7e2d80..8a88dbb 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -17,6 +17,7 @@ import ( "os" "strconv" "strings" + "time" ) // Config is the agent configuration. @@ -25,9 +26,36 @@ type Config struct { Privileged PrivilegedConfig `json:"privileged"` Authz AuthzConfig `json:"authz"` Hub HubConfig `json:"hub"` + Storage StorageConfig `json:"storage"` LogLevel string `json:"log_level"` // debug|info|warn|error (default info) } +// StorageConfig tunes the storage watchdog (slice 5). All optional — zero values fall back +// to the storage package defaults via the accessor methods. The watchdog poll is FAST +// (seconds) to catch a USB drop quickly; the debounce keeps a flapping drive from storming +// the hub; the known-set refresh bounds how often the watchdog re-derives the target set +// from the Proxmox API (liveness is probed every poll regardless). +type StorageConfig struct { + WatchdogIntervalSeconds int `json:"watchdog_interval_seconds"` + WatchdogDebounceSeconds int `json:"watchdog_debounce_seconds"` + KnownRefreshSeconds int `json:"known_refresh_seconds"` +} + +// WatchdogInterval returns the configured poll interval (0 = package default). +func (s StorageConfig) WatchdogInterval() time.Duration { + return time.Duration(s.WatchdogIntervalSeconds) * time.Second +} + +// WatchdogDebounce returns the configured debounce window (0 = package default). +func (s StorageConfig) WatchdogDebounce() time.Duration { + return time.Duration(s.WatchdogDebounceSeconds) * time.Second +} + +// KnownRefresh returns the configured known-set refresh TTL (0 = package default). +func (s StorageConfig) KnownRefresh() time.Duration { + return time.Duration(s.KnownRefreshSeconds) * time.Second +} + // HubConfig configures the outbound hub client + daemon poll loop (internal/hub). // The hub serves a real cert (hub.felhom.eu, cert-manager) — this is standard TLS // (system roots), NOT the Proxmox fingerprint-pinning path. diff --git a/internal/hub/collect.go b/internal/hub/collect.go index 3b2a64e..c768c6d 100644 --- a/internal/hub/collect.go +++ b/internal/hub/collect.go @@ -22,11 +22,21 @@ type proxmoxReader interface { GuestConfig(ctx context.Context, vmid int) (proxmox.GuestConfig, error) } +// StorageObserver is the seam the storage layer (internal/storage) plugs into to fill the +// report's storage_targets. Defined here (consumer-side) so hub does NOT import storage — +// storage imports hub for the wire type, and main.go wires the concrete observer in. Same +// pattern as proxmoxReader / CloudflaredProber. A nil observer (slice-3 behaviour, or a +// host with no storage layer) yields an empty []StorageTarget without error. +type StorageObserver interface { + Observe(ctx context.Context) ([]StorageTarget, error) +} + // Collector builds a HostReport from read-only sources. All deps are behind narrow // interfaces for unit testing. type Collector struct { px proxmoxReader cf CloudflaredProber + storage StorageObserver hostID string agentVersion string logger *slog.Logger @@ -34,14 +44,15 @@ type Collector struct { } // NewCollector builds a collector. hostID echoes config.Hub.HostID; agentVersion is -// the binary version. -func NewCollector(px proxmoxReader, cf CloudflaredProber, hostID, agentVersion string, logger *slog.Logger) *Collector { +// the binary version. storage may be nil (storage_targets emitted empty). +func NewCollector(px proxmoxReader, cf CloudflaredProber, storage StorageObserver, hostID, agentVersion string, logger *slog.Logger) *Collector { if logger == nil { logger = slog.Default() } return &Collector{ px: px, cf: cf, + storage: storage, hostID: hostID, agentVersion: agentVersion, logger: logger, @@ -65,8 +76,9 @@ func (c *Collector) Collect(ctx context.Context) (*HostReport, error) { AgentVersion: c.agentVersion, Host: hostMetrics(c.px.Node(), ns), Guests: c.collectGuests(ctx), - // Defined-but-empty this slice (slices 5/6). Non-nil so they marshal as []. - StorageTargets: []StorageTarget{}, + // storage_targets populated this slice (slice 5) via the observer; the rest stay + // defined-but-empty (slice 6). Non-nil so they marshal as []. + StorageTargets: c.collectStorage(ctx), Backups: []Backup{}, RestoreTests: []RestoreTest{}, PBSSnapshots: []PBSSnapshot{}, @@ -129,6 +141,24 @@ func (c *Collector) collectGuests(ctx context.Context) []Guest { return guests } +// collectStorage builds the storage_targets via the observer. A nil observer (no storage +// layer wired) or an observe error degrades to an empty list — storage detail is +// best-effort and must never sink the heartbeat (host liveness is the priority). +func (c *Collector) collectStorage(ctx context.Context) []StorageTarget { + if c.storage == nil { + return []StorageTarget{} + } + targets, err := c.storage.Observe(ctx) + if err != nil { + c.logger.Warn("hub: storage observe failed; reporting no storage targets", "err", err) + return []StorageTarget{} + } + if targets == nil { + return []StorageTarget{} + } + return targets +} + func (c *Collector) cloudflaredStatus(ctx context.Context) string { if c.cf == nil { return "unknown" diff --git a/internal/hub/collect_test.go b/internal/hub/collect_test.go index d5d29dd..f0be57d 100644 --- a/internal/hub/collect_test.go +++ b/internal/hub/collect_test.go @@ -20,6 +20,41 @@ func newTestNodeStatus() proxmox.NodeStatus { return ns } +// fakeObserver is a StorageObserver returning fixed targets (or an error). +type fakeObserver struct { + targets []StorageTarget + err error +} + +func (f fakeObserver) Observe(context.Context) ([]StorageTarget, error) { return f.targets, f.err } + +func TestCollect_StorageTargetsFromObserver(t *testing.T) { + px := &fakePx{node: "n", ns: newTestNodeStatus()} + obs := fakeObserver{targets: []StorageTarget{ + {Name: "local-lvm", Type: StorageTypeLVMThin, State: StorageStateAttached, Reachable: true}, + }} + c := NewCollector(px, fakeProber{status: "active"}, obs, "h", "0.5.0", quietLogger()) + r, err := c.Collect(context.Background()) + if err != nil { + t.Fatalf("Collect: %v", err) + } + if len(r.StorageTargets) != 1 || r.StorageTargets[0].Name != "local-lvm" { + t.Fatalf("storage targets = %+v", r.StorageTargets) + } +} + +func TestCollect_StorageObserverErrorDegradesToEmpty(t *testing.T) { + px := &fakePx{node: "n", ns: newTestNodeStatus()} + c := NewCollector(px, fakeProber{status: "active"}, fakeObserver{err: errors.New("proxmox down")}, "h", "0.5.0", quietLogger()) + r, err := c.Collect(context.Background()) + if err != nil { + t.Fatalf("a storage observe error must not sink the heartbeat: %v", err) + } + if r.StorageTargets == nil || len(r.StorageTargets) != 0 { + t.Errorf("storage targets must degrade to empty non-nil, got %+v", r.StorageTargets) + } +} + func TestCollect_HostAndGuests(t *testing.T) { px := &fakePx{ node: "demo-felhom", @@ -29,7 +64,7 @@ func TestCollect_HostAndGuests(t *testing.T) { }, cfg: map[int]proxmox.GuestConfig{100: {Cores: 2, Memory: 2048}}, } - c := NewCollector(px, fakeProber{status: "active"}, "demo-host-01", "0.3.0", quietLogger()) + c := NewCollector(px, fakeProber{status: "active"}, nil, "demo-host-01", "0.3.0", quietLogger()) r, err := c.Collect(context.Background()) if err != nil { t.Fatalf("Collect: %v", err) @@ -69,7 +104,7 @@ func TestCollect_GuestConfigFailureKeepsStatusOmitsSpec(t *testing.T) { cfg: map[int]proxmox.GuestConfig{100: {Cores: 2}}, cfgErr: map[int]error{200: errors.New("config read failed")}, } - c := NewCollector(px, fakeProber{status: "active"}, "h", "0.3.1", quietLogger()) + c := NewCollector(px, fakeProber{status: "active"}, nil, "h", "0.3.1", quietLogger()) r, err := c.Collect(context.Background()) if err != nil { t.Fatalf("a per-guest failure must NOT fail the whole report: %v", err) @@ -90,7 +125,7 @@ func TestCollect_GuestConfigFailureKeepsStatusOmitsSpec(t *testing.T) { func TestCollect_NodeStatusFailureIsHardError(t *testing.T) { px := &fakePx{node: "n", nsErr: errors.New("proxmox down")} - c := NewCollector(px, fakeProber{status: "active"}, "h", "0.3.0", quietLogger()) + c := NewCollector(px, fakeProber{status: "active"}, nil, "h", "0.3.0", quietLogger()) if _, err := c.Collect(context.Background()); err == nil { t.Fatal("NodeStatus failure must be a hard error (no useful report)") } @@ -98,7 +133,7 @@ func TestCollect_NodeStatusFailureIsHardError(t *testing.T) { func TestCollect_CloudflaredProbeErrorIsUnknown(t *testing.T) { px := &fakePx{node: "n", ns: newTestNodeStatus()} - c := NewCollector(px, fakeProber{err: errors.New("no systemctl")}, "h", "0.3.0", quietLogger()) + c := NewCollector(px, fakeProber{err: errors.New("no systemctl")}, nil, "h", "0.3.0", quietLogger()) r, err := c.Collect(context.Background()) if err != nil { t.Fatalf("cloudflared failure must not be fatal: %v", err) diff --git a/internal/hub/contract_test.go b/internal/hub/contract_test.go index 754966a..bde5020 100644 --- a/internal/hub/contract_test.go +++ b/internal/hub/contract_test.go @@ -24,7 +24,8 @@ func TestHostReport_ContractMatchesGolden(t *testing.T) { t.Fatalf("golden is not valid JSON: %v", err) } - // A constructed report mirroring the golden's populated shape (guests[0] has spec). + // A constructed report mirroring the golden's populated shape: guests[0] has spec, + // storage_targets[0] is an lvmthin (so its thin_pool + smart sub-objects are exercised). report := &HostReport{ HostID: "demo-host-01", ReportedAt: "2026-06-08T12:00:00Z", AgentVersion: "0.3.1", Host: HostMetrics{Node: "demo-felhom", LoadAvg: []string{"0.10"}}, @@ -32,9 +33,22 @@ func TestHostReport_ContractMatchesGolden(t *testing.T) { {VMID: 100, Name: "a", Status: "running", ControllerVersion: "", Spec: &GuestSpec{Cores: 2}}, {VMID: 101, Name: "b", Status: "stopped", ControllerVersion: ""}, }, - StorageTargets: []StorageTarget{}, Backups: []Backup{}, RestoreTests: []RestoreTest{}, + StorageTargets: []StorageTarget{ + { + Name: "local-lvm", Type: StorageTypeLVMThin, DurableID: "pve/data", + State: StorageStateAttached, Reachable: true, ClassHint: "fast", + ThinPool: &ThinPoolFill{DataUsedFraction: 0.42}, + Smart: SmartSummary{Health: SmartUnknown}, + }, + { + Name: "usb-backup", Type: StorageTypeUSB, DurableID: "uuid:x", + State: StorageStateAttached, Reachable: true, + Smart: SmartSummary{Health: SmartUnknown}, + }, + }, + Backups: []Backup{}, RestoreTests: []RestoreTest{}, PBSSnapshots: []PBSSnapshot{}, AuditTail: []AuditEntry{}, - Cloudflared: Cloudflared{Status: "active"}, + Cloudflared: Cloudflared{Status: "active"}, } b, _ := json.Marshal(report) var got map[string]any @@ -44,6 +58,23 @@ func TestHostReport_ContractMatchesGolden(t *testing.T) { assertSameKeys(t, "host", golden["host"], got["host"]) assertSameKeys(t, "guests[0]", firstElem(golden["guests"]), firstElem(got["guests"])) + + // storage_targets[0] is the slice-5 addition — assert its full key set and both + // sub-objects (smart always present; thin_pool present for the lvmthin element). + gst := firstElem(golden["storage_targets"]) + sst := firstElem(got["storage_targets"]) + assertSameKeys(t, "storage_targets[0]", gst, sst) + assertSameKeys(t, "storage_targets[0].smart", field(gst, "smart"), field(sst, "smart")) + assertSameKeys(t, "storage_targets[0].thin_pool", field(gst, "thin_pool"), field(sst, "thin_pool")) +} + +// field extracts a nested object value from a decoded JSON map (nil if absent/not a map). +func field(v any, key string) any { + m, ok := v.(map[string]any) + if !ok { + return nil + } + return m[key] } func firstElem(v any) any { diff --git a/internal/hub/loop.go b/internal/hub/loop.go index 8592df9..1051e61 100644 --- a/internal/hub/loop.go +++ b/internal/hub/loop.go @@ -30,6 +30,7 @@ type Loop struct { client reporter interval time.Duration logger *slog.Logger + trigger <-chan struct{} // optional: an out-of-band report request (storage watchdog) } // NewLoop builds the loop. interval is the starting cadence (the hub may override it @@ -41,6 +42,12 @@ func NewLoop(collector collectorIface, client reporter, interval time.Duration, return &Loop{collector: collector, client: client, interval: interval, logger: logger} } +// SetTrigger wires an out-of-band report channel. A receive on it runs one extra +// collect→report cycle immediately WITHOUT disturbing the regular ticker cadence — used by +// the storage watchdog to push a disconnect to the hub in seconds. The watchdog debounces, +// so this fires at most once per debounce window. +func (l *Loop) SetTrigger(ch <-chan struct{}) { l.trigger = ch } + // Run reports immediately, then on each tick, until ctx is cancelled (then nil). func (l *Loop) Run(ctx context.Context) error { interval := l.interval @@ -60,6 +67,17 @@ func (l *Loop) Run(ctx context.Context) error { interval = next ticker.Reset(interval) } + case <-l.trigger: + // Out-of-band report (storage watchdog). Run a cycle now; keep the regular + // cadence (do not reset the ticker). The envelope's interval is still adopted + // if it changed, mirroring the normal path. + l.logger.Info("hub: out-of-band report triggered (storage watchdog)") + next := l.cycle(ctx, interval) + if next != interval { + l.logger.Info("hub: poll interval changed", "from", interval, "to", next) + interval = next + ticker.Reset(interval) + } } } } diff --git a/internal/hub/loop_test.go b/internal/hub/loop_test.go index 49e1467..d78ee57 100644 --- a/internal/hub/loop_test.go +++ b/internal/hub/loop_test.go @@ -122,6 +122,33 @@ func TestLoop_RunImmediateAndResilientAfterError(t *testing.T) { } } +func TestLoop_OutOfBandTriggerReportsImmediately(t *testing.T) { + // The storage watchdog's trigger channel runs an extra report between ticks (the slow + // cadence is 1h here, so any report within the window comes from the trigger). + var cn, rn int32 + loop := NewLoop( + &fakeCollector{report: &HostReport{}, n: &cn}, + &fakeReporter{env: &ControlEnvelope{}, n: &rn}, + time.Hour, quietLogger()) + trigger := make(chan struct{}, 1) + loop.SetTrigger(trigger) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- loop.Run(ctx) }() + + // Immediate report fires first (1 collect). Then fire the trigger → one more report. + time.Sleep(20 * time.Millisecond) + trigger <- struct{}{} + time.Sleep(20 * time.Millisecond) + cancel() + <-done + + if got := atomic.LoadInt32(&cn); got < 2 { + t.Errorf("collect calls = %d, want ≥2 (immediate + out-of-band trigger)", got) + } +} + func TestLoop_RunAdoptsSlowerInterval(t *testing.T) { var cn, rn int32 loop := NewLoop( diff --git a/internal/hub/report.go b/internal/hub/report.go index f3deb70..28acd57 100644 --- a/internal/hub/report.go +++ b/internal/hub/report.go @@ -63,10 +63,109 @@ type Cloudflared struct { } // The following element types are declared now so the empty collections above are -// typed and slices 5/6 only fill them. No wire fields are committed yet. +// typed and slices 5/6 only fill them. -type StorageTarget struct{} // slice 5: storage manifest target fields TBD -type Backup struct{} // slice 6: per-target backup status fields TBD +// StorageTarget is one observed host storage target (doc 03 §7). It is the REPORTED +// shape — what the agent observes and tells the hub. The hub holds the AUTHORITATIVE +// manifest (desired class/role/policy/creds, slice 10); the agent reports what it sees +// and reconciles toward the hub's manifest. So `class_hint` is a rotational HINT, never +// authoritative class, and `role` is set only when derivable from an existing Proxmox +// storage definition (else empty — the hub owns it). +// +// This is a cross-repo contract DUPLICATED in felhom.eu/hub (no shared module yet); +// testdata/host-report.golden.json must stay byte-identical with the hub's copy and the +// bidirectional key-set test (contract_test.go) guards drift. +type StorageTarget struct { + Name string `json:"name"` // the Proxmox storage id (its name) + Type string `json:"type"` // local-dir | lvmthin | usb | nfs | cifs | pbs | local + DurableID string `json:"durable_id"` // fs-UUID (usb/local-dir) | server:export (nfs/cifs) | repo+fingerprint (pbs) + // State is the observed lifecycle state. attached (present & usable) | disconnected + // (a KNOWN target whose backing device/mount/reachability dropped) | decommissioned + // (hub-manifest state — built but not served until slice 10). + State string `json:"state"` + Reachable bool `json:"reachable"` // backing device present + mounted (local) / reachable (network) + + TotalBytes int64 `json:"total_bytes"` + UsedBytes int64 `json:"used_bytes"` + AvailBytes int64 `json:"avail_bytes"` + UsedFraction float64 `json:"used_fraction"` + Content string `json:"content"` // Proxmox content list, e.g. "rootdir,images" / "backup,vztmpl" + + MountPath string `json:"mount_path"` // host mountpoint (dir/usb); "" for network/lvm + BackingDevice string `json:"backing_device"` // resolved block device (e.g. /dev/sdb1); "" for network + // ClassHint is a fast|slow HINT derived from the backing disk's rotational flag — a + // hint only; the authoritative class is hub-owned (locked decision). "" when not + // derivable (network targets have no local rotational flag). + ClassHint string `json:"class_hint"` + // Role is primary|vzdump-target|pbs-offsite|bulk-data when derivable from the existing + // storage definition; else "" (the manifest role is hub-owned, slice 10). + Role string `json:"role"` + + // ThinPool carries the lvmthin DATA fill prominently — a full thin-pool corrupts every + // guest on it (the storage analog of single-node OOM). Present ONLY for lvmthin targets; + // metadata fill is null until Phase B's privileged `lvs` read. + ThinPool *ThinPoolFill `json:"thin_pool,omitempty"` + + // Smart is the disk-health summary, populated in Phase B via smartctl (sudoers). In + // Phase A it is {health:"UNKNOWN", } — the keys are committed now so the + // contract is stable across both phases. + Smart SmartSummary `json:"smart"` +} + +// ThinPoolFill is the lvmthin pool fill (doc 03 §7). DataUsedFraction is the live data +// fill (used/total of the lvmthin store); MetadataUsedFraction needs the privileged +// `lvs` read (Phase B) and is null until then. +type ThinPoolFill struct { + DataUsedFraction float64 `json:"data_used_fraction"` + MetadataUsedFraction *float64 `json:"metadata_used_fraction"` +} + +// SmartSummary is a read-only disk-health summary. Health is PASSED|FAILING|UNKNOWN. +// Counters are pointers so "unknown / not-applicable for this device type" (e.g. a USB +// bridge that exposes no SMART, or NVMe counters on a SATA disk) is null, distinct from a +// real zero. The SATA set (reallocated/pending/offline-uncorrectable) and the NVMe set +// (critical_warning/media_errors/percentage_used) are both carried; a device populates +// only its own set. Filled in Phase B. +type SmartSummary struct { + Health string `json:"health"` + + TemperatureC *int `json:"temperature_c"` + PowerOnHours *int `json:"power_on_hours"` + + // SATA attributes. + ReallocatedSectors *int `json:"reallocated_sectors"` + PendingSectors *int `json:"pending_sectors"` + OfflineUncorrectable *int `json:"offline_uncorrectable"` + + // NVMe attributes. + CriticalWarning *int `json:"critical_warning"` + MediaErrors *int `json:"media_errors"` + PercentageUsed *int `json:"percentage_used"` +} + +// SMART health constants (the reported vocabulary). +const ( + SmartPassed = "PASSED" + SmartFailing = "FAILING" + SmartUnknown = "UNKNOWN" +) + +// Storage target type + state constants (the reported vocabulary; doc 03 §7). +const ( + StorageTypeLocalDir = "local-dir" + StorageTypeLVMThin = "lvmthin" + StorageTypeUSB = "usb" + StorageTypeNFS = "nfs" + StorageTypeCIFS = "cifs" + StorageTypePBS = "pbs" + StorageTypeLocal = "local" // builtin dir storage (PVE "local") + + StorageStateAttached = "attached" + StorageStateDisconnected = "disconnected" + StorageStateDecommissioned = "decommissioned" +) + +type Backup struct{} // slice 6: per-target backup status fields TBD type RestoreTest struct{} // slice 6: self-restore-test result fields TBD type PBSSnapshot struct{} // slice 6: PBS snapshot inventory fields TBD type AuditEntry struct{} // audit-log tail entry fields TBD diff --git a/internal/hub/testdata/host-report.golden.json b/internal/hub/testdata/host-report.golden.json index e2a3066..73ff263 100644 --- a/internal/hub/testdata/host-report.golden.json +++ b/internal/hub/testdata/host-report.golden.json @@ -29,7 +29,63 @@ "controller_version": "" } ], - "storage_targets": [], + "storage_targets": [ + { + "name": "local-lvm", + "type": "lvmthin", + "durable_id": "pve/data", + "state": "attached", + "reachable": true, + "total_bytes": 100000000000, + "used_bytes": 42000000000, + "avail_bytes": 58000000000, + "used_fraction": 0.42, + "content": "rootdir,images", + "mount_path": "", + "backing_device": "", + "class_hint": "fast", + "role": "", + "thin_pool": { "data_used_fraction": 0.42, "metadata_used_fraction": null }, + "smart": { + "health": "UNKNOWN", + "temperature_c": null, + "power_on_hours": null, + "reallocated_sectors": null, + "pending_sectors": null, + "offline_uncorrectable": null, + "critical_warning": null, + "media_errors": null, + "percentage_used": null + } + }, + { + "name": "usb-backup", + "type": "usb", + "durable_id": "uuid:0fc63daf-8483-4772-8e79-3d69d8477de4", + "state": "attached", + "reachable": true, + "total_bytes": 2000000000000, + "used_bytes": 500000000000, + "avail_bytes": 1500000000000, + "used_fraction": 0.25, + "content": "backup", + "mount_path": "/mnt/usb-backup", + "backing_device": "/dev/sdb1", + "class_hint": "slow", + "role": "", + "smart": { + "health": "UNKNOWN", + "temperature_c": null, + "power_on_hours": null, + "reallocated_sectors": null, + "pending_sectors": null, + "offline_uncorrectable": null, + "critical_warning": null, + "media_errors": null, + "percentage_used": null + } + } + ], "backups": [], "restore_tests": [], "pbs_snapshots": [], diff --git a/internal/proxmox/types.go b/internal/proxmox/types.go index 19e5adc..626b1d6 100644 --- a/internal/proxmox/types.go +++ b/internal/proxmox/types.go @@ -138,6 +138,12 @@ func (g *GuestConfig) prefixed(prefix string) map[string]string { // Storage is one entry of GET /storage (cluster) and GET /nodes/{node}/storage // (the latter adds usage fields). Unused fields stay zero. +// +// The lower block (Server/Export/Share/Datastore/Fingerprint/VGName/ThinPool) are the +// type-specific config fields the cluster /storage definition carries; they are the +// source for slice-5's deterministic durable_id derivation (server:export for NFS/CIFS, +// repo+fingerprint for PBS, vg/pool for lvmthin). Additive parse-only fields — decoding +// ignores unknown keys, so a storage type that lacks one simply leaves it zero. type Storage struct { Storage string `json:"storage"` Type string `json:"type"` // "dir" | "lvmthin" | "nfs" | "cifs" | "pbs" @@ -150,6 +156,15 @@ type Storage struct { Enabled int `json:"enabled,omitempty"` Shared int `json:"shared,omitempty"` UsedFraction float64 `json:"used_fraction,omitempty"` + + // Type-specific config (durable_id sources). + Server string `json:"server,omitempty"` // nfs/cifs/pbs server host + Export string `json:"export,omitempty"` // nfs export path + Share string `json:"share,omitempty"` // cifs share name + Datastore string `json:"datastore,omitempty"` // pbs datastore name + Fingerprint string `json:"fingerprint,omitempty"` // pbs server cert fingerprint + VGName string `json:"vgname,omitempty"` // lvm/lvmthin volume group + ThinPool string `json:"thinpool,omitempty"` // lvmthin pool LV name } // StorageContent is one entry of GET /nodes/{node}/storage/{store}/content diff --git a/internal/storage/doc.go b/internal/storage/doc.go new file mode 100644 index 0000000..8794a14 --- /dev/null +++ b/internal/storage/doc.go @@ -0,0 +1,29 @@ +// Package storage observes and reconciles the host's storage targets (doc 03 §7). +// +// Slice 5 builds the full model + reconcile machinery; only the read-only, no-hub-desired- +// state parts run live: +// +// - Observe every Proxmox storage target and report it into the host-report +// (hub.StorageTarget). The reported view is what the agent SEES; the hub holds the +// authoritative manifest (desired class/role/policy/creds) and is not served until +// slice 10. So class is a rotational HINT here, never authoritative. +// - A storage watchdog: a fast-poll loop that detects a KNOWN target going +// attached↔disconnected in seconds and triggers an immediate, debounced out-of-band +// host-report (rather than waiting for the slow ~15-minute cycle). +// +// Phase A (this file set) is read-only: every host read it needs — /proc/mounts, +// /dev/disk/by-uuid, /sys/.../rotational, device presence — is non-privileged. Anything +// needing root (SMART via smartctl, lvs for thin-pool metadata, blkid) is deferred to +// Phase B's privileged HostOps surface. +// +// Layout: +// - hostread.go — the HostReader seam + a non-privileged procfs/sysfs implementation. +// - durableid.go — deterministic durable_id derivation per target type (the +// DR-load-bearing field: the hub re-attaches the RIGHT drive by it). +// - observe.go — the Observer: builds []hub.StorageTarget from Proxmox + host reads. +// - watchdog.go — the fast-poll watchdog: transition detection + debounced trigger. +// +// The collector (internal/hub) calls the Observer through a narrow seam, so hub does not +// import storage (storage imports hub for the wire type) — the same interface-seam pattern +// the collector uses for proxmox and cloudflared. +package storage diff --git a/internal/storage/durableid.go b/internal/storage/durableid.go new file mode 100644 index 0000000..52fa645 --- /dev/null +++ b/internal/storage/durableid.go @@ -0,0 +1,80 @@ +package storage + +import ( + "strings" + + "gitea.dooplex.hu/admin/felhom-agent/internal/proxmox" +) + +// deriveDurableID computes the DR-load-bearing durable identifier for a target (doc 03 +// §7). It MUST be deterministic: the hub stores it and, on host loss, re-attaches the +// RIGHT physical target by it — the false-id failure mode is re-attaching the WRONG disk. +// +// Per type: +// - usb / local-dir: the filesystem UUID of the backing device (survives re-cabling / +// re-enumeration that renames /dev/sdX). +// - nfs / cifs: "server:export" (or "server:share") — the network identity. +// - pbs: "server:datastore" plus the cert fingerprint ("…#") — the repo +// identity (the fingerprint pins WHICH PBS, so a spoofed server is a different id). +// - lvmthin / lvm: "vgname/thinpool" (or "vgname") — informational but stable; the VG +// is local and not re-attached cross-host, so a stable name is enough. +// - local (builtin): the backing fs UUID if resolvable, else the path — informational. +// +// uuid is the already-resolved backing-device UUID ("" when unresolved); the caller +// resolves it once (it also needs it for nothing else, so we pass it in to avoid a second +// by-uuid scan). +func deriveDurableID(typ string, s proxmox.Storage, backingDevice, uuid string) string { + switch typ { + case hubTypeNFS: + if s.Server != "" && s.Export != "" { + return s.Server + ":" + s.Export + } + case hubTypeCIFS: + if s.Server != "" && s.Share != "" { + return s.Server + ":" + s.Share + } + case hubTypePBS: + repo := s.Datastore + if s.Server != "" { + repo = s.Server + ":" + s.Datastore + } + if repo != "" { + if s.Fingerprint != "" { + return repo + "#" + strings.ToLower(s.Fingerprint) + } + return repo + } + case hubTypeLVMThin, "lvm": + if s.VGName != "" { + if s.ThinPool != "" { + return s.VGName + "/" + s.ThinPool + } + return s.VGName + } + case hubTypeUSB, hubTypeLocalDir, hubTypeLocal: + if uuid != "" { + return "uuid:" + uuid + } + if backingDevice != "" { + return "dev:" + backingDevice + } + if s.Path != "" { + return "path:" + s.Path + } + } + // Fallback: a stable, unambiguous id from the storage name — never empty (an empty + // durable_id would defeat the hub's re-attach lookup). + return "store:" + s.Storage +} + +// Reported storage-type strings (mirror hub's StorageType* constants without importing +// hub here for the bare strings — the observer maps to these). +const ( + hubTypeLocalDir = "local-dir" + hubTypeLVMThin = "lvmthin" + hubTypeUSB = "usb" + hubTypeNFS = "nfs" + hubTypeCIFS = "cifs" + hubTypePBS = "pbs" + hubTypeLocal = "local" +) diff --git a/internal/storage/durableid_test.go b/internal/storage/durableid_test.go new file mode 100644 index 0000000..a09d241 --- /dev/null +++ b/internal/storage/durableid_test.go @@ -0,0 +1,90 @@ +package storage + +import ( + "testing" + + "gitea.dooplex.hu/admin/felhom-agent/internal/proxmox" +) + +// durable_id is the DR-load-bearing field — the hub re-attaches the RIGHT physical target +// by it. Each type must derive deterministically; the false-id failure mode is +// re-attaching the WRONG disk, so this table pins the per-type shape. +func TestDeriveDurableID(t *testing.T) { + cases := []struct { + name string + typ string + s proxmox.Storage + backingDevice string + uuid string + want string + }{ + { + name: "usb by fs-uuid", + typ: hubTypeUSB, + s: proxmox.Storage{Storage: "usb-backup", Path: "/mnt/usb-backup"}, + uuid: "0fc6-abcd", want: "uuid:0fc6-abcd", + }, + { + name: "local-dir by fs-uuid", + typ: hubTypeLocalDir, s: proxmox.Storage{Storage: "extra"}, uuid: "dead-beef", + want: "uuid:dead-beef", + }, + { + name: "usb falls back to device when uuid unresolved", + typ: hubTypeUSB, s: proxmox.Storage{Storage: "usb-backup"}, backingDevice: "/dev/sdb1", + want: "dev:/dev/sdb1", + }, + { + name: "nfs server:export", + typ: hubTypeNFS, s: proxmox.Storage{Storage: "nfs-arch", Server: "10.0.0.5", Export: "/export/b"}, + want: "10.0.0.5:/export/b", + }, + { + name: "cifs server:share", + typ: hubTypeCIFS, s: proxmox.Storage{Storage: "cifs", Server: "nas.local", Share: "backups"}, + want: "nas.local:backups", + }, + { + name: "pbs repo + fingerprint (lowercased)", + typ: hubTypePBS, + s: proxmox.Storage{Storage: "pbs", Server: "pbs.local", Datastore: "store1", Fingerprint: "AB:CD:EF"}, + want: "pbs.local:store1#ab:cd:ef", + }, + { + name: "pbs without fingerprint", + typ: hubTypePBS, s: proxmox.Storage{Storage: "pbs", Server: "pbs.local", Datastore: "store1"}, + want: "pbs.local:store1", + }, + { + name: "lvmthin vg/pool", + typ: hubTypeLVMThin, s: proxmox.Storage{Storage: "local-lvm", VGName: "pve", ThinPool: "data"}, + want: "pve/data", + }, + { + name: "lvm (thick) vg only", + typ: "lvm", s: proxmox.Storage{Storage: "vg0", VGName: "vg0"}, + want: "vg0", + }, + { + name: "local builtin by path when no uuid", + typ: hubTypeLocal, s: proxmox.Storage{Storage: "local", Path: "/var/lib/vz"}, + want: "path:/var/lib/vz", + }, + { + name: "unknown/unresolvable falls back to store name (never empty)", + typ: hubTypeNFS, s: proxmox.Storage{Storage: "broken-nfs"}, // missing server/export + want: "store:broken-nfs", + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got := deriveDurableID(c.typ, c.s, c.backingDevice, c.uuid) + if got != c.want { + t.Errorf("deriveDurableID = %q, want %q", got, c.want) + } + if got == "" { + t.Error("durable_id must never be empty") + } + }) + } +} diff --git a/internal/storage/hostread.go b/internal/storage/hostread.go new file mode 100644 index 0000000..fd840c3 --- /dev/null +++ b/internal/storage/hostread.go @@ -0,0 +1,231 @@ +package storage + +import ( + "bufio" + "os" + "path/filepath" + "strings" +) + +// HostReader is the non-privileged host-read seam the observer and watchdog need. All of +// it is read-only and root-free: the active mount table, fs-UUID resolution via the +// /dev/disk/by-uuid symlinks, block-device presence, and the rotational/removable sysfs +// flags. Production is *ProcHostReader; tests inject a fake. +// +// Anything that needs root (smartctl, lvs, blkid) is NOT here — it lands on Phase B's +// privileged HostOps surface. Keep this seam root-free. +type HostReader interface { + // Mounts parses the active mount table (/proc/mounts). + Mounts() ([]Mount, error) + // ResolveUUID returns the filesystem UUID of a block device, derived from the + // /dev/disk/by-uuid symlinks. ok=false when the device has no by-uuid entry. + ResolveUUID(device string) (uuid string, ok bool) + // DeviceExists reports whether a block-device node is present (the fast USB-drop + // signal for the watchdog). + DeviceExists(device string) bool + // Rotational reads the backing disk's rotational flag (true=HDD/slow, false=SSD/fast). + // ok=false when it cannot be determined (network fs, missing sysfs, device-mapper). + Rotational(device string) (rotational bool, ok bool) + // Removable reads the backing disk's removable flag (true => a USB/hot-plug device). + // ok=false when it cannot be determined. + Removable(device string) (removable bool, ok bool) +} + +// Mount is one active-mount-table entry. +type Mount struct { + Device string // e.g. "/dev/sdb1", "server:/export", "//server/share" + MountPoint string + FSType string +} + +// ProcHostReader is the production HostReader: it reads the host's /proc, /dev, and /sys. +// The paths are fields so tests CAN point it at fixtures, though most tests use a fully +// fake HostReader instead. +type ProcHostReader struct { + ProcMounts string // default "/proc/mounts" + ByUUIDDir string // default "/dev/disk/by-uuid" + SysClass string // default "/sys/class/block" +} + +// NewProcHostReader builds a ProcHostReader with the standard host paths. +func NewProcHostReader() *ProcHostReader { + return &ProcHostReader{ + ProcMounts: "/proc/mounts", + ByUUIDDir: "/dev/disk/by-uuid", + SysClass: "/sys/class/block", + } +} + +// Mounts parses /proc/mounts. The format is space-separated, octal-escaped fields: +// device mountpoint fstype options dump pass. We unescape the first three. +func (r *ProcHostReader) Mounts() ([]Mount, error) { + f, err := os.Open(r.procMounts()) + if err != nil { + return nil, err + } + defer f.Close() + + var out []Mount + sc := bufio.NewScanner(f) + for sc.Scan() { + fields := strings.Fields(sc.Text()) + if len(fields) < 3 { + continue + } + out = append(out, Mount{ + Device: unescapeMount(fields[0]), + MountPoint: unescapeMount(fields[1]), + FSType: fields[2], + }) + } + return out, sc.Err() +} + +// ResolveUUID reverse-maps a device path to its fs-UUID by reading the /dev/disk/by-uuid +// symlinks and matching the canonical target of each against the device. +func (r *ProcHostReader) ResolveUUID(device string) (string, bool) { + if device == "" { + return "", false + } + want := canonPath(device) + entries, err := os.ReadDir(r.byUUIDDir()) + if err != nil { + return "", false + } + for _, e := range entries { + link := filepath.Join(r.byUUIDDir(), e.Name()) + target, err := os.Readlink(link) + if err != nil { + continue + } + if !filepath.IsAbs(target) { + target = filepath.Join(r.byUUIDDir(), target) + } + if canonPath(target) == want { + return e.Name(), true + } + } + return "", false +} + +// DeviceExists stats the device node (after resolving symlinks like /dev/disk/by-uuid/X). +func (r *ProcHostReader) DeviceExists(device string) bool { + if device == "" { + return false + } + _, err := os.Stat(device) + return err == nil +} + +// Rotational reads /sys/block//queue/rotational for the device's backing +// disk. "1" => rotational (HDD/slow), "0" => SSD/fast. +func (r *ProcHostReader) Rotational(device string) (bool, bool) { + disk, ok := r.parentDisk(device) + if !ok { + return false, false + } + b, err := os.ReadFile(filepath.Join(r.sysBlockDir(), disk, "queue", "rotational")) + if err != nil { + return false, false + } + switch strings.TrimSpace(string(b)) { + case "1": + return true, true + case "0": + return false, true + } + return false, false +} + +// Removable reads /sys/block//removable. "1" => removable (USB/hot-plug). +func (r *ProcHostReader) Removable(device string) (bool, bool) { + disk, ok := r.parentDisk(device) + if !ok { + return false, false + } + b, err := os.ReadFile(filepath.Join(r.sysBlockDir(), disk, "removable")) + if err != nil { + return false, false + } + switch strings.TrimSpace(string(b)) { + case "1": + return true, true + case "0": + return false, true + } + return false, false +} + +// parentDisk maps a device path (possibly a partition like /dev/sdb1 or /dev/nvme0n1p2) +// to its parent disk's sysfs name (sdb / nvme0n1). It uses /sys/class/block/, whose +// real path ends in ...// for a partition and .../ for a whole disk. +func (r *ProcHostReader) parentDisk(device string) (string, bool) { + name := filepath.Base(strings.TrimSpace(device)) + if name == "" || name == "." || name == "/" { + return "", false + } + real, err := filepath.EvalSymlinks(filepath.Join(r.sysClass(), name)) + if err != nil { + return "", false + } + // If is a partition, /sys/class/block//partition exists and its parent + // directory is the disk. Otherwise IS the disk. + if _, err := os.Stat(filepath.Join(real, "partition")); err == nil { + return filepath.Base(filepath.Dir(real)), true + } + return filepath.Base(real), true +} + +// sysBlockDir derives /sys/block from the configured /sys/class/block. +func (r *ProcHostReader) sysBlockDir() string { + return filepath.Join(filepath.Dir(filepath.Dir(r.sysClass())), "block") +} + +func (r *ProcHostReader) procMounts() string { + if r.ProcMounts != "" { + return r.ProcMounts + } + return "/proc/mounts" +} +func (r *ProcHostReader) byUUIDDir() string { + if r.ByUUIDDir != "" { + return r.ByUUIDDir + } + return "/dev/disk/by-uuid" +} +func (r *ProcHostReader) sysClass() string { + if r.SysClass != "" { + return r.SysClass + } + return "/sys/class/block" +} + +// canonPath resolves symlinks for a best-effort canonical comparison, falling back to the +// cleaned path when the target can't be resolved (e.g. the device just disappeared). +func canonPath(p string) string { + if real, err := filepath.EvalSymlinks(p); err == nil { + return real + } + return filepath.Clean(p) +} + +// unescapeMount decodes the octal \040-style escapes /proc/mounts uses for spaces, tabs, +// newlines and backslashes in the device/mountpoint fields. +func unescapeMount(s string) string { + if !strings.Contains(s, `\`) { + return s + } + var b strings.Builder + for i := 0; i < len(s); i++ { + if s[i] == '\\' && i+3 < len(s) && isOctal(s[i+1]) && isOctal(s[i+2]) && isOctal(s[i+3]) { + v := (int(s[i+1]-'0') << 6) | (int(s[i+2]-'0') << 3) | int(s[i+3]-'0') + b.WriteByte(byte(v)) + i += 3 + continue + } + b.WriteByte(s[i]) + } + return b.String() +} + +func isOctal(c byte) bool { return c >= '0' && c <= '7' } diff --git a/internal/storage/observe.go b/internal/storage/observe.go new file mode 100644 index 0000000..6a945e7 --- /dev/null +++ b/internal/storage/observe.go @@ -0,0 +1,398 @@ +package storage + +import ( + "context" + "fmt" + "log/slog" + "strings" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" + "gitea.dooplex.hu/admin/felhom-agent/internal/proxmox" +) + +// thinPoolWarnFraction is the lvmthin DATA-fill level above which the observer logs a +// prominent warning. A full thin-pool corrupts EVERY guest on it (the storage analog of a +// single-node OOM), so it must be visible early — well before slice-10 policy exists. +const thinPoolWarnFraction = 0.85 + +// StorageAPI is the read-only Proxmox surface the observer needs. *proxmox.Client +// satisfies it. ListStorage (cluster) carries the type-specific config (server/export/ +// vgname/thinpool/fingerprint) that NodeStorage may omit; NodeStorage carries live usage +// + the per-node active flag. The observer joins them by storage name. +type StorageAPI interface { + Node() string + ListStorage(ctx context.Context) ([]proxmox.Storage, error) + NodeStorage(ctx context.Context) ([]proxmox.Storage, error) +} + +// Observer builds the observed storage view from Proxmox + non-privileged host reads. +type Observer struct { + api StorageAPI + host HostReader + logger *slog.Logger +} + +// NewObserver builds an Observer. host defaults to a ProcHostReader; logger to the +// default. A nil api makes Observe/Known return an error (misconfiguration), never panic. +func NewObserver(api StorageAPI, host HostReader, logger *slog.Logger) *Observer { + if host == nil { + host = NewProcHostReader() + } + if logger == nil { + logger = slog.Default() + } + return &Observer{api: api, host: host, logger: logger} +} + +// observed is the rich internal view of one target, from which both the reported +// hub.StorageTarget and the watchdog's KnownTarget are projected. +type observed struct { + target hub.StorageTarget + known KnownTarget +} + +// Observe builds the reported []hub.StorageTarget. A non-nil error means the Proxmox read +// failed (the collector then omits storage from this cycle's report but still sends the +// rest). The returned slice is always non-nil so it marshals as [] when empty. +func (o *Observer) Observe(ctx context.Context) ([]hub.StorageTarget, error) { + snap, err := o.snapshot(ctx) + if err != nil { + return nil, err + } + out := make([]hub.StorageTarget, 0, len(snap)) + for _, s := range snap { + out = append(out, s.target) + } + return out, nil +} + +// Known projects the snapshot to the watchdog's lightweight KnownTarget set. Same Proxmox +// + host reads as Observe — callers that poll it fast should wrap it in a cache (the +// watchdog uses CachingKnownTargets). +func (o *Observer) Known(ctx context.Context) ([]KnownTarget, error) { + snap, err := o.snapshot(ctx) + if err != nil { + return nil, err + } + out := make([]KnownTarget, 0, len(snap)) + for _, s := range snap { + out = append(out, s.known) + } + return out, nil +} + +// snapshot does the full build: join cluster config + node usage, then derive each +// target's identity, state, class hint, and (for lvmthin) thin-pool fill from host reads. +func (o *Observer) snapshot(ctx context.Context) ([]observed, error) { + if o.api == nil { + return nil, fmt.Errorf("storage: no proxmox api configured") + } + cluster, err := o.api.ListStorage(ctx) + if err != nil { + return nil, fmt.Errorf("storage: ListStorage: %w", err) + } + cfgByName := make(map[string]proxmox.Storage, len(cluster)) + for _, c := range cluster { + cfgByName[c.Storage] = c + } + nodeStores, err := o.api.NodeStorage(ctx) + if err != nil { + return nil, fmt.Errorf("storage: NodeStorage: %w", err) + } + mounts, err := o.host.Mounts() + if err != nil { + // Host mount read failed: degrade rather than fail the whole report — Proxmox + // usage/active is still meaningful; we just lose mount-derived fields. + o.logger.Warn("storage: reading mounts failed; mount/device fields degraded", "err", err) + mounts = nil + } + + out := make([]observed, 0, len(nodeStores)) + for _, ns := range nodeStores { + // Overlay the cluster config (server/export/vgname/...) onto the node entry. + s := mergeConfig(ns, cfgByName[ns.Storage]) + out = append(out, o.build(s, mounts)) + } + return out, nil +} + +// build derives one observed target from a merged Storage entry + the mount table. +func (o *Observer) build(s proxmox.Storage, mounts []Mount) observed { + category := categorize(s.Type) + + // Resolve the backing device + mount path for dir-like targets. + var backingDevice, mountPath string + var exactMount bool + if category == catDir { + if dev, mp, ok := exactMountDevice(mounts, s.Path); ok { + backingDevice, mountPath, exactMount = dev, mp, true + } else if dev, ok := containingMountDevice(mounts, s.Path); ok { + backingDevice = dev // for the class hint only; not its own mount + } + } + + // Type: distinguish builtin local / removable USB / fixed local-dir within "dir". + removable, removableKnown := false, false + if category == catDir && backingDevice != "" { + removable, removableKnown = o.host.Removable(backingDevice) + } + typ := reportType(s, category, removable, removableKnown) + + // durable_id (DR-load-bearing). + var uuid string + if category == catDir && backingDevice != "" { + uuid, _ = o.host.ResolveUUID(backingDevice) + } + durableID := deriveDurableID(typ, s, backingDevice, uuid) + + // Reachability + state. + reachable := o.reachable(typ, category, s, backingDevice, exactMount) + state := hub.StorageStateAttached + if !reachable { + state = hub.StorageStateDisconnected + } + + // Class hint (rotational; local block-backed only — a HINT, never authoritative). + classHint := "" + if category == catDir && backingDevice != "" { + if rot, ok := o.host.Rotational(backingDevice); ok { + if rot { + classHint = "slow" + } else { + classHint = "fast" + } + } + } + + tgt := hub.StorageTarget{ + Name: s.Storage, + Type: typ, + DurableID: durableID, + State: state, + Reachable: reachable, + TotalBytes: s.Total, + UsedBytes: s.Used, + AvailBytes: s.Avail, + UsedFraction: usedFraction(s), + Content: s.Content, + MountPath: mountPath, + BackingDevice: backingDevice, + ClassHint: classHint, + Role: "", // hub-owned; not derivable from a Proxmox def (slice 10) + Smart: hub.SmartSummary{Health: hub.SmartUnknown}, + } + + // Thin-pool DATA fill: surfaced prominently for lvmthin (metadata fill is Phase B/lvs). + if typ == hub.StorageTypeLVMThin { + frac := usedFraction(s) + tgt.ThinPool = &hub.ThinPoolFill{DataUsedFraction: frac} + if frac >= thinPoolWarnFraction { + o.logger.Warn("storage: lvmthin pool data fill is high (a full pool corrupts every guest on it)", + "storage", s.Storage, "data_used_fraction", frac) + } + } + + return observed{ + target: tgt, + known: KnownTarget{ + Name: s.Storage, + Type: typ, + DurableID: durableID, + Network: category == catNetwork, + MountBacked: typ == hub.StorageTypeUSB || typ == hub.StorageTypeLocalDir, + BackingDevice: backingDevice, + MountPath: s.Path, + ReachEndpoint: reachEndpoint(typ, s), + }, + } +} + +// reachable decides whether the target is currently usable. +// - usb / local-dir: a Felhom extra/removable dir storage is realized as its OWN +// mountpoint, so reachable = it is currently an exact mount AND its device node exists. +// Not-its-own-mount = unplugged/unmounted = disconnected. This is the fast USB-drop +// signal — we deliberately do NOT fall through to PVE's active flag, because the +// mountpoint directory still exists on the root fs when the device is gone, so active +// can read stale-attached. +// - local (builtin PVE "local"): lives within the root fs by design, so trust active. +// - network (nfs/cifs/pbs) and block-pool (lvmthin/lvm): trust PVE's active flag — PVE +// actively probes these and flips active=0 when down. +func (o *Observer) reachable(typ string, category storageCategory, s proxmox.Storage, backingDevice string, exactMount bool) bool { + switch typ { + case hub.StorageTypeUSB, hub.StorageTypeLocalDir: + return exactMount && (backingDevice == "" || o.host.DeviceExists(backingDevice)) + default: + // local, lvmthin, lvm, nfs, cifs, pbs. + _ = category + return s.Active == 1 + } +} + +// usedFraction prefers Proxmox's reported used_fraction, falling back to used/total. +func usedFraction(s proxmox.Storage) float64 { + if s.UsedFraction > 0 { + return s.UsedFraction + } + if s.Total > 0 { + return float64(s.Used) / float64(s.Total) + } + return 0 +} + +// storageCategory groups Proxmox storage types by how state/identity are derived. +type storageCategory int + +const ( + catDir storageCategory = iota // dir-backed (local/usb/local-dir) + catNetwork // nfs/cifs/pbs + catBlock // lvmthin/lvm + catOther +) + +func categorize(pxType string) storageCategory { + switch pxType { + case "dir": + return catDir + case "nfs", "cifs", "smb", "pbs": + return catNetwork + case "lvmthin", "lvm": + return catBlock + default: + return catOther + } +} + +// reportType maps a Proxmox storage type to the reported vocabulary, splitting "dir" into +// builtin local / removable usb / fixed local-dir. +func reportType(s proxmox.Storage, category storageCategory, removable, removableKnown bool) string { + switch category { + case catDir: + if s.Storage == "local" { + return hub.StorageTypeLocal + } + if removableKnown && removable { + return hub.StorageTypeUSB + } + return hub.StorageTypeLocalDir + case catNetwork: + switch s.Type { + case "nfs": + return hub.StorageTypeNFS + case "cifs", "smb": + return hub.StorageTypeCIFS + case "pbs": + return hub.StorageTypePBS + } + case catBlock: + if s.Type == "lvmthin" { + return hub.StorageTypeLVMThin + } + return s.Type // "lvm" (thick) passes through + } + return s.Type +} + +// reachEndpoint builds the host:port the watchdog dials for a network target's +// reachability check (default ports per protocol). "" for non-network targets. +func reachEndpoint(typ string, s proxmox.Storage) string { + if s.Server == "" { + return "" + } + switch typ { + case hub.StorageTypeNFS: + return netJoin(s.Server, "2049") + case hub.StorageTypeCIFS: + return netJoin(s.Server, "445") + case hub.StorageTypePBS: + return netJoin(s.Server, "8007") + } + return "" +} + +func netJoin(host, port string) string { + if strings.Contains(host, ":") && !strings.HasPrefix(host, "[") { + host = "[" + host + "]" // IPv6 literal + } + return host + ":" + port +} + +// exactMountDevice finds the mount whose mountpoint EXACTLY equals path (the target is its +// own mount — the meaningful state for a USB/extra disk). +func exactMountDevice(mounts []Mount, path string) (device, mountPoint string, ok bool) { + if path == "" { + return "", "", false + } + clean := cleanMountPath(path) + for _, m := range mounts { + if cleanMountPath(m.MountPoint) == clean { + return m.Device, m.MountPoint, true + } + } + return "", "", false +} + +// containingMountDevice finds the device of the longest mountpoint that is a prefix of +// path (the filesystem that path lives on) — used only for the class-hint disk lookup. +func containingMountDevice(mounts []Mount, path string) (device string, ok bool) { + if path == "" { + return "", false + } + clean := cleanMountPath(path) + best := -1 + for _, m := range mounts { + mp := cleanMountPath(m.MountPoint) + if clean == mp || strings.HasPrefix(clean, mp+"/") || mp == "/" { + if len(mp) > best { + best, device, ok = len(mp), m.Device, true + } + } + } + return device, ok +} + +func cleanMountPath(p string) string { + p = strings.TrimRight(p, "/") + if p == "" { + return "/" + } + return p +} + +// mergeConfig overlays the cluster-def config fields (which the per-node entry may omit) +// onto a node-storage entry, keeping the node's live usage/active values. +func mergeConfig(node, cluster proxmox.Storage) proxmox.Storage { + if cluster.Storage == "" { + return node + } + if node.Type == "" { + node.Type = cluster.Type + } + if node.Server == "" { + node.Server = cluster.Server + } + if node.Export == "" { + node.Export = cluster.Export + } + if node.Share == "" { + node.Share = cluster.Share + } + if node.Datastore == "" { + node.Datastore = cluster.Datastore + } + if node.Fingerprint == "" { + node.Fingerprint = cluster.Fingerprint + } + if node.VGName == "" { + node.VGName = cluster.VGName + } + if node.ThinPool == "" { + node.ThinPool = cluster.ThinPool + } + if node.Path == "" { + node.Path = cluster.Path + } + if node.Content == "" { + node.Content = cluster.Content + } + return node +} diff --git a/internal/storage/observe_test.go b/internal/storage/observe_test.go new file mode 100644 index 0000000..6964a98 --- /dev/null +++ b/internal/storage/observe_test.go @@ -0,0 +1,209 @@ +package storage + +import ( + "context" + "io" + "log/slog" + "testing" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" + "gitea.dooplex.hu/admin/felhom-agent/internal/proxmox" +) + +func quietLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// fakeStorageAPI serves fixed cluster + node storage lists. +type fakeStorageAPI struct { + node string + cluster []proxmox.Storage + nodeSt []proxmox.Storage + listErr error + nodeErr error +} + +func (f *fakeStorageAPI) Node() string { return f.node } +func (f *fakeStorageAPI) ListStorage(context.Context) ([]proxmox.Storage, error) { + return f.cluster, f.listErr +} +func (f *fakeStorageAPI) NodeStorage(context.Context) ([]proxmox.Storage, error) { + return f.nodeSt, f.nodeErr +} + +// fakeHostReader is a fully synthetic HostReader — no real devices touched. +type fakeHostReader struct { + mounts []Mount + mountsErr error + uuids map[string]string // device -> uuid + exists map[string]bool // device -> present + rotational map[string]bool // device -> rotational (presence => known) + removable map[string]bool // device -> removable (presence => known) +} + +func (h *fakeHostReader) Mounts() ([]Mount, error) { return h.mounts, h.mountsErr } +func (h *fakeHostReader) ResolveUUID(device string) (string, bool) { + u, ok := h.uuids[device] + return u, ok +} +func (h *fakeHostReader) DeviceExists(device string) bool { return h.exists[device] } +func (h *fakeHostReader) Rotational(device string) (bool, bool) { + v, ok := h.rotational[device] + return v, ok +} +func (h *fakeHostReader) Removable(device string) (bool, bool) { + v, ok := h.removable[device] + return v, ok +} + +// byName indexes observed targets for assertions. +func byName(targets []hub.StorageTarget) map[string]hub.StorageTarget { + m := make(map[string]hub.StorageTarget, len(targets)) + for _, t := range targets { + m[t.Name] = t + } + return m +} + +func TestObserve_BuildsTargetsFromProxmoxAndHostReads(t *testing.T) { + api := &fakeStorageAPI{ + node: "demo-felhom", + cluster: []proxmox.Storage{ + {Storage: "local", Type: "dir", Content: "vztmpl,backup", Path: "/var/lib/vz"}, + {Storage: "local-lvm", Type: "lvmthin", Content: "rootdir,images", VGName: "pve", ThinPool: "data"}, + {Storage: "usb-backup", Type: "dir", Content: "backup", Path: "/mnt/usb-backup"}, + {Storage: "nfs-arch", Type: "nfs", Content: "backup", Server: "10.0.0.5", Export: "/export/backups"}, + }, + nodeSt: []proxmox.Storage{ + {Storage: "local", Type: "dir", Content: "vztmpl,backup", Path: "/var/lib/vz", Total: 100, Used: 20, Avail: 80, Active: 1, UsedFraction: 0.2}, + {Storage: "local-lvm", Type: "lvmthin", Content: "rootdir,images", Total: 1000, Used: 900, Avail: 100, Active: 1, UsedFraction: 0.9}, + {Storage: "usb-backup", Type: "dir", Content: "backup", Path: "/mnt/usb-backup", Total: 2000, Used: 500, Avail: 1500, Active: 1, UsedFraction: 0.25}, + {Storage: "nfs-arch", Type: "nfs", Content: "backup", Total: 5000, Used: 1000, Avail: 4000, Active: 1, UsedFraction: 0.2}, + }, + } + host := &fakeHostReader{ + mounts: []Mount{ + {Device: "/dev/mapper/pve-root", MountPoint: "/", FSType: "ext4"}, + {Device: "/dev/sdb1", MountPoint: "/mnt/usb-backup", FSType: "ext4"}, + }, + uuids: map[string]string{"/dev/sdb1": "1111-2222"}, + exists: map[string]bool{"/dev/sdb1": true, "/dev/mapper/pve-root": true}, + rotational: map[string]bool{"/dev/sdb1": true, "/dev/mapper/pve-root": false}, + removable: map[string]bool{"/dev/sdb1": true, "/dev/mapper/pve-root": false}, + } + + got, err := NewObserver(api, host, quietLogger()).Observe(context.Background()) + if err != nil { + t.Fatalf("Observe: %v", err) + } + if len(got) != 4 { + t.Fatalf("got %d targets, want 4", len(got)) + } + m := byName(got) + + // builtin local: dir within root → type "local", attached via active flag. + if local := m["local"]; local.Type != hub.StorageTypeLocal || local.State != hub.StorageStateAttached { + t.Errorf("local = %+v, want type=local state=attached", local) + } + + // lvmthin: thin-pool DATA fill surfaced; durable_id = vg/pool; no mount/device. + lvm := m["local-lvm"] + if lvm.Type != hub.StorageTypeLVMThin || lvm.DurableID != "pve/data" { + t.Errorf("local-lvm type/durable = %q/%q", lvm.Type, lvm.DurableID) + } + if lvm.ThinPool == nil || lvm.ThinPool.DataUsedFraction != 0.9 { + t.Errorf("local-lvm thin_pool = %+v, want data_used_fraction=0.9", lvm.ThinPool) + } + if lvm.ThinPool.MetadataUsedFraction != nil { + t.Errorf("metadata fill must be nil in Phase A (lvs is Phase B)") + } + + // usb: removable dir, mounted → type usb, durable_id from UUID, class_hint slow (rotational). + usb := m["usb-backup"] + if usb.Type != hub.StorageTypeUSB { + t.Errorf("usb-backup type = %q, want usb", usb.Type) + } + if usb.DurableID != "uuid:1111-2222" { + t.Errorf("usb-backup durable_id = %q, want uuid:1111-2222", usb.DurableID) + } + if usb.ClassHint != "slow" { + t.Errorf("usb-backup class_hint = %q, want slow (rotational)", usb.ClassHint) + } + if usb.MountPath != "/mnt/usb-backup" || usb.BackingDevice != "/dev/sdb1" { + t.Errorf("usb-backup mount/device = %q/%q", usb.MountPath, usb.BackingDevice) + } + if usb.State != hub.StorageStateAttached || !usb.Reachable { + t.Errorf("usb-backup should be attached+reachable: %+v", usb) + } + if usb.ThinPool != nil { + t.Errorf("non-lvmthin must omit thin_pool") + } + + // nfs: durable_id = server:export; attached via active flag; no class hint. + nfs := m["nfs-arch"] + if nfs.Type != hub.StorageTypeNFS || nfs.DurableID != "10.0.0.5:/export/backups" { + t.Errorf("nfs-arch type/durable = %q/%q", nfs.Type, nfs.DurableID) + } + if nfs.ClassHint != "" { + t.Errorf("network target must have no class hint, got %q", nfs.ClassHint) + } + + // SMART is UNKNOWN in Phase A for every target. + for _, tgt := range got { + if tgt.Smart.Health != hub.SmartUnknown { + t.Errorf("%s smart health = %q, want UNKNOWN in Phase A", tgt.Name, tgt.Smart.Health) + } + } +} + +func TestObserve_USBUnpluggedIsDisconnected(t *testing.T) { + api := &fakeStorageAPI{ + node: "demo-felhom", + cluster: []proxmox.Storage{ + {Storage: "usb-backup", Type: "dir", Content: "backup", Path: "/mnt/usb-backup"}, + }, + nodeSt: []proxmox.Storage{ + // PVE may still show the storage entry (active possibly 1) but the mount is gone. + {Storage: "usb-backup", Type: "dir", Content: "backup", Path: "/mnt/usb-backup", Active: 1}, + }, + } + host := &fakeHostReader{ + mounts: []Mount{{Device: "/dev/mapper/pve-root", MountPoint: "/", FSType: "ext4"}}, // no /mnt/usb-backup + } + got, err := NewObserver(api, host, quietLogger()).Observe(context.Background()) + if err != nil { + t.Fatal(err) + } + usb := byName(got)["usb-backup"] + // Without its own mount, a USB target is unplugged regardless of PVE's stale active flag. + if usb.State != hub.StorageStateDisconnected || usb.Reachable { + t.Errorf("unplugged usb should be disconnected/unreachable, got state=%q reachable=%v", usb.State, usb.Reachable) + } + // Its durable_id falls back to a stable form (no UUID resolvable while detached). + if usb.DurableID == "" { + t.Errorf("durable_id must never be empty (DR re-attach lookup)") + } +} + +func TestObserve_ProxmoxErrorIsFatalForStorage(t *testing.T) { + api := &fakeStorageAPI{node: "n", listErr: context.DeadlineExceeded} + if _, err := NewObserver(api, &fakeHostReader{}, quietLogger()).Observe(context.Background()); err == nil { + t.Fatal("a Proxmox read error must surface (the collector then omits storage this cycle)") + } +} + +func TestObserve_MountReadFailureDegradesNotFatal(t *testing.T) { + api := &fakeStorageAPI{ + node: "n", + cluster: []proxmox.Storage{{Storage: "local-lvm", Type: "lvmthin", VGName: "pve", ThinPool: "data"}}, + nodeSt: []proxmox.Storage{{Storage: "local-lvm", Type: "lvmthin", Active: 1, UsedFraction: 0.1}}, + } + host := &fakeHostReader{mountsErr: io.ErrUnexpectedEOF} + got, err := NewObserver(api, host, quietLogger()).Observe(context.Background()) + if err != nil { + t.Fatalf("a host mount-read failure must degrade, not fail: %v", err) + } + if len(got) != 1 || got[0].DurableID != "pve/data" { + t.Errorf("lvmthin still derivable without mounts: %+v", got) + } +} diff --git a/internal/storage/watchdog.go b/internal/storage/watchdog.go new file mode 100644 index 0000000..4f6fd08 --- /dev/null +++ b/internal/storage/watchdog.go @@ -0,0 +1,304 @@ +package storage + +import ( + "context" + "log/slog" + "net" + "sync" + "time" +) + +// Default watchdog timings (configurable via WatchdogOptions). The poll is FAST (seconds) +// so a USB drop is caught in seconds, not at the slow ~15-minute host-report cycle; the +// debounce keeps a flapping drive from storming the hub. +const ( + DefaultWatchdogInterval = 8 * time.Second + DefaultWatchdogDebounce = 30 * time.Second +) + +// KnownTarget is the watchdog's lightweight view of a target it watches. "Known" means a +// defined Proxmox storage (and/or a previously-observed-attached one); the watchdog only +// flags transitions for targets it has seen — it never reports a never-attached device. +type KnownTarget struct { + Name string + Type string + DurableID string + Network bool // nfs/cifs/pbs — liveness is a reachability dial, not a device check + MountBacked bool // usb/local-dir — a drop = its mountpoint disappears + BackingDevice string // resolved block device (local targets) + MountPath string // the mountpoint a mount-backed target must occupy + ReachEndpoint string // host:port to dial for a network target's reachability +} + +// KnownTargets enumerates the currently-known target set. Production wraps the Observer in +// CachingKnownTargets so the fast poll doesn't hammer the Proxmox API. +type KnownTargets interface { + Known(ctx context.Context) ([]KnownTarget, error) +} + +// TargetLiveness reports whether one known target is presently up. Production is +// HostLiveness (device/mount presence + a reachability dial, all non-privileged); tests +// inject a fake. +type TargetLiveness interface { + Present(ctx context.Context, t KnownTarget) bool +} + +// Transition is one observed state change for a known target (for logging/diagnostics). +type Transition struct { + Name string + From string // attached | disconnected + To string +} + +// Watchdog is the third daemon goroutine (alongside the hub loop + reconcile engine). It +// fast-polls the known target set, detects attached↔disconnected transitions, and triggers +// an immediate, debounced out-of-band host-report so the hub learns of a drop in seconds. +// +// It NEVER mutates anything (Phase A is read-only) — the benign re-mount-by-UUID response +// to a return lands in Phase B. Here it only observes and signals. +type Watchdog struct { + targets KnownTargets + liveness TargetLiveness + interval time.Duration + debounce time.Duration + trigger func() // request an out-of-band report (debounced by the watchdog) + logger *slog.Logger + now func() time.Time + + mu sync.Mutex + last map[string]bool // name -> last observed present (only for seen targets) + lastFire time.Time + fired bool // lastFire is valid + pending bool // a transition is awaiting the debounce window +} + +// WatchdogOptions configures a Watchdog. Targets, Liveness and Trigger are required; the +// rest default. +type WatchdogOptions struct { + Targets KnownTargets + Liveness TargetLiveness + Trigger func() + Interval time.Duration + Debounce time.Duration + Logger *slog.Logger +} + +// NewWatchdog builds a Watchdog. A nil Trigger is tolerated (the watchdog still tracks +// state, just signals nothing) so it degrades cleanly when no report sink is wired. +func NewWatchdog(opts WatchdogOptions) *Watchdog { + interval := opts.Interval + if interval <= 0 { + interval = DefaultWatchdogInterval + } + debounce := opts.Debounce + if debounce <= 0 { + debounce = DefaultWatchdogDebounce + } + logger := opts.Logger + if logger == nil { + logger = slog.Default() + } + trigger := opts.Trigger + if trigger == nil { + trigger = func() {} + } + return &Watchdog{ + targets: opts.Targets, + liveness: opts.Liveness, + interval: interval, + debounce: debounce, + trigger: trigger, + logger: logger, + now: func() time.Time { return time.Now().UTC() }, + last: map[string]bool{}, + } +} + +// Run fast-polls until ctx is cancelled. The first tick establishes the baseline (no +// trigger); subsequent ticks detect transitions. Returns nil on ctx cancellation. +func (w *Watchdog) Run(ctx context.Context) error { + if w.targets == nil || w.liveness == nil { + w.logger.Info("storage: watchdog idle (no target source / liveness probe configured)") + <-ctx.Done() + return nil + } + w.logger.Info("storage: watchdog starting", "interval", w.interval, "debounce", w.debounce) + t := time.NewTicker(w.interval) + defer t.Stop() + w.tick(ctx) // immediate baseline + for { + select { + case <-ctx.Done(): + w.logger.Info("storage: watchdog shutting down", "reason", ctx.Err()) + return nil + case <-t.C: + w.tick(ctx) + } + } +} + +// tick performs one poll: read the known set, probe each target's liveness, diff against +// the last-seen state, and fire a debounced trigger on any transition for a SEEN target. +// It is deterministic given w.now — tests drive it directly with a fake clock. +func (w *Watchdog) tick(ctx context.Context) { + known, err := w.targets.Known(ctx) + if err != nil { + w.logger.Warn("storage: watchdog could not read known targets; skipping tick", "err", err) + return + } + + w.mu.Lock() + defer w.mu.Unlock() + + var transitions []Transition + current := make(map[string]bool, len(known)) + for _, k := range known { + present := w.liveness.Present(ctx, k) + current[k.Name] = present + prev, seen := w.last[k.Name] + if !seen { + continue // first observation → baseline only (never flag a never-attached drop) + } + if prev != present { + transitions = append(transitions, Transition{Name: k.Name, From: stateStr(prev), To: stateStr(present)}) + } + } + // Replace the baseline with the current snapshot (targets no longer known drop out). + w.last = current + + now := w.now() + if len(transitions) > 0 { + for _, tr := range transitions { + w.logger.Warn("storage: watchdog detected target state change", + "target", tr.Name, "from", tr.From, "to", tr.To) + } + if !w.fired || now.Sub(w.lastFire) >= w.debounce { + w.fire(now, len(transitions)) + } else { + w.pending = true + w.logger.Debug("storage: watchdog debouncing transition", "pending_until", w.lastFire.Add(w.debounce)) + } + return + } + // No new transition, but a debounced one is pending and the window has elapsed → fire. + if w.pending && now.Sub(w.lastFire) >= w.debounce { + w.fire(now, 0) + } +} + +// fire requests the out-of-band report and resets the debounce window. Called under w.mu. +func (w *Watchdog) fire(now time.Time, n int) { + w.lastFire = now + w.fired = true + w.pending = false + w.logger.Info("storage: watchdog triggering out-of-band host-report", "transitions", n) + w.trigger() +} + +func stateStr(present bool) string { + if present { + return "attached" + } + return "disconnected" +} + +// --- production liveness + a caching known-target source --- + +// HostLiveness is the production TargetLiveness: device + mount presence for local +// targets (the fast USB-drop signal) and a short reachability dial for network targets. +// All non-privileged. +type HostLiveness struct { + host HostReader + dialTimeout time.Duration + dial func(network, address string, timeout time.Duration) (net.Conn, error) +} + +// NewHostLiveness builds a HostLiveness over a HostReader. dialTimeout defaults to 3s. +func NewHostLiveness(host HostReader, dialTimeout time.Duration) *HostLiveness { + if host == nil { + host = NewProcHostReader() + } + if dialTimeout <= 0 { + dialTimeout = 3 * time.Second + } + return &HostLiveness{host: host, dialTimeout: dialTimeout, dial: net.DialTimeout} +} + +// Present probes one target without touching Proxmox. +func (h *HostLiveness) Present(ctx context.Context, t KnownTarget) bool { + if t.Network { + if t.ReachEndpoint == "" { + return true // can't probe → don't false-alarm; the 15-min cycle uses the active flag + } + conn, err := h.dial("tcp", t.ReachEndpoint, h.dialTimeout) + if err != nil { + return false + } + _ = conn.Close() + return true + } + if t.MountBacked { + // A mount-backed target (USB / extra disk) is present iff its mountpoint is an + // active mount AND the backing device node exists. + if !h.mounted(t.MountPath) { + return false + } + return t.BackingDevice == "" || h.host.DeviceExists(t.BackingDevice) + } + // Non-removable builtin targets (local/lvmthin): treated as present here — they don't + // "drop" without the whole host going down, which the heartbeat covers. + return true +} + +func (h *HostLiveness) mounted(path string) bool { + if path == "" { + return false + } + mounts, err := h.host.Mounts() + if err != nil { + return false + } + _, _, ok := exactMountDevice(mounts, path) + return ok +} + +// CachingKnownTargets wraps a slow KnownTargets source (the Observer, which hits Proxmox) +// with a TTL so the fast watchdog poll re-derives the known SET only every ttl, while +// still probing liveness every tick. A read error returns the last good set (so a +// transient Proxmox blip doesn't blank the watchdog's world). +type CachingKnownTargets struct { + src KnownTargets + ttl time.Duration + now func() time.Time + mu sync.Mutex + cached []KnownTarget + at time.Time + loaded bool +} + +// NewCachingKnownTargets wraps src, refreshing at most every ttl (default 60s). +func NewCachingKnownTargets(src KnownTargets, ttl time.Duration) *CachingKnownTargets { + if ttl <= 0 { + ttl = 60 * time.Second + } + return &CachingKnownTargets{src: src, ttl: ttl, now: func() time.Time { return time.Now().UTC() }} +} + +// Known returns the cached set, refreshing it when the TTL has elapsed. +func (c *CachingKnownTargets) Known(ctx context.Context) ([]KnownTarget, error) { + c.mu.Lock() + defer c.mu.Unlock() + now := c.now() + if c.loaded && now.Sub(c.at) < c.ttl { + return c.cached, nil + } + fresh, err := c.src.Known(ctx) + if err != nil { + if c.loaded { + return c.cached, nil // serve stale rather than blank on a transient error + } + return nil, err + } + c.cached, c.at, c.loaded = fresh, now, true + return c.cached, nil +} diff --git a/internal/storage/watchdog_test.go b/internal/storage/watchdog_test.go new file mode 100644 index 0000000..68f9f91 --- /dev/null +++ b/internal/storage/watchdog_test.go @@ -0,0 +1,251 @@ +package storage + +import ( + "context" + "errors" + "net" + "sync" + "testing" + "time" +) + +// staticKnown is a settable KnownTargets fake. +type staticKnown struct { + mu sync.Mutex + targets []KnownTarget + err error + calls int +} + +func (s *staticKnown) Known(context.Context) ([]KnownTarget, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.calls++ + return s.targets, s.err +} + +// mapLiveness is a settable per-target presence fake. +type mapLiveness struct { + mu sync.Mutex + present map[string]bool +} + +func (m *mapLiveness) set(name string, p bool) { + m.mu.Lock() + defer m.mu.Unlock() + m.present[name] = p +} +func (m *mapLiveness) Present(_ context.Context, t KnownTarget) bool { + m.mu.Lock() + defer m.mu.Unlock() + return m.present[t.Name] +} + +// newTestWatchdog builds a watchdog with a manual clock and a trigger counter. +func newTestWatchdog(known KnownTargets, live TargetLiveness, debounce time.Duration) (*Watchdog, *int, *time.Time) { + var fires int + clock := time.Unix(1_700_000_000, 0).UTC() + w := NewWatchdog(WatchdogOptions{ + Targets: known, + Liveness: live, + Trigger: func() { fires++ }, + Interval: time.Second, + Debounce: debounce, + Logger: quietLogger(), + }) + w.now = func() time.Time { return clock } + return w, &fires, &clock +} + +func TestWatchdog_BaselineThenDropTriggers(t *testing.T) { + known := &staticKnown{targets: []KnownTarget{{Name: "usb", MountBacked: true}}} + live := &mapLiveness{present: map[string]bool{"usb": true}} + w, fires, _ := newTestWatchdog(known, live, 30*time.Second) + ctx := context.Background() + + w.tick(ctx) // baseline: present, no trigger + if *fires != 0 { + t.Fatalf("baseline tick must not trigger, fires=%d", *fires) + } + live.set("usb", false) // drop + w.tick(ctx) + if *fires != 1 { + t.Fatalf("a known target drop must trigger an out-of-band report, fires=%d", *fires) + } +} + +func TestWatchdog_NeverAttachedNotFlagged(t *testing.T) { + // A defined-but-absent target (never seen present) must not be flagged on its absence. + known := &staticKnown{targets: []KnownTarget{{Name: "usb", MountBacked: true}}} + live := &mapLiveness{present: map[string]bool{"usb": false}} + w, fires, _ := newTestWatchdog(known, live, 30*time.Second) + ctx := context.Background() + + w.tick(ctx) // baseline absent + w.tick(ctx) // still absent + if *fires != 0 { + t.Fatalf("a never-attached target must not trigger, fires=%d", *fires) + } + // Now it appears (reconnect) → that IS a transition worth reporting. + live.set("usb", true) + w.tick(ctx) + if *fires != 1 { + t.Fatalf("attach transition should trigger, fires=%d", *fires) + } +} + +func TestWatchdog_DebounceCoalescesFlaps(t *testing.T) { + known := &staticKnown{targets: []KnownTarget{{Name: "usb", MountBacked: true}}} + live := &mapLiveness{present: map[string]bool{"usb": true}} + w, fires, clock := newTestWatchdog(known, live, 30*time.Second) + ctx := context.Background() + + w.tick(ctx) // baseline present + + // First drop fires immediately (leading edge). + live.set("usb", false) + w.tick(ctx) + if *fires != 1 { + t.Fatalf("first drop should fire, fires=%d", *fires) + } + + // Flap within the debounce window: re-attach then drop again — suppressed (pending). + *clock = clock.Add(5 * time.Second) + live.set("usb", true) + w.tick(ctx) + *clock = clock.Add(5 * time.Second) + live.set("usb", false) + w.tick(ctx) + if *fires != 1 { + t.Fatalf("flaps within the debounce window must be coalesced, fires=%d", *fires) + } + + // After the window elapses, the pending change fires (trailing edge), even with no new + // transition this tick. + *clock = clock.Add(30 * time.Second) + w.tick(ctx) + if *fires != 2 { + t.Fatalf("a pending change must fire once the window elapses, fires=%d", *fires) + } +} + +func TestWatchdog_ReadErrorSkipsTick(t *testing.T) { + known := &staticKnown{err: errors.New("proxmox blip")} + live := &mapLiveness{present: map[string]bool{}} + w, fires, _ := newTestWatchdog(known, live, time.Second) + w.tick(context.Background()) + if *fires != 0 { + t.Fatalf("a known-target read error must not trigger, fires=%d", *fires) + } +} + +func TestWatchdog_RunBaselinesAndStops(t *testing.T) { + // Smoke test of the goroutine wiring under -race: Run establishes a baseline and exits + // cleanly on ctx cancellation. + known := &staticKnown{targets: []KnownTarget{{Name: "usb", MountBacked: true}}} + live := &mapLiveness{present: map[string]bool{"usb": true}} + w, _, _ := newTestWatchdog(known, live, time.Second) + w.now = func() time.Time { return time.Now().UTC() } + w.interval = 5 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + time.Sleep(30 * time.Millisecond) + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("Run returned %v, want nil on cancel", err) + } + case <-time.After(time.Second): + t.Fatal("watchdog did not stop on cancel") + } +} + +func TestCachingKnownTargets_RefreshesOnTTL(t *testing.T) { + src := &staticKnown{targets: []KnownTarget{{Name: "a"}}} + clock := time.Unix(1_700_000_000, 0).UTC() + c := NewCachingKnownTargets(src, 60*time.Second) + c.now = func() time.Time { return clock } + ctx := context.Background() + + if _, err := c.Known(ctx); err != nil { + t.Fatal(err) + } + if _, err := c.Known(ctx); err != nil { // within TTL → cached + t.Fatal(err) + } + if src.calls != 1 { + t.Fatalf("within TTL the source must be hit once, calls=%d", src.calls) + } + clock = clock.Add(61 * time.Second) // past TTL + if _, err := c.Known(ctx); err != nil { + t.Fatal(err) + } + if src.calls != 2 { + t.Fatalf("past TTL the source must refresh, calls=%d", src.calls) + } +} + +func TestCachingKnownTargets_ServesStaleOnError(t *testing.T) { + src := &staticKnown{targets: []KnownTarget{{Name: "a"}}} + clock := time.Unix(1_700_000_000, 0).UTC() + c := NewCachingKnownTargets(src, 1*time.Second) + c.now = func() time.Time { return clock } + ctx := context.Background() + + if _, err := c.Known(ctx); err != nil { // prime the cache + t.Fatal(err) + } + clock = clock.Add(2 * time.Second) + src.mu.Lock() + src.err = errors.New("blip") + src.mu.Unlock() + got, err := c.Known(ctx) + if err != nil { + t.Fatalf("a transient error must serve stale, got err=%v", err) + } + if len(got) != 1 || got[0].Name != "a" { + t.Fatalf("stale set not served: %+v", got) + } +} + +func TestHostLiveness_MountBackedPresence(t *testing.T) { + host := &fakeHostReader{ + mounts: []Mount{{Device: "/dev/sdb1", MountPoint: "/mnt/usb", FSType: "ext4"}}, + exists: map[string]bool{"/dev/sdb1": true}, + } + hl := NewHostLiveness(host, time.Second) + tgt := KnownTarget{Name: "usb", MountBacked: true, MountPath: "/mnt/usb", BackingDevice: "/dev/sdb1"} + if !hl.Present(context.Background(), tgt) { + t.Error("mounted device should be present") + } + + // Unmount it: no exact mount entry → absent. + host.mounts = []Mount{{Device: "/dev/mapper/root", MountPoint: "/", FSType: "ext4"}} + if hl.Present(context.Background(), tgt) { + t.Error("unmounted device should be absent") + } +} + +func TestHostLiveness_NetworkDial(t *testing.T) { + hl := NewHostLiveness(&fakeHostReader{}, time.Second) + var dialed string + hl.dial = func(network, addr string, _ time.Duration) (net.Conn, error) { + dialed = addr + return nil, errors.New("refused") + } + tgt := KnownTarget{Name: "nfs", Network: true, ReachEndpoint: "10.0.0.5:2049"} + if hl.Present(context.Background(), tgt) { + t.Error("a refused dial should report not-present") + } + if dialed != "10.0.0.5:2049" { + t.Errorf("dialed %q, want 10.0.0.5:2049", dialed) + } + + // No endpoint to probe → don't false-alarm (the slow cycle uses the active flag). + if !hl.Present(context.Background(), KnownTarget{Name: "x", Network: true}) { + t.Error("network target without endpoint must not be flagged down") + } +}