Add selectable canary-first fleet updates
CI / Test (rest) (pull_request) Successful in 39s
CI / Test (store) (pull_request) Successful in 42s
CI / Lint (pull_request) Successful in 11s
CI / Build (windows/amd64) (pull_request) Successful in 7s
CI / Build (linux/amd64) (pull_request) Successful in 8s
CI / Build (linux/arm64) (pull_request) Successful in 8s
CI / Test (server-http) (pull_request) Successful in 1m45s
e2e / Playwright vs docker-compose (pull_request) Successful in 1m17s

This commit is contained in:
2026-08-22 11:08:53 +01:00
parent b64f029892
commit 383bdb7d36
11 changed files with 301 additions and 53 deletions
+31 -1
View File
@@ -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)
}