F20-BUG3: run mkfs detached (survives request deadline + agent restart); v0.31.0
The format ran mkfs under the HTTP request context, so the controller's 15s client timeout cancelled it → SIGKILL mid-write → corrupt disk. Now mkfs runs DETACHED off s.baseCtx (a dropped request can't kill it) via a persisted formatJob record; the handler still waits to return the synchronous result (backward-compatible with the v0.62.0 controller) but abandoning the wait on client-disconnect leaves the mkfs running to completion. New GET /disks/format/status surfaces the job (additive). RecoverFormatJob runs on agent startup: a record left 'running' (agent died mid-format) is re-resolved by durable-id (anti-retarget — absent/swapped disk NOT re-formatted) and the mkfs re-run; a blank/path-bound interrupted format is marked failed (retry), never auto-re-run. Tests: detached run persists running→done + binds durable-id; status endpoint; recovery re-runs an interrupted durable-id-bound format; skips blank; skips unresolvable durable-id. Version 0.30.0 → 0.31.0.
This commit is contained in:
@@ -43,7 +43,7 @@ import (
|
|||||||
|
|
||||||
// version is the agent version. Overridable at build time with
|
// version is the agent version. Overridable at build time with
|
||||||
// -ldflags "-X main.version=<v>"; defaults to the in-repo CHANGELOG version.
|
// -ldflags "-X main.version=<v>"; defaults to the in-repo CHANGELOG version.
|
||||||
var version = "0.30.0"
|
var version = "0.31.0"
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
var (
|
var (
|
||||||
@@ -347,6 +347,13 @@ func runDaemon(cfg config.Config, logger *slog.Logger) int {
|
|||||||
logger.Warn("storage: guest-bind store unavailable — startup bind re-assert disabled (F9)", "err", gbErr)
|
logger.Warn("storage: guest-bind store unavailable — startup bind re-assert disabled (F9)", "err", gbErr)
|
||||||
guestBindStore = nil
|
guestBindStore = nil
|
||||||
}
|
}
|
||||||
|
// F20-BUG3: persisted disk-format job — lets mkfs run detached from the request and survive an agent
|
||||||
|
// restart (RecoverFormatJob re-runs an interrupted durable-id-bound format).
|
||||||
|
formatJobStore, fjErr := localapi.OpenFormatJobStore(filepath.Join(intentStateDir, "format-job.json"))
|
||||||
|
if fjErr != nil {
|
||||||
|
logger.Warn("storage: format-job store unavailable — format restart-recovery disabled (F20-BUG3)", "err", fjErr)
|
||||||
|
formatJobStore = nil
|
||||||
|
}
|
||||||
watchdog := storage.NewWatchdog(storage.WatchdogOptions{
|
watchdog := storage.NewWatchdog(storage.WatchdogOptions{
|
||||||
Targets: storage.NewCachingKnownTargets(observer, cfg.Storage.KnownRefresh()),
|
Targets: storage.NewCachingKnownTargets(observer, cfg.Storage.KnownRefresh()),
|
||||||
Liveness: storage.NewHostLiveness(hostReader, 0),
|
Liveness: storage.NewHostLiveness(hostReader, 0),
|
||||||
@@ -431,7 +438,7 @@ func runDaemon(cfg config.Config, logger *slog.Logger) int {
|
|||||||
jobsRunner := signedjobs.NewRunner(client, gate, signedjobs.ExecutorChain{wipeExec, decommExec}, cfg.Hub.HostID, logger)
|
jobsRunner := signedjobs.NewRunner(client, gate, signedjobs.ExecutorChain{wipeExec, decommExec}, cfg.Hub.HostID, logger)
|
||||||
loop.SetEnvelopeObserver(hub.MultiObserver(desiredSyncer, jobsRunner))
|
loop.SetEnvelopeObserver(hub.MultiObserver(desiredSyncer, jobsRunner))
|
||||||
|
|
||||||
localSrv := buildLocalAPIServer(cfg, px, backupStore, observer, hostOps, gate, collector, intentRec, guestBindStore, logger, &localTokens)
|
localSrv := buildLocalAPIServer(cfg, px, backupStore, observer, hostOps, gate, collector, intentRec, guestBindStore, formatJobStore, logger, &localTokens)
|
||||||
if localTokens != nil {
|
if localTokens != nil {
|
||||||
defer localTokens.Close()
|
defer localTokens.Close()
|
||||||
}
|
}
|
||||||
@@ -473,6 +480,8 @@ func runDaemon(cfg config.Config, logger *slog.Logger) int {
|
|||||||
// bind that a re-provision dropped — before serving, so the drive is back in the guest config
|
// bind that a re-provision dropped — before serving, so the drive is back in the guest config
|
||||||
// (activates on the guest's next reboot). On-durable-id-match; absent/swapped drives are skipped.
|
// (activates on the guest's next reboot). On-durable-id-match; absent/swapped drives are skipped.
|
||||||
localSrv.ReassertGuestBinds(ctx)
|
localSrv.ReassertGuestBinds(ctx)
|
||||||
|
// F20-BUG3: complete a disk format that an agent restart interrupted (durable-id-bound, re-resolved).
|
||||||
|
localSrv.RecoverFormatJob(ctx)
|
||||||
go func() { errc <- localSrv.Run(ctx) }()
|
go func() { errc <- localSrv.Run(ctx) }()
|
||||||
}
|
}
|
||||||
if lanLoop != nil {
|
if lanLoop != nil {
|
||||||
@@ -587,7 +596,7 @@ func buildRestoreTestScheduler(cfg config.Config, px *proxmox.Client, engine *re
|
|||||||
// leaf (stable fingerprint). Any failure DISABLES the server (returns nil) WITHOUT crashing the
|
// leaf (stable fingerprint). Any failure DISABLES the server (returns nil) WITHOUT crashing the
|
||||||
// daemon — the host still reports/reconciles; only the controller channel is unavailable until
|
// daemon — the host still reports/reconciles; only the controller channel is unavailable until
|
||||||
// fixed. The opened token store is returned via outTokens so the caller can Close it.
|
// fixed. The opened token store is returned via outTokens so the caller can Close it.
|
||||||
func buildLocalAPIServer(cfg config.Config, px *proxmox.Client, store *backup.Store, observer *storage.Observer, hostOps storage.HostOps, gate *reconcile.Gate, collector *hub.Collector, intent localapi.IntentRecorder, guestBinds *localapi.GuestBindStore, logger *slog.Logger, outTokens **localapi.TokenStore) *localapi.Server {
|
func buildLocalAPIServer(cfg config.Config, px *proxmox.Client, store *backup.Store, observer *storage.Observer, hostOps storage.HostOps, gate *reconcile.Gate, collector *hub.Collector, intent localapi.IntentRecorder, guestBinds *localapi.GuestBindStore, formatJobs *localapi.FormatJobStore, logger *slog.Logger, outTokens **localapi.TokenStore) *localapi.Server {
|
||||||
if !cfg.LocalAPI.Enabled() {
|
if !cfg.LocalAPI.Enabled() {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -632,6 +641,7 @@ func buildLocalAPIServer(cfg config.Config, px *proxmox.Client, store *backup.St
|
|||||||
GuestAttach: guestBinder, // slice 10 P2: bind enrolled data drives into the guest
|
GuestAttach: guestBinder, // slice 10 P2: bind enrolled data drives into the guest
|
||||||
Intent: intent, // slice 10 P3: record enroll/eject intent for self-heal
|
Intent: intent, // slice 10 P3: record enroll/eject intent for self-heal
|
||||||
GuestBinds: guestBinds, // F9: per-guest bind record for the startup re-assert
|
GuestBinds: guestBinds, // F9: per-guest bind record for the startup re-assert
|
||||||
|
FormatJobs: formatJobs, // F20-BUG3: detached-format job record + restart recovery
|
||||||
|
|
||||||
// Host metrics (slice 9): the shared collector serves GET /host/metrics — a fresh host +
|
// Host metrics (slice 9): the shared collector serves GET /host/metrics — a fresh host +
|
||||||
// per-storage view to the customer's monitoring page (reuses the slice-4 collector).
|
// per-storage view to the customer's monitoring page (reuses the slice-4 collector).
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package localapi
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -425,6 +426,44 @@ type PendingOp struct {
|
|||||||
FSType string `json:"fstype"` // the filesystem to mkfs after the wipe
|
FSType string `json:"fstype"` // the filesystem to mkfs after the wipe
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// errFormatClientGone signals the request context was cancelled (client/controller deadline) while the
|
||||||
|
// detached mkfs keeps running — the handler returns without writing; the job record records the outcome.
|
||||||
|
var errFormatClientGone = fmt.Errorf("format client gone (mkfs continues detached)")
|
||||||
|
|
||||||
|
// awaitFormat waits for the detached mkfs result, OR returns errFormatClientGone if the request context
|
||||||
|
// is cancelled first. Crucially the mkfs itself runs off s.baseCtx, so a cancelled request never kills it
|
||||||
|
// (F20-BUG3) — abandoning the wait here only abandons the HTTP response, not the format.
|
||||||
|
func (s *Server) awaitFormat(reqCtx context.Context, done <-chan error, vmid int, device string) error {
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
return err
|
||||||
|
case <-reqCtx.Done():
|
||||||
|
s.logger.Warn("local-api: format client disconnected — mkfs continues detached (poll GET /disks/format/status)",
|
||||||
|
"vmid", vmid, "device", device)
|
||||||
|
return errFormatClientGone
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// handleDiskFormatStatus reports the most-recent/in-flight format job (F20-BUG3), so a controller whose
|
||||||
|
// request timed out (or that reconnects after an agent restart) can learn the real outcome instead of
|
||||||
|
// assuming failure. Self-scoped (benign read).
|
||||||
|
func (s *Server) handleDiskFormatStatus(w http.ResponseWriter, r *http.Request, vmid int) {
|
||||||
|
if s.formatJobs == nil {
|
||||||
|
writeOK(w, map[string]any{"vmid": vmid, "phase": "idle"})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
job := s.formatJobs.get()
|
||||||
|
if job == nil {
|
||||||
|
writeOK(w, map[string]any{"vmid": vmid, "phase": "idle"})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeOK(w, map[string]any{
|
||||||
|
"vmid": vmid, "phase": job.Phase, "device": job.Device, "fstype": job.FSType,
|
||||||
|
"durable_id": job.DurableID, "error": job.Error, "started_at": job.StartedAt, "updated_at": job.UpdatedAt,
|
||||||
|
"job_id": job.JobID,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
// handleDiskFormat is the security centerpiece. The agent INSPECTS the device; if it is
|
// handleDiskFormat is the security centerpiece. The agent INSPECTS the device; if it is
|
||||||
// data-bearing it is classified destructive and the gate refuses it `pending_signature` — the
|
// data-bearing it is classified destructive and the gate refuses it `pending_signature` — the
|
||||||
// caller's claim is never trusted. Only a device the agent itself reads as blank is formatted.
|
// caller's claim is never trusted. Only a device the agent itself reads as blank is formatted.
|
||||||
@@ -456,8 +495,14 @@ func (s *Server) handleDiskFormat(w http.ResponseWriter, r *http.Request, vmid i
|
|||||||
// inspect error → fail-safe data-bearing (probe.DataBearing() is true on !Probed)
|
// inspect error → fail-safe data-bearing (probe.DataBearing() is true on !Probed)
|
||||||
}
|
}
|
||||||
if !probe.DataBearing() {
|
if !probe.DataBearing() {
|
||||||
// Blank device → benign → mkfs (role is irrelevant; there is nothing to destroy).
|
// Blank device → benign → mkfs (role is irrelevant; there is nothing to destroy). F20-BUG3: run
|
||||||
if err := s.disks.Format(r.Context(), req.Device, req.FSType); err != nil {
|
// it DETACHED off s.baseCtx so a request/client deadline can't SIGKILL mkfs mid-write; we still
|
||||||
|
// wait here to return the synchronous result (backward-compatible with the controller's client).
|
||||||
|
done := s.startFormatDetached(req.Device, "", req.FSType)
|
||||||
|
if err := s.awaitFormat(r.Context(), done, vmid, req.Device); err != nil {
|
||||||
|
if err == errFormatClientGone {
|
||||||
|
return // client gone; mkfs continues detached + the job record records the outcome
|
||||||
|
}
|
||||||
s.logger.Error("local-api: format", "vmid", vmid, "device", req.Device, "err", err)
|
s.logger.Error("local-api: format", "vmid", vmid, "device", req.Device, "err", err)
|
||||||
writeErr(w, http.StatusBadGateway, "format failed: "+err.Error())
|
writeErr(w, http.StatusBadGateway, "format failed: "+err.Error())
|
||||||
return
|
return
|
||||||
@@ -496,7 +541,14 @@ func (s *Server) handleDiskFormat(w http.ResponseWriter, r *http.Request, vmid i
|
|||||||
"wipe refused (device may have changed since confirmation): "+rerr.Error())
|
"wipe refused (device may have changed since confirmation): "+rerr.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if err := s.disks.Format(r.Context(), device, req.FSType); err != nil {
|
// F20-BUG3: run the destructive mkfs DETACHED off s.baseCtx (bound durable id recorded for
|
||||||
|
// restart-recovery), so a request/client deadline can never SIGKILL it mid-write and corrupt the
|
||||||
|
// disk. We still wait to return the synchronous result (backward-compatible with the controller).
|
||||||
|
done := s.startFormatDetached(device, deviceDurable, req.FSType)
|
||||||
|
if err := s.awaitFormat(r.Context(), done, vmid, device); err != nil {
|
||||||
|
if err == errFormatClientGone {
|
||||||
|
return // client gone; the wipe continues detached + survives a restart via the job record
|
||||||
|
}
|
||||||
s.logger.Error("local-api: customer-confirmed format", "vmid", vmid, "device", device, "err", err)
|
s.logger.Error("local-api: customer-confirmed format", "vmid", vmid, "device", device, "err", err)
|
||||||
writeErr(w, http.StatusBadGateway, "format failed: "+err.Error())
|
writeErr(w, http.StatusBadGateway, "format failed: "+err.Error())
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -0,0 +1,174 @@
|
|||||||
|
package localapi
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// formatJob is the persisted record of the most recent / in-flight disk format (F20-BUG3). It exists so
|
||||||
|
// that (a) an mkfs runs DETACHED from the HTTP request — a client/request deadline can no longer SIGKILL
|
||||||
|
// an in-progress mkfs and corrupt the disk — and (b) an in-progress format SURVIVES an agent restart:
|
||||||
|
// on boot, a record still in `running` is re-resolved by durable-id and re-run (mkfs is idempotent).
|
||||||
|
type formatJob struct {
|
||||||
|
JobID string `json:"job_id"`
|
||||||
|
Device string `json:"device"`
|
||||||
|
DurableID string `json:"durable_id"` // "" for a blank (benign) format — never auto-recovered
|
||||||
|
FSType string `json:"fstype"`
|
||||||
|
Phase string `json:"phase"` // running | done | failed
|
||||||
|
Error string `json:"error,omitempty"`
|
||||||
|
StartedAt string `json:"started_at"`
|
||||||
|
UpdatedAt string `json:"updated_at"`
|
||||||
|
}
|
||||||
|
|
||||||
|
const (
|
||||||
|
formatPhaseRunning = "running"
|
||||||
|
formatPhaseDone = "done"
|
||||||
|
formatPhaseFailed = "failed"
|
||||||
|
)
|
||||||
|
|
||||||
|
// FormatJobStore persists the single most-recent format job (atomic tmp+rename, 0600). One slot: a host
|
||||||
|
// formats one device at a time. Mirrors storage.IntentStore.
|
||||||
|
type FormatJobStore struct {
|
||||||
|
path string
|
||||||
|
mu sync.Mutex
|
||||||
|
cur *formatJob
|
||||||
|
}
|
||||||
|
|
||||||
|
// OpenFormatJobStore loads (or initializes) the store. Missing file = no job; corrupt file = error.
|
||||||
|
func OpenFormatJobStore(path string) (*FormatJobStore, error) {
|
||||||
|
s := &FormatJobStore{path: path}
|
||||||
|
data, err := os.ReadFile(path)
|
||||||
|
if err != nil {
|
||||||
|
if os.IsNotExist(err) {
|
||||||
|
return s, nil
|
||||||
|
}
|
||||||
|
return nil, fmt.Errorf("format-job store: read %s: %w", path, err)
|
||||||
|
}
|
||||||
|
if len(data) > 0 {
|
||||||
|
var j formatJob
|
||||||
|
if err := json.Unmarshal(data, &j); err != nil {
|
||||||
|
return nil, fmt.Errorf("format-job store: parse %s: %w", path, err)
|
||||||
|
}
|
||||||
|
s.cur = &j
|
||||||
|
}
|
||||||
|
return s, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *FormatJobStore) get() *formatJob {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
if s.cur == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
cp := *s.cur
|
||||||
|
return &cp
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *FormatJobStore) save(j *formatJob) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
cp := *j
|
||||||
|
s.cur = &cp
|
||||||
|
data, err := json.MarshalIndent(s.cur, "", " ")
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
tmp := s.path + ".tmp"
|
||||||
|
if err := os.WriteFile(tmp, data, 0o600); err != nil {
|
||||||
|
os.Remove(tmp)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return os.Rename(tmp, s.path)
|
||||||
|
}
|
||||||
|
|
||||||
|
// startFormatDetached persists a `running` record and runs mkfs in a goroutine off s.baseCtx (NOT the
|
||||||
|
// request context), with a long bound. It returns a channel that yields the format error (nil on
|
||||||
|
// success). The caller may stop waiting (client disconnect) without killing the mkfs — the goroutine
|
||||||
|
// runs to completion and records the outcome. device is the ALREADY anti-retarget-resolved device; the
|
||||||
|
// record carries durableID so a restart can re-resolve + re-run.
|
||||||
|
func (s *Server) startFormatDetached(device, durableID, fstype string) <-chan error {
|
||||||
|
base := s.baseCtx
|
||||||
|
if base == nil {
|
||||||
|
base = context.Background()
|
||||||
|
}
|
||||||
|
job := &formatJob{
|
||||||
|
JobID: s.nowFn().UTC().Format("20060102T150405Z"), Device: device, DurableID: durableID,
|
||||||
|
FSType: fstype, Phase: formatPhaseRunning,
|
||||||
|
StartedAt: s.nowFn().UTC().Format(time.RFC3339), UpdatedAt: s.nowFn().UTC().Format(time.RFC3339),
|
||||||
|
}
|
||||||
|
if s.formatJobs != nil {
|
||||||
|
if err := s.formatJobs.save(job); err != nil {
|
||||||
|
s.logger.Warn("format-job: could not persist running record (format will still run)", "device", device, "err", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
ctx, cancel := context.WithTimeout(base, 60*time.Minute)
|
||||||
|
defer cancel()
|
||||||
|
err := s.disks.Format(ctx, device, fstype)
|
||||||
|
s.finishFormatJob(job, err)
|
||||||
|
done <- err
|
||||||
|
}()
|
||||||
|
return done
|
||||||
|
}
|
||||||
|
|
||||||
|
// finishFormatJob updates the persisted record to done/failed.
|
||||||
|
func (s *Server) finishFormatJob(job *formatJob, err error) {
|
||||||
|
if s.formatJobs == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
job.UpdatedAt = s.nowFn().UTC().Format(time.RFC3339)
|
||||||
|
if err != nil {
|
||||||
|
job.Phase = formatPhaseFailed
|
||||||
|
job.Error = err.Error()
|
||||||
|
} else {
|
||||||
|
job.Phase = formatPhaseDone
|
||||||
|
}
|
||||||
|
if serr := s.formatJobs.save(job); serr != nil {
|
||||||
|
s.logger.Warn("format-job: could not persist final record", "device", job.Device, "phase", job.Phase, "err", serr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// RecoverFormatJob, on agent startup, completes a format that an agent restart interrupted (F20-BUG3 +
|
||||||
|
// the operator's "survive a restart" decision). A record still in `running` is re-resolved by its
|
||||||
|
// durable-id (anti-retarget — a swapped/absent disk is NOT re-formatted) and the mkfs is re-run detached
|
||||||
|
// (mkfs is idempotent). A blank (no-durable-id) format is NOT auto-re-run — it is marked failed for the
|
||||||
|
// caller to retry, so recovery never formats a mutable /dev path.
|
||||||
|
func (s *Server) RecoverFormatJob(ctx context.Context) {
|
||||||
|
if s.formatJobs == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
job := s.formatJobs.get()
|
||||||
|
if job == nil || job.Phase != formatPhaseRunning {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if job.DurableID == "" {
|
||||||
|
s.logger.Warn("format-job recover: interrupted blank format — marking failed (retry needed; not auto-re-running a path-bound format)", "device", job.Device)
|
||||||
|
s.finishFormatJob(job, fmt.Errorf("interrupted by agent restart; retry the format"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
device, err := s.reresolveWipe(ctx, job.DurableID)
|
||||||
|
if err != nil {
|
||||||
|
s.logger.Warn("format-job recover: durable-id no longer resolves — NOT re-formatting (anti-retarget)", "durable_id", job.DurableID, "err", err)
|
||||||
|
s.finishFormatJob(job, fmt.Errorf("durable-id %s did not re-resolve after restart: %w", job.DurableID, err))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
s.logger.Warn("format-job recover: re-running interrupted format detached", "durable_id", job.DurableID, "device", device, "fstype", job.FSType)
|
||||||
|
_ = s.startFormatDetached(device, job.DurableID, job.FSType) // detached; updates the record on completion
|
||||||
|
}
|
||||||
|
|
||||||
|
// nowFn returns the server clock (testable), defaulting to time.Now.
|
||||||
|
func (s *Server) nowFn() time.Time {
|
||||||
|
if s.now != nil {
|
||||||
|
return s.now()
|
||||||
|
}
|
||||||
|
return time.Now()
|
||||||
|
}
|
||||||
@@ -0,0 +1,168 @@
|
|||||||
|
package localapi
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"io"
|
||||||
|
"log/slog"
|
||||||
|
"net/http"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.dooplex.hu/admin/felhom-agent/internal/storage"
|
||||||
|
)
|
||||||
|
|
||||||
|
func tempFormatStore(t *testing.T) *FormatJobStore {
|
||||||
|
t.Helper()
|
||||||
|
fj, err := OpenFormatJobStore(filepath.Join(t.TempDir(), "format-job.json"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
return fj
|
||||||
|
}
|
||||||
|
|
||||||
|
// formatServer builds a *Server with the destructive-format path wired (data-bearing probe, a gate, the
|
||||||
|
// format-job store), plus the test stubs (reresolveWipe/deviceDurableID) so no real /dev is touched.
|
||||||
|
func formatServer(t *testing.T, d *fakeDiskOps, g StorageGate, fj *FormatJobStore) *Server {
|
||||||
|
t.Helper()
|
||||||
|
srv, err := NewServer(Options{
|
||||||
|
ListenAddr: "127.0.0.1:0", Guests: &fakeGuests{}, Backups: &fakeBackups{}, Store: &fakeStore{},
|
||||||
|
Storage: fakeStorage{}, Tokens: staticTokens{"A": 8200},
|
||||||
|
Disks: d, DiskGate: g, FormatJobs: fj, HostReader: sysOnSDA(),
|
||||||
|
Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
srv.baseCtx = context.Background()
|
||||||
|
srv.reresolveWipe = func(_ context.Context, _ string) (string, error) { return "/dev/sdb", nil }
|
||||||
|
srv.deviceDurableID = func(device string) (string, error) { return "byid:wwn-" + device, nil }
|
||||||
|
return srv
|
||||||
|
}
|
||||||
|
|
||||||
|
func waitFormatPhase(t *testing.T, fj *FormatJobStore, want string) *formatJob {
|
||||||
|
t.Helper()
|
||||||
|
deadline := time.Now().Add(2 * time.Second)
|
||||||
|
for time.Now().Before(deadline) {
|
||||||
|
if j := fj.get(); j != nil && j.Phase == want {
|
||||||
|
return j
|
||||||
|
}
|
||||||
|
time.Sleep(10 * time.Millisecond)
|
||||||
|
}
|
||||||
|
got := fj.get()
|
||||||
|
t.Fatalf("format job did not reach phase %q in time (got %+v)", want, got)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFormat_DetachedPersistsJobRecord asserts F20-BUG3: a customer-confirmed wipe runs through the
|
||||||
|
// detached runner and a persisted job record reaches `done` with the device + durable-id bound (so it
|
||||||
|
// can be polled / recovered). Also confirms backward-compat: the handler still returns 200 synchronously.
|
||||||
|
func TestFormat_DetachedPersistsJobRecord(t *testing.T) {
|
||||||
|
d := &fakeDiskOps{probe: deviceProbeDataBearing()}
|
||||||
|
g := &fakeGate{decision: WipeDecision{Allowed: true, Tier: "customer_confirmable", Reason: "customer_confirmed"}}
|
||||||
|
fj := tempFormatStore(t)
|
||||||
|
srv := formatServer(t, d, g, fj)
|
||||||
|
h := srv.Handler()
|
||||||
|
|
||||||
|
w := do(t, h, "POST", "/disks/format", "A", `{"device":"/dev/sdb1","fstype":"ext4","confirmed":true,"durable_id":"byid:wwn-/dev/sdb1"}`)
|
||||||
|
if w.Code != http.StatusOK {
|
||||||
|
t.Fatalf("confirmed format: %d (%s)", w.Code, w.Body.String())
|
||||||
|
}
|
||||||
|
job := waitFormatPhase(t, fj, formatPhaseDone)
|
||||||
|
if job.Device != "/dev/sdb" { // the anti-retarget re-resolved device (stub returns /dev/sdb)
|
||||||
|
t.Fatalf("job device = %q, want /dev/sdb (re-resolved)", job.Device)
|
||||||
|
}
|
||||||
|
if job.DurableID == "" {
|
||||||
|
t.Fatalf("job durable-id empty — recovery could not re-bind")
|
||||||
|
}
|
||||||
|
if len(d.formatted()) != 1 {
|
||||||
|
t.Fatalf("mkfs called %d times, want 1", len(d.formatted()))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFormatStatus reports the persisted job via GET /disks/format/status.
|
||||||
|
func TestFormatStatus(t *testing.T) {
|
||||||
|
d := &fakeDiskOps{probe: deviceProbeDataBearing()}
|
||||||
|
g := &fakeGate{decision: WipeDecision{Allowed: true, Tier: "customer_confirmable", Reason: "customer_confirmed"}}
|
||||||
|
fj := tempFormatStore(t)
|
||||||
|
srv := formatServer(t, d, g, fj)
|
||||||
|
h := srv.Handler()
|
||||||
|
|
||||||
|
do(t, h, "POST", "/disks/format", "A", `{"device":"/dev/sdb1","fstype":"ext4","confirmed":true,"durable_id":"byid:wwn-/dev/sdb1"}`)
|
||||||
|
waitFormatPhase(t, fj, formatPhaseDone)
|
||||||
|
|
||||||
|
w := do(t, h, "GET", "/disks/format/status", "A", "")
|
||||||
|
if w.Code != http.StatusOK {
|
||||||
|
t.Fatalf("status: %d", w.Code)
|
||||||
|
}
|
||||||
|
var resp struct {
|
||||||
|
Data struct {
|
||||||
|
Phase string `json:"phase"`
|
||||||
|
Device string `json:"device"`
|
||||||
|
} `json:"data"`
|
||||||
|
}
|
||||||
|
_ = json.Unmarshal(w.Body.Bytes(), &resp)
|
||||||
|
if resp.Data.Phase != formatPhaseDone {
|
||||||
|
t.Fatalf("status phase = %q, want done", resp.Data.Phase)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRecoverFormatJob_ReRunsInterrupted asserts the operator's "survive an agent restart" decision: a
|
||||||
|
// record left in `running` (agent died mid-format) is re-resolved by durable-id and the mkfs re-run.
|
||||||
|
func TestRecoverFormatJob_ReRunsInterrupted(t *testing.T) {
|
||||||
|
d := &fakeDiskOps{probe: deviceProbeDataBearing()}
|
||||||
|
fj := tempFormatStore(t)
|
||||||
|
// Pre-seed an interrupted, durable-id-bound running job (as if the agent died mid-mkfs).
|
||||||
|
if err := fj.save(&formatJob{JobID: "j1", Device: "/dev/sdb", DurableID: "byid:wwn-x", FSType: "ext4", Phase: formatPhaseRunning}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
srv := formatServer(t, d, &fakeGate{}, fj)
|
||||||
|
|
||||||
|
srv.RecoverFormatJob(context.Background())
|
||||||
|
waitFormatPhase(t, fj, formatPhaseDone)
|
||||||
|
if got := d.formatted(); len(got) != 1 || got[0] != "/dev/sdb" {
|
||||||
|
t.Fatalf("recovery should have re-run mkfs on /dev/sdb (re-resolved), got %v", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRecoverFormatJob_SkipsBlank: an interrupted BLANK (no durable-id) format is NOT auto-re-run (never
|
||||||
|
// format a mutable /dev path on recovery) — it is marked failed for the caller to retry.
|
||||||
|
func TestRecoverFormatJob_SkipsBlank(t *testing.T) {
|
||||||
|
d := &fakeDiskOps{probe: deviceProbeDataBearing()}
|
||||||
|
fj := tempFormatStore(t)
|
||||||
|
_ = fj.save(&formatJob{JobID: "j2", Device: "/dev/sdb", DurableID: "", FSType: "ext4", Phase: formatPhaseRunning})
|
||||||
|
srv := formatServer(t, d, &fakeGate{}, fj)
|
||||||
|
|
||||||
|
srv.RecoverFormatJob(context.Background())
|
||||||
|
if got := d.formatted(); len(got) != 0 {
|
||||||
|
t.Fatalf("recovery must NOT re-run a path-bound (blank) format, but mkfs ran: %v", got)
|
||||||
|
}
|
||||||
|
if j := fj.get(); j == nil || j.Phase != formatPhaseFailed {
|
||||||
|
t.Fatalf("blank interrupted job should be marked failed, got %+v", j)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRecoverFormatJob_SkipsUnresolvable: if the durable-id no longer resolves (drive removed/replaced),
|
||||||
|
// recovery must NOT format anything (anti-retarget) and marks the job failed.
|
||||||
|
func TestRecoverFormatJob_SkipsUnresolvable(t *testing.T) {
|
||||||
|
d := &fakeDiskOps{probe: deviceProbeDataBearing()}
|
||||||
|
fj := tempFormatStore(t)
|
||||||
|
_ = fj.save(&formatJob{JobID: "j3", Device: "/dev/sdb", DurableID: "byid:wwn-gone", FSType: "ext4", Phase: formatPhaseRunning})
|
||||||
|
srv := formatServer(t, d, &fakeGate{}, fj)
|
||||||
|
srv.reresolveWipe = func(_ context.Context, _ string) (string, error) {
|
||||||
|
return "", context.DeadlineExceeded // simulate "durable-id no longer resolves"
|
||||||
|
}
|
||||||
|
|
||||||
|
srv.RecoverFormatJob(context.Background())
|
||||||
|
if got := d.formatted(); len(got) != 0 {
|
||||||
|
t.Fatalf("recovery must NOT format when the durable-id is unresolvable: %v", got)
|
||||||
|
}
|
||||||
|
if j := fj.get(); j == nil || j.Phase != formatPhaseFailed {
|
||||||
|
t.Fatalf("unresolvable interrupted job should be marked failed, got %+v", j)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func deviceProbeDataBearing() storage.DeviceProbe {
|
||||||
|
return storage.DeviceProbe{Probed: true, HasFilesystem: true, FSType: "ext4"}
|
||||||
|
}
|
||||||
@@ -91,6 +91,10 @@ type Options struct {
|
|||||||
// startup re-assert (ReassertGuestBinds) can restore a bind that a re-provision dropped (F9).
|
// startup re-assert (ReassertGuestBinds) can restore a bind that a re-provision dropped (F9).
|
||||||
// OPTIONAL — when nil, guest binds are not recorded and the startup re-assert is a no-op.
|
// OPTIONAL — when nil, guest binds are not recorded and the startup re-assert is a no-op.
|
||||||
GuestBinds *GuestBindStore
|
GuestBinds *GuestBindStore
|
||||||
|
// FormatJobs persists the in-flight/last disk-format job so mkfs runs detached from the request
|
||||||
|
// (F20-BUG3: a request deadline can't kill it) and survives an agent restart (RecoverFormatJob).
|
||||||
|
// OPTIONAL — when nil, formats still run detached but are not persisted/recovered.
|
||||||
|
FormatJobs *FormatJobStore
|
||||||
// HostReader is the root-free host topology reader used to classify a device/mount's protection
|
// HostReader is the root-free host topology reader used to classify a device/mount's protection
|
||||||
// ROLE (it backs SystemDisks for the eject role-gate + the /disks role hints). OPTIONAL — when nil
|
// ROLE (it backs SystemDisks for the eject role-gate + the /disks role hints). OPTIONAL — when nil
|
||||||
// it defaults to the production *storage.ProcHostReader. Injectable so the role-gate is testable.
|
// it defaults to the production *storage.ProcHostReader. Injectable so the role-gate is testable.
|
||||||
@@ -148,6 +152,7 @@ type Server struct {
|
|||||||
guestAttach GuestAttacher // slice 10 P2 (optional)
|
guestAttach GuestAttacher // slice 10 P2 (optional)
|
||||||
intent IntentRecorder // slice 10 P3 (optional)
|
intent IntentRecorder // slice 10 P3 (optional)
|
||||||
guestBinds *GuestBindStore // F9 startup bind re-assert record (optional)
|
guestBinds *GuestBindStore // F9 startup bind re-assert record (optional)
|
||||||
|
formatJobs *FormatJobStore // F20-BUG3 detached-format job record (optional)
|
||||||
host storage.HostReader // role classification source (optional; defaults to ProcHostReader)
|
host storage.HostReader // role classification source (optional; defaults to ProcHostReader)
|
||||||
|
|
||||||
hostMetrics HostMetricsProvider // slice 9 (optional)
|
hostMetrics HostMetricsProvider // slice 9 (optional)
|
||||||
@@ -204,6 +209,7 @@ func NewServer(o Options) (*Server, error) {
|
|||||||
guestAttach: o.GuestAttach,
|
guestAttach: o.GuestAttach,
|
||||||
intent: o.Intent,
|
intent: o.Intent,
|
||||||
guestBinds: o.GuestBinds,
|
guestBinds: o.GuestBinds,
|
||||||
|
formatJobs: o.FormatJobs,
|
||||||
host: o.HostReader,
|
host: o.HostReader,
|
||||||
hostMetrics: o.HostMetrics,
|
hostMetrics: o.HostMetrics,
|
||||||
hostID: o.HostID,
|
hostID: o.HostID,
|
||||||
@@ -232,6 +238,7 @@ func (s *Server) Handler() http.Handler {
|
|||||||
mux.HandleFunc("POST /disks/assign", s.withGuest(s.handleDiskAssign))
|
mux.HandleFunc("POST /disks/assign", s.withGuest(s.handleDiskAssign))
|
||||||
mux.HandleFunc("POST /disks/eject", s.withGuest(s.handleDiskEject))
|
mux.HandleFunc("POST /disks/eject", s.withGuest(s.handleDiskEject))
|
||||||
mux.HandleFunc("POST /disks/format", s.withGuest(s.handleDiskFormat))
|
mux.HandleFunc("POST /disks/format", s.withGuest(s.handleDiskFormat))
|
||||||
|
mux.HandleFunc("GET /disks/format/status", s.withGuest(s.handleDiskFormatStatus))
|
||||||
// Guest data-drive passthrough (slice 10 P2): bind an enrolled drive's felhom-data namespace in.
|
// Guest data-drive passthrough (slice 10 P2): bind an enrolled drive's felhom-data namespace in.
|
||||||
mux.HandleFunc("POST /disks/guest-attach", s.withGuest(s.handleDiskGuestAttach))
|
mux.HandleFunc("POST /disks/guest-attach", s.withGuest(s.handleDiskGuestAttach))
|
||||||
// Guest reboot (slice 10 P2 activation): user-triggered restart to activate pending drive binds.
|
// Guest reboot (slice 10 P2 activation): user-triggered restart to activate pending drive binds.
|
||||||
|
|||||||
Reference in New Issue
Block a user