diff --git a/internal/server/fleetupdate/worker.go b/internal/server/fleetupdate/worker.go index 1442832..2a53636 100644 --- a/internal/server/fleetupdate/worker.go +++ b/internal/server/fleetupdate/worker.go @@ -71,6 +71,16 @@ func NewWorker(st *store.Store, hub Hub, disp Dispatcher, alerts AlertRaiser) *W // worker goroutine. Returns the new fleet_update_id on success. // store.ErrFleetUpdateRunning bubbles up unchanged. func (w *Worker) Start(ctx context.Context, userID, targetVersion string, hostIDs []string) (string, error) { + return w.start(ctx, userID, targetVersion, hostIDs, false) +} + +// StartCanary starts the same sequential rollout but pauses after the first +// verified host so an operator can inspect it before starting the remainder. +func (w *Worker) StartCanary(ctx context.Context, userID, targetVersion string, hostIDs []string) (string, error) { + return w.start(ctx, userID, targetVersion, hostIDs, true) +} + +func (w *Worker) start(ctx context.Context, userID, targetVersion string, hostIDs []string, pauseAfterFirst bool) (string, error) { if userID == "" || targetVersion == "" { return "", errors.New("fleetupdate: userID and targetVersion required") } @@ -85,6 +95,7 @@ func (w *Worker) Start(ctx context.Context, userID, targetVersion string, hostID StartedByUserID: userID, TargetVersion: targetVersion, Status: "running", + PauseAfterFirst: pauseAfterFirst, }, hostIDs); err != nil { return "", err } @@ -137,6 +148,13 @@ func (w *Worker) run(ctx context.Context, fuID, userID, targetVersion string) { next := pending[0] w.processHost(ctx, fuID, userID, next) + if fu.PauseAfterFirst && next.Position == 0 { + updated, _, gerr := w.store.GetFleetUpdate(ctx, fuID) + if gerr == nil && updated.Status == "running" { + _ = w.store.HaltFleetUpdate(ctx, fuID, "canary succeeded; review the host, then resume remaining agents", time.Now().UTC()) + } + return + } } } @@ -195,7 +213,19 @@ func (w *Worker) processHost(ctx context.Context, fuID, userID string, slot stor return } } - reason := fmt.Sprintf("timeout waiting for %s to reach %s", hostID, w.targetVersion) + // One authoritative final read closes the poll/deadline race. AgentVersion + // is written directly from the reconnect hello handshake. + lastVersion := "unknown" + if h, err := w.store.GetHost(ctx, hostID); err == nil && h != nil { + if h.AgentVersion == w.targetVersion { + _ = w.store.SetFleetUpdateHostStatus(ctx, fuID, hostID, "succeeded", "", jobID) + return + } + if h.AgentVersion != "" { + lastVersion = h.AgentVersion + } + } + reason := fmt.Sprintf("timeout waiting for %s reconnect hello at %s (last reported %s)", hostID, w.targetVersion, lastVersion) _ = w.store.SetFleetUpdateHostStatus(ctx, fuID, hostID, "failed", reason, jobID) w.halt(ctx, fuID, reason) } diff --git a/internal/server/fleetupdate/worker_test.go b/internal/server/fleetupdate/worker_test.go index c1cdac1..4a36978 100644 --- a/internal/server/fleetupdate/worker_test.go +++ b/internal/server/fleetupdate/worker_test.go @@ -160,6 +160,35 @@ func TestWorkerTwoHostsBothSucceed(t *testing.T) { } } +func TestWorkerCanaryPausesAfterFirstVerifiedHost(t *testing.T) { + st := openStore(t) + uid := mustCreateAdmin(t, st) + h1 := mustCreateHost(t, st, "canary", "v0") + h2 := mustCreateHost(t, st, "remainder", "v0") + hub := &fakeHub{online: map[string]bool{h1: true, h2: true}} + disp := &fakeDispatcher{st: st, target: "v2", delayMS: 20} + alerts := &recAlert{} + w := NewWorker(st, hub, disp, alerts) + w.pollPeriod = 10 * time.Millisecond + w.hostTimeout = time.Second + + fuID, err := w.StartCanary(context.Background(), uid, "v2", []string{h1, h2}) + if err != nil { + t.Fatalf("start canary: %v", err) + } + fu := waitForStatus(t, st, fuID, "halted", 2*time.Second) + if fu.HaltedReason != "canary succeeded; review the host, then resume remaining agents" { + t.Fatalf("halt reason: %q", fu.HaltedReason) + } + _, hosts, _ := st.GetFleetUpdate(context.Background(), fuID) + if hosts[0].Status != "succeeded" || hosts[1].Status != "pending" { + t.Fatalf("canary statuses: %+v", hosts) + } + if len(alerts.reasons) != 0 { + t.Fatalf("successful canary pause must not alert: %v", alerts.reasons) + } +} + func TestWorkerSecondHostTimesOutHalts(t *testing.T) { st := openStore(t) uid := mustCreateAdmin(t, st) diff --git a/internal/server/http/fleet_update.go b/internal/server/http/fleet_update.go index 42c67f8..f70b03e 100644 --- a/internal/server/http/fleet_update.go +++ b/internal/server/http/fleet_update.go @@ -32,6 +32,13 @@ import ( type fleetUpdateStartReq struct { TargetVersion string `json:"target_version,omitempty"` HostIDs []string `json:"host_ids,omitempty"` + CanaryFirst bool `json:"canary_first,omitempty"` +} + +type fleetUpdateCandidate struct { + Host store.Host + Eligible bool + Reason string } // fleetUpdateHostView is one row in the JSON response for GET @@ -56,6 +63,7 @@ type fleetUpdateView struct { CurrentHostID string `json:"current_host_id,omitempty"` HaltedReason string `json:"halted_reason,omitempty"` CompletedAt *string `json:"completed_at,omitempty"` + CanaryFirst bool `json:"canary_first"` Hosts []fleetUpdateHostView `json:"hosts"` } @@ -65,6 +73,7 @@ type fleetUpdateView struct { type fleetUpdatePage struct { // Idle-state fields. OutOfDateHosts []store.Host // online hosts whose version != target + Candidates []fleetUpdateCandidate TargetVersion string // Active-state fields. Nil when no fleet update has ever run. @@ -100,6 +109,11 @@ func (s *Server) handleAPIFleetUpdateStart(w stdhttp.ResponseWriter, r *stdhttp. if target == "" { target = version.Version } + if target != version.Version { + writeJSONError(w, stdhttp.StatusUnprocessableEntity, "unsupported_target_version", + "fleet updates can only target the running server version") + return + } hostIDs := body.HostIDs if len(hostIDs) == 0 { derived, err := s.deriveOutOfDateOnlineHostIDs(r.Context(), target) @@ -108,6 +122,19 @@ func (s *Server) handleAPIFleetUpdateStart(w stdhttp.ResponseWriter, r *stdhttp. return } hostIDs = derived + } else { + validated, reasons, err := s.validateFleetUpdateHostIDs(r.Context(), target, hostIDs) + if err != nil { + writeJSONError(w, stdhttp.StatusInternalServerError, "internal", err.Error()) + return + } + if len(reasons) > 0 { + writeJSON(w, stdhttp.StatusUnprocessableEntity, map[string]any{ + "code": "ineligible_hosts", "message": "one or more selected hosts are ineligible", "reasons": reasons, + }) + return + } + hostIDs = validated } if len(hostIDs) == 0 { writeJSONError(w, stdhttp.StatusConflict, "no_hosts_eligible", @@ -115,7 +142,20 @@ func (s *Server) handleAPIFleetUpdateStart(w stdhttp.ResponseWriter, r *stdhttp. return } - fuID, err := s.deps.FleetWorker.Start(r.Context(), user.ID, target, hostIDs) + var fuID string + var err error + if body.CanaryFirst && len(hostIDs) > 1 { + worker, supported := s.deps.FleetWorker.(interface { + StartCanary(context.Context, string, string, []string) (string, error) + }) + if !supported { + writeJSONError(w, stdhttp.StatusServiceUnavailable, "canary_unavailable", "") + return + } + fuID, err = worker.StartCanary(r.Context(), user.ID, target, hostIDs) + } else { + fuID, err = s.deps.FleetWorker.Start(r.Context(), user.ID, target, hostIDs) + } if err != nil { if errors.Is(err, store.ErrFleetUpdateRunning) { writeJSONError(w, stdhttp.StatusConflict, "fleet_update_in_progress", err.Error()) @@ -129,6 +169,7 @@ func (s *Server) handleAPIFleetUpdateStart(w stdhttp.ResponseWriter, r *stdhttp. "fleet_update_id": fuID, "target_version": target, "host_count": len(hostIDs), + "canary_first": body.CanaryFirst, }) _ = s.deps.Store.AppendAudit(r.Context(), store.AuditEntry{ ID: ulid.Make().String(), UserID: &user.ID, Actor: "user", @@ -209,6 +250,7 @@ func (s *Server) handleAPIFleetUpdateGet(w stdhttp.ResponseWriter, r *stdhttp.Re Status: fu.Status, CurrentHostID: fu.CurrentHostID, HaltedReason: fu.HaltedReason, + CanaryFirst: fu.PauseAfterFirst, Hosts: make([]fleetUpdateHostView, 0, len(hosts)), } if fu.CompletedAt != nil { @@ -286,6 +328,27 @@ func (s *Server) buildFleetUpdatePage(r *stdhttp.Request) (fleetUpdatePage, erro } for _, h := range hosts { page.HostNames[h.ID] = h.Name + candidate := fleetUpdateCandidate{Host: h} + switch { + case h.AgentVersion == "": + candidate.Reason = "version unknown" + case h.AgentVersion == page.TargetVersion: + candidate.Reason = "already current" + case s.deps.Hub == nil || !s.deps.Hub.Connected(h.ID): + candidate.Reason = "offline" + default: + updating, uerr := s.deps.Store.RunningUpdateJobForHost(r.Context(), h.ID) + if uerr != nil { + return page, uerr + } + if updating != "" { + candidate.Reason = "update already running" + } else { + candidate.Eligible = true + page.OutOfDateHosts = append(page.OutOfDateHosts, h) + } + } + page.Candidates = append(page.Candidates, candidate) } active, err := s.deps.Store.ActiveFleetUpdate(r.Context()) @@ -328,20 +391,54 @@ func (s *Server) buildFleetUpdatePage(r *stdhttp.Request) (fleetUpdatePage, erro } } - // Idle list (or "still out of date" reference even when an active - // roll is running — cheap to compute, harmless to attach). - for _, h := range hosts { - if h.Status != "online" { - continue - } - if h.AgentVersion == "" || h.AgentVersion == page.TargetVersion { - continue - } - page.OutOfDateHosts = append(page.OutOfDateHosts, h) - } return page, nil } +// validateFleetUpdateHostIDs deduplicates an explicit selection while +// preserving review order and rejects the entire request if membership has +// changed or any selected host is not dispatchable. +func (s *Server) validateFleetUpdateHostIDs(ctx context.Context, target string, requested []string) ([]string, map[string]string, error) { + hosts, err := s.deps.Store.ListHosts(ctx) + if err != nil { + return nil, nil, err + } + byID := make(map[string]store.Host, len(hosts)) + for _, h := range hosts { + byID[h.ID] = h + } + seen := map[string]bool{} + validated := make([]string, 0, len(requested)) + reasons := map[string]string{} + for _, id := range requested { + if seen[id] { + continue + } + seen[id] = true + h, ok := byID[id] + switch { + case !ok: + reasons[id] = "host not found" + case h.AgentVersion == "": + reasons[id] = "agent version unknown" + case h.AgentVersion == target: + reasons[id] = "already at target version" + case s.deps.Hub == nil || !s.deps.Hub.Connected(id): + reasons[id] = "host offline" + default: + jobID, jerr := s.deps.Store.RunningUpdateJobForHost(ctx, id) + if jerr != nil { + return nil, nil, jerr + } + if jobID != "" { + reasons[id] = "update already in progress" + } else { + validated = append(validated, id) + } + } + } + return validated, reasons, nil +} + // deriveOutOfDateOnlineHostIDs returns the list of host IDs that // (a) are online (Hub.Connected) and (b) have an agent_version that's // non-empty AND != target. Used by the start endpoint when the caller diff --git a/internal/server/http/fleet_update_test.go b/internal/server/http/fleet_update_test.go index ca82561..fdba686 100644 --- a/internal/server/http/fleet_update_test.go +++ b/internal/server/http/fleet_update_test.go @@ -50,6 +50,10 @@ func (f *fakeFleetWorker) Start(_ context.Context, userID, target string, hostID return f.startID, nil } +func (f *fakeFleetWorker) StartCanary(ctx context.Context, userID, target string, hostIDs []string) (string, error) { + return f.Start(ctx, userID, target, hostIDs) +} + func (f *fakeFleetWorker) Cancel(_ context.Context, id string) error { f.mu.Lock() defer f.mu.Unlock() @@ -190,6 +194,67 @@ func TestFleetUpdateStartDerivesHostIDsWhenEmpty(t *testing.T) { } } +func TestFleetUpdateStartDeduplicatesExplicitSelection(t *testing.T) { + t.Parallel() + srv, ts, st := rawTestServer(t) + worker := &fakeFleetWorker{startID: ulid.Make().String()} + srv.deps.FleetWorker = worker + cookie := loginAsAdmin(t, st) + hostID := helloOnlineHost(t, srv, st, "duplicate-host", "v0") + raw, _ := json.Marshal(map[string]any{"host_ids": []string{hostID, hostID}}) + req, _ := stdhttp.NewRequest("POST", ts.URL+"/api/fleet/update", bytes.NewReader(raw)) + req.AddCookie(cookie) + req.Header.Set("Content-Type", "application/json") + res, err := stdhttp.DefaultClient.Do(req) + if err != nil { + t.Fatalf("do: %v", err) + } + defer res.Body.Close() + if res.StatusCode != stdhttp.StatusAccepted { + t.Fatalf("status: %d", res.StatusCode) + } + if got := worker.startCalls[0].HostIDs; len(got) != 1 || got[0] != hostID { + t.Fatalf("deduplicated ids: %v", got) + } +} + +func TestFleetUpdateStartRejectsUnknownAndIneligibleHosts(t *testing.T) { + t.Parallel() + srv, ts, st := rawTestServer(t) + worker := &fakeFleetWorker{startID: ulid.Make().String()} + srv.deps.FleetWorker = worker + cookie := loginAsAdmin(t, st) + offline := makeHost(t, st, "offline-host") + if err := st.MarkHostHello(context.Background(), offline, "v0", "0.17", api.CurrentProtocolVersion, time.Now().UTC()); err != nil { + t.Fatalf("mark offline host: %v", err) + } + raw, _ := json.Marshal(map[string]any{"host_ids": []string{offline, "does-not-exist"}}) + req, _ := stdhttp.NewRequest("POST", ts.URL+"/api/fleet/update", bytes.NewReader(raw)) + req.AddCookie(cookie) + req.Header.Set("Content-Type", "application/json") + res, err := stdhttp.DefaultClient.Do(req) + if err != nil { + t.Fatalf("do: %v", err) + } + defer res.Body.Close() + if res.StatusCode != stdhttp.StatusUnprocessableEntity { + t.Fatalf("status: got %d, want 422", res.StatusCode) + } + var body struct { + Code string `json:"code"` + Reasons map[string]string `json:"reasons"` + } + if err := json.NewDecoder(res.Body).Decode(&body); err != nil { + t.Fatalf("decode: %v", err) + } + if body.Code != "ineligible_hosts" || len(body.Reasons) != 2 { + t.Fatalf("structured reasons: %+v", body) + } + if len(worker.startCalls) != 0 { + t.Fatal("worker must not start with invalid membership") + } +} + func TestFleetUpdateCancelHappyPath(t *testing.T) { t.Parallel() srv, ts, st := rawTestServer(t) diff --git a/internal/store/fleet_updates.go b/internal/store/fleet_updates.go index ae9fec2..0207da0 100644 --- a/internal/store/fleet_updates.go +++ b/internal/store/fleet_updates.go @@ -42,9 +42,9 @@ func (st *Store) CreateFleetUpdate(ctx context.Context, fu FleetUpdate, hostIDs } if _, err := tx.ExecContext(ctx, - `INSERT INTO fleet_updates (id, started_at, started_by_user_id, target_version, status) - VALUES (?, ?, ?, ?, ?)`, - fu.ID, fu.StartedAt.UTC().Format(time.RFC3339Nano), fu.StartedByUserID, fu.TargetVersion, fu.Status, + `INSERT INTO fleet_updates (id, started_at, started_by_user_id, target_version, status, pause_after_first) + VALUES (?, ?, ?, ?, ?, ?)`, + fu.ID, fu.StartedAt.UTC().Format(time.RFC3339Nano), fu.StartedByUserID, fu.TargetVersion, fu.Status, fu.PauseAfterFirst, ); err != nil { return fmt.Errorf("store: insert fleet_updates: %w", err) } @@ -67,12 +67,13 @@ func (st *Store) ActiveFleetUpdate(ctx context.Context) (*FleetUpdate, error) { var current sql.NullString var halted sql.NullString var completedAt sql.NullString + var pauseAfterFirst int err := st.db.QueryRowContext(ctx, `SELECT id, started_at, started_by_user_id, target_version, status, - current_host_id, halted_reason, completed_at + current_host_id, halted_reason, completed_at, pause_after_first FROM fleet_updates WHERE status = 'running' LIMIT 1`). Scan(&fu.ID, &startedAt, &fu.StartedByUserID, &fu.TargetVersion, &fu.Status, - ¤t, &halted, &completedAt) + ¤t, &halted, &completedAt, &pauseAfterFirst) if errors.Is(err, sql.ErrNoRows) { return nil, nil } @@ -82,6 +83,7 @@ func (st *Store) ActiveFleetUpdate(ctx context.Context) (*FleetUpdate, error) { fu.StartedAt, _ = time.Parse(time.RFC3339Nano, startedAt) fu.CurrentHostID = current.String fu.HaltedReason = halted.String + fu.PauseAfterFirst = pauseAfterFirst != 0 if completedAt.Valid { t, _ := time.Parse(time.RFC3339Nano, completedAt.String) fu.CompletedAt = &t @@ -97,12 +99,13 @@ func (st *Store) GetFleetUpdate(ctx context.Context, id string) (*FleetUpdate, [ var current sql.NullString var halted sql.NullString var completedAt sql.NullString + var pauseAfterFirst int err := st.db.QueryRowContext(ctx, `SELECT id, started_at, started_by_user_id, target_version, status, - current_host_id, halted_reason, completed_at + current_host_id, halted_reason, completed_at, pause_after_first FROM fleet_updates WHERE id = ?`, id). Scan(&fu.ID, &startedAt, &fu.StartedByUserID, &fu.TargetVersion, &fu.Status, - ¤t, &halted, &completedAt) + ¤t, &halted, &completedAt, &pauseAfterFirst) if errors.Is(err, sql.ErrNoRows) { return nil, nil, ErrNotFound } @@ -112,6 +115,7 @@ func (st *Store) GetFleetUpdate(ctx context.Context, id string) (*FleetUpdate, [ fu.StartedAt, _ = time.Parse(time.RFC3339Nano, startedAt) fu.CurrentHostID = current.String fu.HaltedReason = halted.String + fu.PauseAfterFirst = pauseAfterFirst != 0 if completedAt.Valid { t, _ := time.Parse(time.RFC3339Nano, completedAt.String) fu.CompletedAt = &t diff --git a/internal/store/migrations/0026_fleet_update_canary.sql b/internal/store/migrations/0026_fleet_update_canary.sql new file mode 100644 index 0000000..57d890a --- /dev/null +++ b/internal/store/migrations/0026_fleet_update_canary.sql @@ -0,0 +1 @@ +ALTER TABLE fleet_updates ADD COLUMN pause_after_first INTEGER NOT NULL DEFAULT 0; diff --git a/internal/store/types.go b/internal/store/types.go index a467557..b5f9e11 100644 --- a/internal/store/types.go +++ b/internal/store/types.go @@ -231,6 +231,7 @@ type FleetUpdate struct { CurrentHostID string HaltedReason string CompletedAt *time.Time + PauseAfterFirst bool } // FleetUpdateHost is one host's slot in a fleet update. Position is diff --git a/web/templates/pages/dashboard.html b/web/templates/pages/dashboard.html index 1788904..80d5fec 100644 --- a/web/templates/pages/dashboard.html +++ b/web/templates/pages/dashboard.html @@ -69,7 +69,7 @@ {{/* ---------- Hosts-behind hero tile (P6-18) ---------- */}} {{if gt $page.UpdatesBehind 0}}
- + {{$page.UpdatesBehind}} {{if eq $page.UpdatesBehind 1}}host behind{{else}}hosts behind{{end}} · review → diff --git a/web/templates/pages/fleet_update.html b/web/templates/pages/fleet_update.html index 3373e0d..fa3d815 100644 --- a/web/templates/pages/fleet_update.html +++ b/web/templates/pages/fleet_update.html @@ -8,18 +8,18 @@
Dashboard/ Settings/ - fleet update + agent updates
{{/* page header */}}

- Fleet update + Agent updates target {{$page.TargetVersion}}

- Rolling, sequential agent self-update. One host at a time, halts on first failure, + Rolling, sequential agent self-update. One host at a time, halts on first failure, cancellable mid-roll. Only online hosts whose agent_version differs from the server are eligible.

diff --git a/web/templates/pages/settings.html b/web/templates/pages/settings.html index f7c1ae2..b6eab3a 100644 --- a/web/templates/pages/settings.html +++ b/web/templates/pages/settings.html @@ -40,6 +40,7 @@ {{if not $page.Form}}{{len $page.Channels}}{{end}} Users + Agent updates Authentication
diff --git a/web/templates/partials/fleet_update_inner.html b/web/templates/partials/fleet_update_inner.html index 357881f..464be9b 100644 --- a/web/templates/partials/fleet_update_inner.html +++ b/web/templates/partials/fleet_update_inner.html @@ -32,6 +32,7 @@
target {{$page.Active.TargetVersion}} + {{if $page.Active.PauseAfterFirst}} · canary-first{{end}} · started {{relTime $page.Active.StartedAt}} {{if $page.Active.CurrentHostID}} · waiting on {{index $page.HostNames $page.Active.CurrentHostID}} @@ -68,6 +69,12 @@ {{if $page.Active.HaltedReason}}
{{$page.Active.HaltedReason}}
{{end}} + {{if eq $page.Active.Status "halted"}} +
+ Resume remaining as new rollout + {{range $page.ActiveRows}}{{if eq .Status "failed"}}{{end}}{{end}} +
+ {{end}}
{{template "fleet_update_rows" $page}} @@ -122,11 +129,11 @@ {{define "fleet_update_idle_panel"}} {{$page := .}}
- {{if eq (len $page.OutOfDateHosts) 0}} + {{if eq (len $page.Candidates) 0}}
-
All hosts are up to date.
+
No hosts enrolled.
Every online agent matches server version {{$page.TargetVersion}}.
@@ -137,35 +144,48 @@

{{len $page.OutOfDateHosts}} host{{if ne (len $page.OutOfDateHosts) 1}}s{{end}} out of date

target {{$page.TargetVersion}}
-
    - {{range $page.OutOfDateHosts}} -
  • - - {{.Name}} - {{if .AgentVersion}}{{.AgentVersion}}{{else}}—{{end}} → {{$page.TargetVersion}} -
    • - {{end}} -
- -
-
{{end}}