From 2751e9796ac57765ade91b7de848fc58ee2e73ae Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Wed, 28 Oct 2026 19:44:25 +0000 Subject: [PATCH] fix(commitrequest): settle every save before the controller exits A clean shutdown lost the saves it held, in two ways. The branch workers ran on a context derived from the manager's, so the manager's cancel also canceled the shutdown's own last push: over a real transport it was refused before it reached the remote, and the save attached to the open window failed with "worker stopped before the commit reached the remote". A worker now stops on one context and works on another, which ends 10s after the stop or when its loop exits, so the last window is committed, pushed and its saves settled. The CommitRequest controllers stop at the same moment as the workers (controller-runtime stops every leader-election runnable together), so nothing read the outcomes the workers decided while they drained, and the next process attached the request again as new: NoWindow, or its message on a later window. The core now remembers every request it attaches, with its path, and once every worker has stopped the worker manager runs CommitRequestsInFlight.Settle on a fresh 5s context: one step per request that reads the stopped worker's outcome (LookupCommitRequestOutcome, which sends nothing) and writes it with the core's own decide, conditional on resourceVersion, on the control cluster and on source clusters alike. A request no worker decided keeps its status for the next process, as before. Nothing is persisted: a crash, or a stop past its grace period, behaves as before. docs/TODO.md folds the withheld-field item into one "Restart recovery: persist current state" entry. Co-Authored-By: Claude Opus 5.5 --- .coverage-baseline | 2 +- cmd/main.go | 7 + docs/TODO.md | 17 +- docs/api-first-publication.md | 6 +- docs/configuration.md | 7 +- docs/design/branch-worker-event-model.md | 13 +- docs/spec/commitrequest-design.md | 15 +- docs/spec/e2e-serial-registry.md | 3 +- .../controller/commitrequest_controller.go | 9 +- .../commitrequest_controller_unit_test.go | 11 + internal/controller/commitrequest_core.go | 9 + internal/controller/commitrequest_decide.go | 1 + internal/controller/commitrequest_server.go | 7 +- internal/controller/commitrequest_settle.go | 149 +++++++++++ .../controller/commitrequest_settle_test.go | 235 ++++++++++++++++++ .../controller/commitrequest_wakes_test.go | 11 + internal/git/branch_worker.go | 33 ++- internal/git/shutdown_drain_test.go | 137 ++++++++++ internal/git/worker_manager.go | 37 ++- internal/git/worker_repo_identity_test.go | 4 +- internal/watch/event_router.go | 15 ++ internal/watch/event_router_outcome_test.go | 78 ++++++ test/e2e/commit_request_restart_e2e_test.go | 125 ++++++++++ 23 files changed, 905 insertions(+), 26 deletions(-) create mode 100644 internal/controller/commitrequest_settle.go create mode 100644 internal/controller/commitrequest_settle_test.go create mode 100644 internal/git/shutdown_drain_test.go create mode 100644 internal/watch/event_router_outcome_test.go create mode 100644 test/e2e/commit_request_restart_e2e_test.go diff --git a/.coverage-baseline b/.coverage-baseline index a03b4942..8d0afb62 100644 --- a/.coverage-baseline +++ b/.coverage-baseline @@ -1 +1 @@ -82.5 +82.6 diff --git a/cmd/main.go b/cmd/main.go index 66457091..c1e01129 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -331,6 +331,11 @@ func main() { // A worker's change to a CommitRequest takes the request's next step at once, on either path. commitRequestWakes := controller.NewCommitRequestWakes(workerManager.CommitRequestChanges()) fatalIfErr(mgr.Add(commitRequestWakes), "unable to add the CommitRequest wakes") + // At shutdown the workers settle every save they hold as they stop, and the CommitRequest + // controllers stop with them: the requests either path attached have their outcomes written once + // the workers have stopped, before the process exits. + commitRequestsInFlight := &controller.CommitRequestsInFlight{Log: ctrl.Log.WithName("commitrequests")} + workerManager.SetAfterWorkersStopped(commitRequestsInFlight.Settle) commitRequestServer := &controller.CommitRequestServer{ Client: mgr.GetClient(), APIReader: mgr.GetAPIReader(), @@ -340,6 +345,7 @@ func main() { FactStreams: watchMgr.FactStreams, TTL: cfg.commitRequestTTL, Wakes: commitRequestWakes, + InFlight: commitRequestsInFlight, Log: ctrl.Log.WithName("commitrequests"), } fatalIfErr(mgr.Add(commitRequestServer), "unable to add the CommitRequest server") @@ -429,6 +435,7 @@ func main() { FactStreams: watchMgr.FactStreams, TTL: cfg.commitRequestTTL, Wakes: commitRequestWakes, + InFlight: commitRequestsInFlight, }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "unable to create controller", "controller", "CommitRequest") os.Exit(1) diff --git a/docs/TODO.md b/docs/TODO.md index 893a395b..b79eaecd 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -11,9 +11,20 @@ Nothing open. ## Applier claims and bi-directional editing -- [ ] **Persist the withheld-field record.** The fields a withheld apply reverted are kept in memory - (`internal/git/withheld_fields.go`), so after a restart the next edit of such an object carries the - apply's values into Git again, until the `Repair` lands. +- [ ] **Restart recovery: persist current state.** A graceful stop pushes its windows and writes the + outcomes of the saves they carried (`CommitRequestsInFlight.Settle`); everything else the branch + worker keeps in memory is lost on any restart, and a crash, or a stop past its grace period, loses + all of it. Nothing is persisted. What must survive: the + open windows with absolute deadlines, author, attribution, message and membership; each attached + save's `attachDeadline`, `seq` and window; save outcomes until their TTL; the pending writes; the + withheld-field records (`internal/git/withheld_fields.go`: after a restart the next edit of such an + object carries the apply's values into Git again, until the `Repair` lands). "Git took the push, + then the process died" is made safe by writing a push intent {old, new, request→SHA} before every + `PushAtomic` and comparing it with the remote tip on restart: commits carry no trailers, and SHAs + are not reproducible. Two restore traps: the startup cold replay (no Redis) closes any restored + window (`resync_flush.go`), and `finalizeWindow` drops a window while render fidelity is Unknown, + which it is after a restart. The worker state root `/tmp` is an `emptyDir` a new pod loses. + [branch-worker-event-model.md](design/branch-worker-event-model.md#saves-lose-execution-identity-across-a-restart). - [ ] **Reopen the render-fidelity gate after a Git repair, automatically.** Today a target on `RenderMatchesLive=False` clears only on an explicit recheck (the `reconcile.configbutler.ai/requestedAt` annotation, which restarts every scope; the render diff --git a/docs/api-first-publication.md b/docs/api-first-publication.md index d7926290..ed315df0 100644 --- a/docs/api-first-publication.md +++ b/docs/api-first-publication.md @@ -310,8 +310,10 @@ sequenceDiagram A local no-diff result still consults the remote before reporting `AlreadyPresent`. A competing Git edit can make that same captured API object require a real commit after replay. -A worker that stops while still holding the write fails the request with an error rather than -leaving it to time out, so a shutdown mid-publication is reported as a failure and not as silence. +A worker that stops while still holding the write pushes it first. If that last push fails, it fails +the request with an error rather than leaving it to time out, so a shutdown mid-publication is +reported as a failure and not as silence. Either outcome is written to the request's status before +the process exits. Use `Pushed=True` plus `status.commit` when a particular commit must exist in Git. `Ready=True` also covers successful outcomes with no commit. Status visibility follows the controller's polling diff --git a/docs/configuration.md b/docs/configuration.md index fd44a76d..f6b6e5a3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -3326,7 +3326,12 @@ request produced a pushed commit: does not mean the commit was applied. A request is served again after GitOps Reverser restarts: the path that owns it finds its own -claim, and resumes with the author it recorded. +claim, and resumes with the author it recorded. A graceful stop, such as an upgrade or a rollout, +first finishes the saves it holds: a request collecting its window is committed with its message, +pushed, and its outcome written before the pod exits. A request still waiting for a window is +picked up again by the next pod. A crash, or a stop that runs out of its grace period, loses the +open window: the next pod's startup replay writes its edits under a generated message, and the +request attaches again as if new, so it can end `NoWindow`. ### How long a request lives diff --git a/docs/design/branch-worker-event-model.md b/docs/design/branch-worker-event-model.md index d44f4087..b3855a11 100644 --- a/docs/design/branch-worker-event-model.md +++ b/docs/design/branch-worker-event-model.md @@ -24,8 +24,9 @@ The write-path refactor in [`gittarget-branch-worker-pending-writes.md`](gittarg what this section asked for, in the order it set: ordered pending writes, one materializer for the checkout, and one retry deadline; then the watch's admission boundary (5a), Git operation deadlines (6), intake pause and resume (5b), and the publication status (5c). None of it depends on a storage -backend or on HA. A saturated branch pauses intake while its recovery loop stays alive; a process -restart still loses the pending writes and unanswered saves held in memory. +backend or on HA. A saturated branch pauses intake while its recovery loop stays alive. A graceful +stop pushes the pending writes and writes the saves' outcomes before it exits; a crash, or a stop +that runs out of time, still loses the pending writes and unanswered saves held in memory. What comes next is the [transition boundary](#prepare-the-transition-boundary-for-later-durability): recording timers, attaches, withdrawals, and refreshes as transitions, then the HA plan's journal. @@ -554,9 +555,11 @@ The live producer is attribution and never independently supplies mirrored object state. The [event router](../../internal/watch/event_router.go) binds an already-held request to its -original worker during retirement. Today shutdown is best effort: it tries to finalize and push, -then fails unresolved requests and discards unhandled work. Durable worker retirement needs an -explicit transfer or terminal decision before that responsibility can be released. +original worker during retirement. Today shutdown is best effort, within a 10s budget: it finalizes +and pushes, fails the requests riding a write it could not push, and discards unhandled work; the +controller then writes every outcome the stopped workers decided (`CommitRequestsInFlight.Settle`). +Durable worker retirement needs an explicit transfer or terminal decision before that responsibility +can be released. Snapshots establish render fidelity even while ordinary live writes are refused. Suspension suppresses writes while allowing observation. A single dispatcher-level write gate would block diff --git a/docs/spec/commitrequest-design.md b/docs/spec/commitrequest-design.md index b4ce2fc5..cacce0e0 100644 --- a/docs/spec/commitrequest-design.md +++ b/docs/spec/commitrequest-design.md @@ -320,10 +320,17 @@ created successfully nor a later source snapshot proves that the worker accepted Resource watches and save commands arrive independently: a save is not a barrier covering every Kubernetes mutation before its creation. -Pausing the live worker preserves accepted work in memory. Restarting the process does not preserve -its attachments or outcome receipts. Startup replays current watched state, subject to pruning, -but cannot recreate the same save history. A push accepted immediately before a crash can therefore be -followed by another empty commit when the request is resent. A lost push response also needs remote +Pausing the live worker preserves accepted work in memory. A graceful stop settles what it holds +before the process exits: every worker closes its windows, pushes, and settles the saves it carries, +and once every worker has stopped the controller writes each settled outcome to its request's status +through the path that attached it, as one more conditional step (`CommitRequestsInFlight.Settle`). +Both are bounded (10s for the workers, 5s for the statuses) to fit the chart's 20s grace period; what +runs over is left as a crash leaves it. A request still waiting for a window, or whose attach was +still queued, is not committed; the next process attaches it again from its status. Restarting the +process does not otherwise preserve its attachments or outcome receipts: nothing is persisted. +Startup replays current watched state, subject to pruning, but cannot recreate the same save history. +A push accepted immediately before a crash can therefore be followed by another empty commit when +the request is resent. A lost push response also needs remote evidence; equal tree content alone cannot prove that an empty save's message was published. Durable save recovery belongs to the [future HA plan](../future/ha-gittarget-distribution-plan.md). diff --git a/docs/spec/e2e-serial-registry.md b/docs/spec/e2e-serial-registry.md index c10f5bf6..c3c9cd78 100644 --- a/docs/spec/e2e-serial-registry.md +++ b/docs/spec/e2e-serial-registry.md @@ -11,7 +11,7 @@ under bounded Ginkgo parallelism (Phase 2.5 of > The two audit-consumer specs, `bi_directional`, and `aggregated_apiserver` have > all been de-serialized (see the "De-serialized" section). `crd_lifecycle` was > de-serialized and then **re-serialized** (commit `3d249e3`) when a residual -> cluster-wide CRD-discovery churn re-flaked it. The current Serial set is the four +> cluster-wide CRD-discovery churn re-flaked it. The current Serial set is the five > singleton-controller specs, `crd_lifecycle`, and `playground`. ## Rule @@ -39,6 +39,7 @@ the `full-core` leg beside the restart specs instead of at the end of the longes | [test/e2e/restart_reconcile_e2e_test.go](../../test/e2e/restart_reconcile_e2e_test.go) | `Restart Reconcile Safety` | Rollout-restarts the controller deployment. | The controller is a singleton; restarting it disrupts in-flight reconciles/commits for every other spec. | | [test/e2e/flux_restart_recovery_e2e_test.go](../../test/e2e/flux_restart_recovery_e2e_test.go) | `Flux Restart Recovery` | Rollout-restarts the controller deployment while a Flux apply runs, and labels a node so that apply's Deployment can schedule. | The controller is a singleton, as for `Restart Reconcile Safety`; a node label is cluster-wide. | | [test/e2e/bi_directional_restart_e2e_test.go](../../test/e2e/bi_directional_restart_e2e_test.go) | `Bi Directional restart with work made while down (Flux)`, `(Argo CD)` | Scales the controller deployment to zero and back, to make edits and a Git-side push while the operator is down. | The controller is a singleton, as for `Restart Reconcile Safety`. | +| [test/e2e/commit_request_restart_e2e_test.go](../../test/e2e/commit_request_restart_e2e_test.go) | `Save Request across a controller restart` | Scales the controller deployment to zero and back, to stop it while a save collects its window. | The controller is a singleton, as for `Restart Reconcile Safety`. | | [test/e2e/image_refresh_test.go](../../test/e2e/image_refresh_test.go) | `image refresh dependency chain` | Changes the controller image / redeploys the controller. | Same singleton controller; an image swap perturbs all concurrent specs. | | [test/e2e/crd_lifecycle_e2e_test.go](../../test/e2e/crd_lifecycle_e2e_test.go) | `CRD Lifecycle` | Installs/deletes a cluster-scoped CRD and watches all `customresourcedefinitions` cluster-wide, asserting exact Git file presence/absence. | A concurrent CRD install/delete bumps the global discovery-catalog generation and re-resolves every GitTarget's watched-type tables, delaying this spec's reconcile (CRD file appears late / post-delete sweep lags — the 649↔673 flake). A cluster-wide catalog bump cannot be scoped by name. | | [test/e2e/tilt_playground_e2e_test.go](../../test/e2e/tilt_playground_e2e_test.go) | `playground` | Reusable manual-playground fixture: fixed (non-randomized) `tilt-playground` namespace and `playground` repo/provider names, preserved across runs. | A singleton by construction — its resource names are intentionally stable so `task playground-*` can re-attach, so per-run name isolation does not apply. | diff --git a/internal/controller/commitrequest_controller.go b/internal/controller/commitrequest_controller.go index 00f2fee7..19a7211e 100644 --- a/internal/controller/commitrequest_controller.go +++ b/internal/controller/commitrequest_controller.go @@ -38,6 +38,9 @@ type CommitRequestFinalizer interface { // answer: resolved with git.ErrCommitRequestWithdrawn when it is cancelled, the real outcome // when the worker got there first, or resolved=false with the worker's phase while it decides. WithdrawCommitRequest(ctx context.Context, attach git.AttachCommitRequest) (git.FinalizeResult, bool, error) + // LookupCommitRequestOutcome reports the outcome the worker that holds the request decided, and + // false while it decided none or no worker holds it. It sends nothing to the worker. + LookupCommitRequestOutcome(ctx context.Context, attach git.AttachCommitRequest) (git.FinalizeResult, bool, error) // LookupCommitRequestApply reports where the apply a request with catchUp: // AfterPush asked for stands, and false while the worker knows nothing of it. LookupCommitRequestApply(ctx context.Context, attach git.AttachCommitRequest) (git.SaveApply, bool, error) @@ -169,6 +172,10 @@ type CommitRequestReconciler struct { // request is polled only as the fallback. Nil polls every request. Wakes *CommitRequestWakes + // InFlight remembers the requests this reconciler attached, so a shutdown writes the outcomes + // their workers decide while they drain. Nil remembers nothing. + InFlight *CommitRequestsInFlight + streamsOnce sync.Once streams *commitRequestStreams } @@ -193,7 +200,7 @@ func (r *CommitRequestReconciler) core() *commitRequestCore { r.streamsOnce.Do(func() { r.streams = &commitRequestStreams{set: r.FactStreams} }) return &commitRequestCore{ reader: r.Client, finalizer: r.Finalizer, authors: r.Authors, streams: r.streams, ttl: r.TTL, - wakeFallback: r.Wakes.fallback(), + wakeFallback: r.Wakes.fallback(), inFlight: r.InFlight, } } diff --git a/internal/controller/commitrequest_controller_unit_test.go b/internal/controller/commitrequest_controller_unit_test.go index 5129b8cb..78d41e8d 100644 --- a/internal/controller/commitrequest_controller_unit_test.go +++ b/internal/controller/commitrequest_controller_unit_test.go @@ -36,6 +36,8 @@ type fakeFinalizer struct { err error calls []git.AttachCommitRequest + // outcomeLookups are the lookups of a decided outcome, which send nothing. + outcomeLookups []git.AttachCommitRequest // The withdraw reply. Left unset, the withdraw answers like the router does before it cancels // anything: with the worker's outcome, or its phase while unresolved. withdrawPending makes an @@ -53,6 +55,15 @@ type fakeFinalizer struct { applyLookups int } +// LookupCommitRequestOutcome answers with the canned outcome, as a worker that decided it does, and +// records the lookup: nothing is sent to the worker. +func (f *fakeFinalizer) LookupCommitRequestOutcome( + _ context.Context, attach git.AttachCommitRequest, +) (git.FinalizeResult, bool, error) { + f.outcomeLookups = append(f.outcomeLookups, attach) + return f.result, f.resolved, f.err +} + func (f *fakeFinalizer) LookupCommitRequestApply( _ context.Context, _ git.AttachCommitRequest, ) (git.SaveApply, bool, error) { diff --git a/internal/controller/commitrequest_core.go b/internal/controller/commitrequest_core.go index 76841ed5..6b245278 100644 --- a/internal/controller/commitrequest_core.go +++ b/internal/controller/commitrequest_core.go @@ -110,6 +110,8 @@ type commitRequestCore struct { streams *commitRequestStreams // ttl is how long a finished request is kept, written as its delete-after time; zero writes none. ttl time.Duration + // inFlight remembers the requests this core attached, for the shutdown to settle; nil remembers none. + inFlight *CommitRequestsInFlight } // serve takes one step for the request key on path p, and says how soon to take the next. Zero means @@ -119,12 +121,17 @@ func (c *commitRequestCore) serve( ) (time.Duration, error) { cr, err := p.store.get(ctx, key) if err != nil { + if apierrors.IsNotFound(err) { + c.inFlight.forget(p, key) + } return 0, client.IgnoreNotFound(err) } if !cr.DeletionTimestamp.IsZero() { + c.inFlight.forget(p, key) return 0, nil } if commitRequestIsTerminal(cr) { + c.inFlight.forget(p, key) return c.serveFinished(ctx, p, cr) } if c.finalizer == nil { @@ -483,6 +490,8 @@ func (c *commitRequestCore) drive( ctx context.Context, p commitRequestPath, cr *configbutleraiv1alpha3.CommitRequest, target *configbutleraiv1alpha3.GitTarget, provider *configbutleraiv1alpha3.KubeProvider, ) (time.Duration, error) { + // Remembered before anything reaches a worker, so whatever it decides is written at shutdown. + c.inFlight.track(c, p, client.ObjectKeyFromObject(cr)) attach := c.attachFor(cr) if target != nil && string(target.UID) != cr.Status.GitTargetUID { admitted, refusal, err := c.admit(ctx, p, cr, target, provider) diff --git a/internal/controller/commitrequest_decide.go b/internal/controller/commitrequest_decide.go index 2600a4cf..95e31126 100644 --- a/internal/controller/commitrequest_decide.go +++ b/internal/controller/commitrequest_decide.go @@ -93,6 +93,7 @@ func (c *commitRequestCore) decide( if err != nil || !written { return commitRequestRequeueNow, err } + c.inFlight.forget(p, client.ObjectKeyFromObject(current)) if !afterClaim { recordCommitRequestOutcome(ctx, o.metricOutcome(), target) } diff --git a/internal/controller/commitrequest_server.go b/internal/controller/commitrequest_server.go index 4d13e2da..885da10e 100644 --- a/internal/controller/commitrequest_server.go +++ b/internal/controller/commitrequest_server.go @@ -108,7 +108,10 @@ type CommitRequestServer struct { // Wakes wakes a request's steps as soon as its branch worker reports a change, so a registered // request is polled only as the fallback. Nil polls every request. Wakes *CommitRequestWakes - Log logr.Logger + // InFlight remembers the requests the server attached, so a shutdown writes the outcomes their + // workers decide while they drain, through the identity that serves them. Nil remembers nothing. + InFlight *CommitRequestsInFlight + Log logr.Logger mu sync.Mutex watchers map[k8stypes.NamespacedName]*commitRequestWatcher @@ -600,7 +603,7 @@ func (s *CommitRequestServer) core() *commitRequestCore { s.streamsOnce.Do(func() { s.streams = &commitRequestStreams{set: s.FactStreams} }) return &commitRequestCore{ reader: s.Client, controlRequests: s.APIReader, finalizer: s.Finalizer, authors: s.Authors, - streams: s.streams, ttl: s.TTL, wakeFallback: s.Wakes.fallback(), + streams: s.streams, ttl: s.TTL, wakeFallback: s.Wakes.fallback(), inFlight: s.InFlight, } } diff --git a/internal/controller/commitrequest_settle.go b/internal/controller/commitrequest_settle.go new file mode 100644 index 00000000..c0b04575 --- /dev/null +++ b/internal/controller/commitrequest_settle.go @@ -0,0 +1,149 @@ +// SPDX-License-Identifier: Apache-2.0 + +package controller + +import ( + "context" + "sync" + + "github.com/go-logr/logr" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// A save's outcome is decided by its branch worker and lives in the worker's memory until a step of +// the request reads it and writes it to the request's status. A clean shutdown is where those two +// part: the workers close their windows, push and settle every save they hold as they stop, but the +// CommitRequest controllers stop at the same moment, so nothing reads those outcomes, and the process +// exits with them. The next process sends the request's attach again, and its new worker, which never +// saw it, ends the save NoWindow, or attaches its message to a later window. +// +// So the shutdown writes them itself. Every request a path attached is remembered here with that +// path, and once every worker has stopped (git.WorkerManager.SetAfterWorkersStopped), Settle writes +// the outcome each worker decided through the request's own path: the same decide a step makes, +// conditional on the resourceVersion, re-applying the outcome on a conflict. A request no worker +// decided anything for is left as it is: the next process attaches it again, from its status. +// +// Nothing is persisted. A crash, or a shutdown that runs out of time, behaves as before. + +// CommitRequestsInFlight remembers every CommitRequest a path of this process attached to a branch +// worker and has not finished, with that path, for Settle. Nil remembers nothing. +type CommitRequestsInFlight struct { + mu sync.Mutex + requests map[inFlightKey]inFlightRequest + // Log reports what Settle wrote and what it left. + Log logr.Logger +} + +// inFlightKey is one request on one path: the KubeProvider whose watch serves it, empty on the control +// cluster, and its name there. +type inFlightKey struct { + provider types.NamespacedName + request types.NamespacedName +} + +// inFlightRequest is how to take a step of a remembered request. +type inFlightRequest struct { + core *commitRequestCore + path commitRequestPath +} + +// track remembers a request p attaches. +func (f *CommitRequestsInFlight) track(c *commitRequestCore, p commitRequestPath, key types.NamespacedName) { + if f == nil { + return + } + f.mu.Lock() + defer f.mu.Unlock() + if f.requests == nil { + f.requests = map[inFlightKey]inFlightRequest{} + } + f.requests[inFlightKey{provider: p.provider, request: key}] = inFlightRequest{core: c, path: p} +} + +// forget drops a request that finished, or is gone. +func (f *CommitRequestsInFlight) forget(p commitRequestPath, key types.NamespacedName) { + if f == nil { + return + } + f.mu.Lock() + defer f.mu.Unlock() + delete(f.requests, inFlightKey{provider: p.provider, request: key}) +} + +// remembered copies the requests out, for Settle to step through without the lock. +func (f *CommitRequestsInFlight) remembered() map[inFlightKey]inFlightRequest { + f.mu.Lock() + defer f.mu.Unlock() + out := make(map[inFlightKey]inFlightRequest, len(f.requests)) + for key, request := range f.requests { + out[key] = request + } + return out +} + +// Settle writes, on ctx, the outcome each remembered request's worker decided, commitRequestConcurrency +// at a time. It runs at shutdown once every worker has stopped, so no worker decides anything more. +// What ctx leaves no time for is left for the next process, as a request with no outcome is. +func (f *CommitRequestsInFlight) Settle(ctx context.Context) { + if f == nil { + return + } + requests := f.remembered() + if len(requests) == 0 { + return + } + f.Log.Info("Writing the outcomes of the CommitRequests in flight at shutdown", "requests", len(requests)) + slots := make(chan struct{}, commitRequestConcurrency) + var wg sync.WaitGroup + for key, request := range requests { + if ctx.Err() != nil { + break + } + select { + case <-ctx.Done(): + case slots <- struct{}{}: + wg.Go(func() { + defer func() { <-slots }() + if err := request.core.settle(ctx, request.path, key.request); err != nil { + request.path.log.Error(err, "Could not write a CommitRequest's outcome at shutdown; "+ + "the next process attaches it again", "name", key.request) + } + }) + } + } + wg.Wait() + if left := f.remembered(); len(left) > 0 { + names := make([]string, 0, len(left)) + for key := range left { + names = append(names, key.request.String()) + } + f.Log.Info("CommitRequests left without an outcome at shutdown; the next process attaches them again", + "requests", names) + } +} + +// settle is a request's last step at shutdown: it writes the outcome the worker that held it decided, +// as drive would, and nothing else. It sends no attach and no withdraw, and records no phase: no worker +// runs any more, and a request no worker decided keeps its status for the next process to resume. +func (c *commitRequestCore) settle(ctx context.Context, p commitRequestPath, key types.NamespacedName) error { + cr, err := p.store.get(ctx, key) + if err != nil { + if client.IgnoreNotFound(err) == nil { + c.inFlight.forget(p, key) + } + return client.IgnoreNotFound(err) + } + if !cr.DeletionTimestamp.IsZero() || !commitRequestClaimed(cr) { + c.inFlight.forget(p, key) + return nil + } + result, resolved, err := c.finalizer.LookupCommitRequestOutcome(ctx, c.attachFor(cr)) + if err != nil || !resolved { + return err + } + p.log.Info("CommitRequest settled at shutdown", "name", key, "outcome", string(result.Outcome), + "commit", result.Commit, "err", result.Err) + _, err = c.decide(ctx, p, cr, commitRequestOutcome{result: result, err: result.Err}, true) + return err +} diff --git a/internal/controller/commitrequest_settle_test.go b/internal/controller/commitrequest_settle_test.go new file mode 100644 index 00000000..77c7310b --- /dev/null +++ b/internal/controller/commitrequest_settle_test.go @@ -0,0 +1,235 @@ +// SPDX-License-Identifier: Apache-2.0 + +package controller + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + dynamicfake "k8s.io/client-go/dynamic/fake" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + "github.com/ConfigButler/gitops-reverser/internal/git" +) + +// drainingWorker is a branch worker that holds a request, collecting its window, until the shutdown +// settles it: then it has decided the request's outcome and stopped, and only a lookup reads it. +type drainingWorker struct { + mu sync.Mutex + outcome git.FinalizeResult + stopped bool + attaches int + lookups int +} + +func (w *drainingWorker) stop(outcome git.FinalizeResult) { + w.mu.Lock() + defer w.mu.Unlock() + w.outcome, w.stopped = outcome, true +} + +func (w *drainingWorker) ServiceCommitRequest( + context.Context, git.AttachCommitRequest, +) (git.FinalizeResult, bool, error) { + w.mu.Lock() + defer w.mu.Unlock() + w.attaches++ + if w.stopped { + return w.outcome, true, nil + } + return git.FinalizeResult{Phase: git.PhaseCollectingWindow}, false, nil +} + +func (w *drainingWorker) WithdrawCommitRequest( + ctx context.Context, attach git.AttachCommitRequest, +) (git.FinalizeResult, bool, error) { + return w.ServiceCommitRequest(ctx, attach) +} + +func (w *drainingWorker) LookupCommitRequestOutcome( + context.Context, git.AttachCommitRequest, +) (git.FinalizeResult, bool, error) { + w.mu.Lock() + defer w.mu.Unlock() + w.lookups++ + return w.outcome, w.stopped, nil +} + +func (w *drainingWorker) LookupCommitRequestApply( + context.Context, git.AttachCommitRequest, +) (git.SaveApply, bool, error) { + return git.SaveApply{}, false, nil +} + +// counts are the attaches and the lookups the worker was asked for. +func (w *drainingWorker) counts() (int, int) { + w.mu.Lock() + defer w.mu.Unlock() + return w.attaches, w.lookups +} + +var shutdownCommit = git.FinalizeResult{Outcome: git.FinalizeCommitted, Commit: "c0ffee", Branch: "main"} + +// The race a clean shutdown had: the reconciler attached the save, the worker settled it while it +// stopped, and no step of the reconciler ran after that to read the outcome. Settle writes it, from +// the worker's lookup alone, without attaching the request again. +func TestCommitRequestsInFlight_TheShutdownWritesWhatTheWorkerSettled(t *testing.T) { + c := newCommitRequestClient(t, nil, newCommitRequest("save")) + worker := &drainingWorker{} + inFlight := &CommitRequestsInFlight{Log: logr.Discard()} + r := &CommitRequestReconciler{Client: c, Finalizer: worker, InFlight: inFlight} + key := types.NamespacedName{Namespace: "default", Name: "save"} + _, err := r.Reconcile(context.Background(), ctrl.Request{NamespacedName: key}) + require.NoError(t, err) + require.Equal(t, string(git.PhaseCollectingWindow), commitRequestReady(t, c, "save").Reason) + attachedBefore, _ := worker.counts() + + worker.stop(shutdownCommit) + inFlight.Settle(context.Background()) + + stored := fetchCommitRequest(t, c, "save") + assert.Equal(t, crReasonCommitted, commitRequestReady(t, c, "save").Reason) + assert.Equal(t, "c0ffee", stored.Status.Commit) + attaches, lookups := worker.counts() + assert.Equal(t, attachedBefore, attaches, "the shutdown attaches nothing") + assert.Equal(t, 1, lookups) + assert.Empty(t, inFlight.remembered(), "a written outcome is forgotten") +} + +// A failure the worker decided at shutdown, its last push refused, is written too: the request +// reports it instead of being sent to a new worker that never saw it. +func TestCommitRequestsInFlight_AFailureAtShutdownIsWritten(t *testing.T) { + c := newCommitRequestClient(t, nil, withInProgress(newCommitRequest("save"))) + worker := &drainingWorker{} + inFlight := &CommitRequestsInFlight{Log: logr.Discard()} + core := &commitRequestCore{reader: c, finalizer: worker, inFlight: inFlight} + serveStep(t, core, controlPath(c), "save") + + worker.stop(git.FinalizeResult{Branch: "main", Err: git.ErrBranchWorkerStopped}) + inFlight.Settle(context.Background()) + + stalled := conditionIsTrue(fetchCommitRequest(t, c, "save").Status.Conditions, ConditionTypeStalled) + assert.True(t, stalled, "the request fails with the worker's answer") +} + +// A request no worker decided anything for, waiting for a window or still on its way to one, keeps its +// status: the next process attaches it again from there. Settle writes nothing for it. +func TestCommitRequestsInFlight_ARequestNoWorkerDecidedIsLeftForTheNextProcess(t *testing.T) { + c := newCommitRequestClient(t, nil, withInProgress(newCommitRequest("save"))) + worker := &drainingWorker{} + inFlight := &CommitRequestsInFlight{Log: logr.Discard()} + core := &commitRequestCore{reader: c, finalizer: worker, inFlight: inFlight} + serveStep(t, core, controlPath(c), "save") + before := fetchCommitRequest(t, c, "save") + + inFlight.Settle(context.Background()) + + after := fetchCommitRequest(t, c, "save") + assert.Equal(t, before.ResourceVersion, after.ResourceVersion, "nothing is written") + assert.Len(t, inFlight.remembered(), 1, "it stays in flight") +} + +// Settle's write is the owner's write: a conflict reads the request again and applies the worker's +// outcome onto it, and an outcome already written is left alone. +func TestCommitRequestsInFlight_AConflictReappliesTheOutcome(t *testing.T) { + conflicts := 1 + fns := interceptor.Funcs{ + SubResourceUpdate: func(ctx context.Context, c client.Client, sub string, obj client.Object, + opts ...client.SubResourceUpdateOption) error { + cr, ok := obj.(*configv1alpha3.CommitRequest) + if ok && conflicts > 0 && commitRequestIsTerminal(cr) { + conflicts-- + return apierrors.NewConflict(commitRequestGVR().GroupResource(), cr.Name, nil) + } + return c.SubResource(sub).Update(ctx, obj, opts...) + }, + } + c := newCommitRequestClient(t, &fns, withInProgress(newCommitRequest("save"))) + worker := &drainingWorker{} + inFlight := &CommitRequestsInFlight{Log: logr.Discard()} + serveStep(t, &commitRequestCore{reader: c, finalizer: worker, inFlight: inFlight}, controlPath(c), "save") + + worker.stop(shutdownCommit) + inFlight.Settle(context.Background()) + + assert.Zero(t, conflicts) + assert.Equal(t, crReasonCommitted, commitRequestReady(t, c, "save").Reason) + assert.Equal(t, "c0ffee", fetchCommitRequest(t, c, "save").Status.Commit) +} + +// A request on a source cluster is settled through the identity that serves it, after the server's +// watch has stopped with the rest of the controllers. +func TestCommitRequestsInFlight_ASourceClusterRequestIsSettledOnItsCluster(t *testing.T) { + provider := kubeconfigProvider("edge", true) + c := newCommitRequestClient(t, nil, mirroringTarget("edge"), provider) + gvr := commitRequestGVR() + source := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), + map[schema.GroupVersionResource]string{gvr: "CommitRequestList"}, sourceCommitRequest("default", "save")) + worker := &drainingWorker{} + inFlight := &CommitRequestsInFlight{Log: logr.Discard()} + s := &CommitRequestServer{ + Client: c, Finalizer: worker, InFlight: inFlight, Log: logr.Discard(), + Clusters: fakeClusters{dynamic: source, discovery: fakeDiscovery(nil, + commitRequestResources("configbutler.ai/v1alpha3", "commitrequests"))}, + } + ctx, cancel := context.WithCancel(context.Background()) + s.syncProviders(ctx) + readReady := func() string { + u, err := source.Resource(gvr).Namespace("default").Get(context.Background(), "save", metav1.GetOptions{}) + require.NoError(t, err) + cr, err := commitRequestFromUnstructured(u) + require.NoError(t, err) + if ready := findCondition(cr.Status.Conditions, ConditionTypeReady); ready != nil { + return ready.Reason + } + return "" + } + require.Eventually(t, func() bool { return readReady() == string(git.PhaseCollectingWindow) }, + 10*time.Second, 20*time.Millisecond, "the request is attached through the provider's watch") + cancel() + s.stopAll() + + worker.stop(shutdownCommit) + inFlight.Settle(context.Background()) + + assert.Equal(t, crReasonCommitted, readReady(), "written on the source cluster") +} + +// A nil set remembers nothing, and settles nothing. +func TestCommitRequestsInFlight_Nil(t *testing.T) { + var inFlight *CommitRequestsInFlight + assert.NotPanics(t, func() { + inFlight.track(&commitRequestCore{}, commitRequestPath{}, types.NamespacedName{Name: "save"}) + inFlight.forget(commitRequestPath{}, types.NamespacedName{Name: "save"}) + inFlight.Settle(context.Background()) + }) +} + +// A shutdown out of time writes nothing more: what is left is attached again by the next process. +func TestCommitRequestsInFlight_OutOfTimeLeavesTheRest(t *testing.T) { + c := newCommitRequestClient(t, nil, withInProgress(newCommitRequest("save"))) + worker := &drainingWorker{} + inFlight := &CommitRequestsInFlight{Log: logr.Discard()} + serveStep(t, &commitRequestCore{reader: c, finalizer: worker, inFlight: inFlight}, controlPath(c), "save") + worker.stop(shutdownCommit) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + inFlight.Settle(ctx) + + assert.Equal(t, string(git.PhaseCollectingWindow), commitRequestReady(t, c, "save").Reason) + assert.Len(t, inFlight.remembered(), 1) +} diff --git a/internal/controller/commitrequest_wakes_test.go b/internal/controller/commitrequest_wakes_test.go index 10364841..476c81b8 100644 --- a/internal/controller/commitrequest_wakes_test.go +++ b/internal/controller/commitrequest_wakes_test.go @@ -67,6 +67,17 @@ func (w *simulatedWorker) WithdrawCommitRequest( return w.ServiceCommitRequest(ctx, attach) } +func (w *simulatedWorker) LookupCommitRequestOutcome( + context.Context, git.AttachCommitRequest, +) (git.FinalizeResult, bool, error) { + w.mu.Lock() + defer w.mu.Unlock() + if w.resolved { + return git.FinalizeResult{Outcome: git.FinalizeCommitted, Commit: "c0ffee", Branch: "main"}, true, nil + } + return git.FinalizeResult{}, false, nil +} + func (w *simulatedWorker) LookupCommitRequestApply( context.Context, git.AttachCommitRequest, ) (git.SaveApply, bool, error) { diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index 39b0f1a3..12e48185 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -186,9 +186,12 @@ type BranchWorker struct { // worker's, and no applier is asked for anything. Nil in the CLI and in tests. replacementPending func() bool - // Event processing + // Event processing. The loop stops on stop and works on ctx, which outlives the stop by + // workerShutdownBudget, so a stopping loop can still close its windows, push and settle its saves. + // A worker built without Start has no stop, and stops with ctx. eventQueue chan WorkItem ctx context.Context + stop context.Context cancelFunc context.CancelFunc wg sync.WaitGroup started bool @@ -557,7 +560,13 @@ func (w *BranchWorker) Start(parentCtx context.Context) error { w.mu.Unlock() return errors.New("worker already started") } - w.ctx, w.cancelFunc = context.WithCancel(parentCtx) + // The parent's cancel is a stop, not the end of the work: a context canceled with it would refuse + // the shutdown's last push before it reached the remote, and fail every save it carried. The work + // ends workerShutdownBudget after the stop, or once the loop has exited, whichever is first. + work, endWork := context.WithCancel(context.WithoutCancel(parentCtx)) + w.stop, w.cancelFunc = context.WithCancel(parentCtx) + context.AfterFunc(w.stop, func() { time.AfterFunc(workerShutdownBudget, endWork) }) + w.ctx = work w.started = true w.mu.Unlock() @@ -566,6 +575,7 @@ func (w *BranchWorker) Start(parentCtx context.Context) error { w.wg.Add(1) go func() { defer w.wg.Done() + defer endWork() w.processEvents() w.settleCommitRequestsAfterExit() }() @@ -573,6 +583,23 @@ func (w *BranchWorker) Start(parentCtx context.Context) error { return nil } +// workerShutdownBudget bounds a stopping worker's last work: closing its open windows, pushing its +// pending writes and settling its saves. Workers stop together, so it bounds the whole drain. With the +// CommitRequest outcomes written after it (WorkerManager.SetAfterWorkersStopped), it fits the chart's +// 20s terminationGracePeriodSeconds. A push still running then is cut off, and the saves it carried +// fail as a refused last push fails them (failUnpushedCommitRequests). +// +//nolint:gochecknoglobals // a var, not a const, so a test can shorten it, like gitCallTimeout. +var workerShutdownBudget = 10 * time.Second + +// stopping is closed once the worker is asked to stop. +func (w *BranchWorker) stopping() <-chan struct{} { + if w.stop == nil { + return w.ctx.Done() + } + return w.stop.Done() +} + // Stop gracefully shuts down the worker. func (w *BranchWorker) Stop() { w.mu.Lock() @@ -1165,7 +1192,7 @@ func (l *branchWorkerEventLoop) run() { for { commitC, pushC, attachC, refusalC := l.timerChannels() select { - case <-l.w.ctx.Done(): + case <-l.w.stopping(): l.handleShutdown() l.publishLoopState(0) return diff --git a/internal/git/shutdown_drain_test.go b/internal/git/shutdown_drain_test.go new file mode 100644 index 00000000..f89d9489 --- /dev/null +++ b/internal/git/shutdown_drain_test.go @@ -0,0 +1,137 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "context" + "testing" + "time" + + gogit "github.com/go-git/go-git/v6" + "github.com/go-git/go-git/v6/plumbing/object" + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/ConfigButler/gitops-reverser/internal/types" +) + +// A clean shutdown closes the open window and pushes it. The manager stops a worker by canceling the +// context it started it with, and a push made on that context was refused before it reached the +// remote: the save attached to the window failed with "worker stopped before the commit reached the +// remote", and the edit was lost with the checkout. Only a real server shows it; go-git's file:// +// transport never reads the context. +func TestShutdown_TheLastWindowIsPushedOnceTheManagerCancels(t *testing.T) { + f := newLedgerFixture(t, "shutdown-push", true) + worker := startDrainWorker(t, f) + parent, cancel := context.WithCancel(context.Background()) + require.NoError(t, worker.Start(parent)) + + const message = "save: the window open at shutdown" + collectAttached(t, worker, "saved-at-shutdown", message) + + cancel() + worker.Stop() + + res, ok := outcome(t, worker) + require.True(t, ok, "the save is settled before the worker exits") + require.NoError(t, res.Err) + assert.Equal(t, FinalizeCommitted, res.Outcome) + assert.Contains(t, configMapFileNames(t, f.repoDir), "saved-at-shutdown.yaml") + tip := remoteTip(t, f.repoDir) + assert.Equal(t, tip.Hash.String(), res.Commit, "the save reports the commit on the remote") + assert.Equal(t, message, tip.Message) +} + +// The shutdown's work is bounded: a remote that stalls the last push is cut off at +// workerShutdownBudget, and the save it carried fails rather than holding the process past its grace. +func TestShutdown_AStalledLastPushEndsAtTheBudget(t *testing.T) { + f := newLedgerFixture(t, "shutdown-stall", true) + worker := startDrainWorker(t, f) + budget := workerShutdownBudget + workerShutdownBudget = stallBound + t.Cleanup(func() { workerShutdownBudget = budget }) + parent, cancel := context.WithCancel(context.Background()) + require.NoError(t, worker.Start(parent)) + + collectAttached(t, worker, "stalled-at-shutdown", "save: stalled") + stallPhase(f.sim, isReceivePack) + t.Cleanup(func() { f.sim.around.Store(nil) }) + + cancel() + require.NoError(t, requireReturns(t, "Stop", func() error { + worker.Stop() + return nil + })) + + res, ok := outcome(t, worker) + require.True(t, ok) + require.Error(t, res.Err) + assert.Contains(t, res.Err.Error(), "worker stopped") +} + +// startDrainWorker readies the fixture's worker to run its own loop on team-a. +func startDrainWorker(t *testing.T, f *ledgerFixture) *BranchWorker { + t.Helper() + createPlainGitTarget(t, f.worker, "team-a", "team-a") + t.Cleanup(f.worker.Stop) + return f.worker +} + +// collectAttached opens a window with one edit and attaches a request to it, then waits until the +// running loop reports it collecting. The request's timers are long, so only a stop closes it. +func collectAttached(t *testing.T, worker *BranchWorker, name, message string) { + t.Helper() + require.True(t, worker.Enqueue(configMapTargetEvent(name, "alice", "team-a"))) + req := attachReq("alice", time.Hour) + req.Message = message + worker.EnqueueAttach(req) + require.Eventually(t, func() bool { + return worker.LookupCommitRequestPhase("default", crName, "uid-"+crName) == PhaseCollectingWindow + }, 10*time.Second, 10*time.Millisecond, "the request attaches to the open window") +} + +// remoteTip is the commit at main in a bare repository. +func remoteTip(t *testing.T, bareRepoDir string) *object.Commit { + t.Helper() + repo, err := gogit.PlainOpen(bareRepoDir) + require.NoError(t, err) + ref, err := repo.Reference("refs/heads/main", true) + require.NoError(t, err) + commit, err := repo.CommitObject(ref.Hash()) + require.NoError(t, err) + return commit +} + +// The manager runs what SetAfterWorkersStopped injected once every worker has stopped, so the outcomes +// the workers settled are there to read, and on a context of its own, since the manager's is canceled. +func TestShutdown_TheManagerRunsTheSettleOnceTheWorkersStopped(t *testing.T) { + f := newLedgerFixture(t, "shutdown-manager", true) + worker := startDrainWorker(t, f) + m := NewWorkerManager(f.worker.Client, logr.Discard(), BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) + m.workers[BranchKey{Repository: f.sim.RepoURL, Branch: "main"}] = worker + parent, cancel := context.WithCancel(context.Background()) + require.NoError(t, worker.Start(parent)) + collectAttached(t, worker, "settled-after-stop", "save: read after the stop") + + var ( + settled FinalizeResult + resolved bool + ctxErr error + deadline time.Time + ) + m.SetAfterWorkersStopped(func(ctx context.Context) { + settled, resolved = outcome(t, worker) + ctxErr = ctx.Err() + deadline, _ = ctx.Deadline() + }) + done := make(chan error, 1) + go func() { done <- m.Start(parent) }() + cancel() + require.NoError(t, <-done) + + require.True(t, resolved, "the worker had settled the save before the step ran") + assert.Equal(t, FinalizeCommitted, settled.Outcome) + require.NoError(t, ctxErr, "the step runs on a live context") + assert.WithinDuration(t, time.Now().Add(afterWorkersStoppedBudget), deadline, afterWorkersStoppedBudget) +} diff --git a/internal/git/worker_manager.go b/internal/git/worker_manager.go index 4acf3f9e..566cf12e 100644 --- a/internal/git/worker_manager.go +++ b/internal/git/worker_manager.go @@ -178,6 +178,9 @@ type WorkerManager struct { // CommitRequestChanges. commitRequestChanges chan CommitRequestChange + // afterWorkersStopped runs once every worker has stopped at shutdown. See SetAfterWorkersStopped. + afterWorkersStopped func(ctx context.Context) + // claims is which GitBranch holds each repository branch (branch_owner.go). The GitBranch // reconciler writes it; EnsureWorker and the sweep read it. Reach it through Claims, which makes // one for a manager not built by NewWorkerManager. @@ -296,6 +299,22 @@ func (m *WorkerManager) SetApplyBarrier(barrier ApplyBarrierFunc) { m.applyBarrier = barrier } +// SetAfterWorkersStopped injects what runs at shutdown once every worker has stopped: the workers +// have closed their windows, pushed and settled their saves, and nothing will settle another. It runs +// on a context of its own, bounded by afterWorkersStoppedBudget, because the manager's is canceled by +// then. The CommitRequest controllers write the outcomes the workers decided there, while the caches +// they read still run: those stop only once this runnable has returned. Like SetMapper, it is called +// once at startup. +func (m *WorkerManager) SetAfterWorkersStopped(fn func(ctx context.Context)) { + m.mu.Lock() + defer m.mu.Unlock() + m.afterWorkersStopped = fn +} + +// afterWorkersStoppedBudget bounds SetAfterWorkersStopped's step. With workerShutdownBudget before it, +// it fits the chart's 20s terminationGracePeriodSeconds. +const afterWorkersStoppedBudget = 5 * time.Second + // SetSSHHostKeyConfig injects the SSH host-key resolution config used by every worker's credential // reads. Like SetMapper, it is called once at startup before any worker is created. func (m *WorkerManager) SetSSHHostKeyConfig(cfg SSHHostKeyConfig) { @@ -904,15 +923,31 @@ func (m *WorkerManager) Start(ctx context.Context) error { m.workers = make(map[BranchKey]*BranchWorker) m.mu.Unlock() - // Stop all workers gracefully + // Every worker saw the cancel and is draining already, so these waits overlap: the whole drain takes + // about as long as the slowest worker, at most workerShutdownBudget. for key, worker := range workers { m.Log.Info("Stopping worker for shutdown", "key", key.String()) worker.Stop() } m.Log.Info("WorkerManager stopped") + m.runAfterWorkersStopped(ctx) return nil } +// runAfterWorkersStopped runs the step SetAfterWorkersStopped injected, on a context that is not the +// manager's canceled one. +func (m *WorkerManager) runAfterWorkersStopped(ctx context.Context) { + m.mu.RLock() + fn := m.afterWorkersStopped + m.mu.RUnlock() + if fn == nil { + return + } + settleCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), afterWorkersStoppedBudget) + defer cancel() + fn(settleCtx) +} + // WorkerSweepInterval is the floor under the orphan sweep. It is a FLOOR and not the mechanism: a // deleted GitTarget's reconcile sweeps immediately, and this catches what that path cannot see. const WorkerSweepInterval = time.Minute diff --git a/internal/git/worker_repo_identity_test.go b/internal/git/worker_repo_identity_test.go index 11ab2fde..28e2ff4f 100644 --- a/internal/git/worker_repo_identity_test.go +++ b/internal/git/worker_repo_identity_test.go @@ -69,7 +69,7 @@ func TestEnsureWorker_KeepsTheWorkerWhenTheRepositoryIsUnchanged(t *testing.T) { second, ok := manager.WorkerForGitBranch(testGitBranch("repo1", "gitops-system", "main")) require.True(t, ok) assert.Same(t, first, second, "a steady tick must leave the worker exactly where it was") - assert.NoError(t, first.ctx.Err(), "and must not have stopped it") + assert.NoError(t, first.stop.Err(), "and must not have stopped it") } // TestEnsureWorker_ReplacesTheWorkerWhenTheProviderIsRecreated is the ordinary repoint: the @@ -90,7 +90,7 @@ func TestEnsureWorker_ReplacesTheWorkerWhenTheProviderIsRecreated(t *testing.T) require.True(t, ok) assert.NotSame(t, old, replacement, "the slot holds the worker for the repository named NOW") assert.Equal(t, after, replacement.repo) - assert.Error(t, old.ctx.Err(), "the worker for the old repository must have been stopped") + assert.Error(t, old.stop.Err(), "the worker for the old repository must have been stopped") } // TestEnsureWorker_ReplacesOnARecreateThatKeepsTheURL. The UID is the identity, and it changes on diff --git a/internal/watch/event_router.go b/internal/watch/event_router.go index 2ac5f531..8cf18060 100644 --- a/internal/watch/event_router.go +++ b/internal/watch/event_router.go @@ -161,6 +161,21 @@ func (r *EventRouter) WithdrawCommitRequest( return git.FinalizeResult{Branch: branch, Phase: phase}, false, nil } +// LookupCommitRequestOutcome reports the outcome the worker that holds a CommitRequest decided, and +// false while it decided none or no worker holds the request. It sends nothing: it is how the +// controller reads what a stopped worker settled, at shutdown, without attaching anything anew. +func (r *EventRouter) LookupCommitRequestOutcome( + _ context.Context, + attach git.AttachCommitRequest, +) (git.FinalizeResult, bool, error) { + owner, ok := r.WorkerManager.CommitRequestOwner(attach.Namespace, attach.Name, attach.UID) + if !ok { + return git.FinalizeResult{}, false, nil + } + result, resolved := owner.LookupCommitRequestOutcome(attach.Namespace, attach.Name, attach.UID) + return result, resolved, nil +} + // LookupCommitRequestApply reports where the apply a CommitRequest asked for stands, from the // branch worker of its GitTarget, and false while that worker knows nothing of it or none runs. func (r *EventRouter) LookupCommitRequestApply( diff --git a/internal/watch/event_router_outcome_test.go b/internal/watch/event_router_outcome_test.go new file mode 100644 index 00000000..b95dc539 --- /dev/null +++ b/internal/watch/event_router_outcome_test.go @@ -0,0 +1,78 @@ +// SPDX-License-Identifier: Apache-2.0 + +package watch + +import ( + "context" + "testing" + "time" + + meta "github.com/fluxcd/pkg/apis/meta" + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8stypes "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + "github.com/ConfigButler/gitops-reverser/internal/git" + "github.com/ConfigButler/gitops-reverser/internal/types" +) + +// TestLookupCommitRequestOutcome_ReadsWhatAStoppedWorkerDecided: the shutdown reads the outcome the +// worker that held the request decided, after every worker has stopped, without sending it anything. +// A request no worker holds has none. +func TestLookupCommitRequestOutcome_ReadsWhatAStoppedWorkerDecided(t *testing.T) { + scheme := eventRouterScheme(t) + provider := &configv1alpha3.GitProvider{ + ObjectMeta: metav1.ObjectMeta{Name: "team-b-provider", Namespace: "team-a"}, + Spec: configv1alpha3.GitProviderSpec{URL: "file:///tmp/does-not-need-to-exist"}, + } + gitTarget := &configv1alpha3.GitTarget{ + ObjectMeta: metav1.ObjectMeta{Name: "team-a-config", Namespace: "team-a"}, + Spec: configv1alpha3.GitTargetSpec{ + GitBranchRef: meta.NamespacedObjectReference{Name: gitBranchNameForTest("team-b-provider", "main")}, + }, + } + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(provider, gitTarget).Build() + workerManager := git.NewWorkerManager( + client, logr.Discard(), git.BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) + router := NewEventRouter(workerManager, nil, client, logr.Discard()) + attach := saveAttach("team-a-config", "team-a") + + _, resolved, err := router.LookupCommitRequestOutcome(context.Background(), attach) + require.NoError(t, err) + require.False(t, resolved, "no worker holds the request") + + ctx, cancel := context.WithCancel(context.Background()) + stopped := make(chan struct{}) + go func() { + defer close(stopped) + _ = workerManager.Start(ctx) + }() + time.Sleep(100 * time.Millisecond) // allow the manager to record its context + repo := git.RepoIdentity{URL: "file:///tmp/does-not-need-to-exist"} + gitBranch := gitBranchForTest("team-b-provider", "team-a", "main") + workerManager.Claims().Observe(git.BranchKeyFor(repo.URL, "main"), git.BranchClaimant{ + GitBranch: gitBranch, UID: "uid-team-b-branch", Created: time.Unix(0, 0), + }, true) + require.NoError(t, workerManager.EnsureWorker(ctx, git.WorkerRequest{ + GitBranch: gitBranch, + Provider: k8stypes.NamespacedName{Namespace: "team-a", Name: "team-b-provider"}, + Branch: "main", + Repo: repo, + })) + require.Eventually(t, func() bool { + _, resolved, err := router.ServiceCommitRequest(context.Background(), attach) + require.NoError(t, err) + return resolved + }, 5*time.Second, 50*time.Millisecond, "the worker resolves the request") + cancel() + <-stopped + + result, resolved, err := router.LookupCommitRequestOutcome(context.Background(), attach) + require.NoError(t, err) + require.True(t, resolved, "the stopped worker still answers for the request it decided") + assert.Equal(t, git.FinalizeNoOpenWindow, result.Outcome) +} diff --git a/test/e2e/commit_request_restart_e2e_test.go b/test/e2e/commit_request_restart_e2e_test.go new file mode 100644 index 00000000..7f816331 --- /dev/null +++ b/test/e2e/commit_request_restart_e2e_test.go @@ -0,0 +1,125 @@ +// SPDX-License-Identifier: Apache-2.0 + +package e2e + +import ( + "fmt" + "os" + "path/filepath" + "strings" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// A save in progress when the controller stops: its CommitRequest is attached to an open commit +// window, and the window is still collecting. A clean shutdown closes the window under the request's +// message, pushes it, and writes the request's outcome before the process exits, so the save ends as +// one commit with the request's message and the request reports it. +// +// The pipeline is the Save Request suite's: a GitTarget with a long spec.commit.window on a Gitea +// repo of its own, so main does not exist until the window is closed, and the one commit is this +// spec's. The request's own timers are minutes long, so only the shutdown can close the window. +// +// The outcome is read while the controller is scaled to zero: the stopping process wrote it, and no +// new process could have. A rollout restart would not show that, and with the e2e manifest's default +// surge the new pod is Ready before the old one stops, so two operators would write the branch at once. +// +// Serial: scales the controller deployment, a singleton every concurrent spec depends on. See +// docs/spec/e2e-serial-registry.md. +var _ = Describe("Save Request across a controller restart", Label("commit-request-restart"), Serial, Ordered, + func() { + var ( + testNs string + repo *RepoArtifacts + gitProvName string + gitTargetName string + watchRuleName string + ) + + const ( + basePath = "e2e/commit-request-restart" + commitWindow = "300s" + // requestTimers are the request's attachTimeout and maxDuration, the schema's maximum: the + // window it attaches to is still collecting when the controller stops. + requestTimers = "5m" + ) + + BeforeAll(func() { + p := setupWindowedDeploymentPipeline("commit-request-restart", basePath, commitWindow) + testNs, repo, gitProvName, gitTargetName, watchRuleName = p.ns, p.repo, p.gitProvider, p.gitTarget, + p.watchRule + }) + + AfterAll(func() { + cleanupPipeline(testNs, gitProvName, gitTargetName, watchRuleName) + cleanupNamespace(testNs) + }) + + It("closes the save's window, pushes it and writes its outcome before the controller exits", func() { + seed := GinkgoRandomSeed() + deployName := fmt.Sprintf("commit-request-restart-deploy-%d", seed) + requestName := fmt.Sprintf("commit-request-restart-save-%d", seed) + message := fmt.Sprintf("save: settled across a restart, e2e seed %d", seed) + + By("opening a commit window with an edit, and attaching a CommitRequest to it") + applyScaleTestDeployment(testNs, deployName, 0) + _, err := kubectlRunWithStdin(testNs, + commitRequestManifest(testNs, requestName, gitTargetName, message, requestTimers), "apply", "-f", "-") + Expect(err).NotTo(HaveOccurred(), "failed to apply CommitRequest %s", requestName) + Eventually(func(g Gomega) { + reason := commitRequestField(g, testNs, requestName, + `{.status.conditions[?(@.type=="Reconciling")].reason}`) + g.Expect(reason).To(Equal("CollectingWindow"), "the request is attached to the open window\n%s", + readCommitRequestOutcome(g, testNs, requestName)) + }, 2*time.Minute, time.Second).Should(Succeed()) + Expect(remoteBranchHead(Default, repo.CheckoutDir)).To(BeEmpty(), + "the window is still collecting: nothing is on main") + + By("stopping the controller while the window collects") + deployment, err := controllerDeploymentName() + Expect(err).NotTo(HaveOccurred()) + scaleController(deployment, 0) + DeferCleanup(func() { scaleController(deployment, 1) }) + + By("reading the outcome the stopping controller wrote") + var reportedSHA string + Eventually(func(g Gomega) { + reportedSHA = expectCommitRequestCommitted(g, testNs, requestName, + recentCommitDiagnostics(repo.CheckoutDir, basePath)) + }, 30*time.Second, time.Second).Should(Succeed(), + "no controller runs now, so only the one that stopped can have written this") + + By("starting the controller again") + scaleController(deployment, 1) + + By("verifying the save is one commit with the request's message, and the request still says so") + Eventually(func(g Gomega) { + pullLatestRepoState(g, repo.CheckoutDir) + g.Expect(mustCommitCount(repo.CheckoutDir)).To(Equal(1), + "the save is one commit\n%s", recentCommitDiagnostics(repo.CheckoutDir, basePath)) + subject, logErr := gitRun(repo.CheckoutDir, "log", "-1", "--pretty=%B") + g.Expect(logErr).NotTo(HaveOccurred()) + g.Expect(strings.TrimSpace(subject)).To(Equal(message)) + headSHA, shaErr := gitRun(repo.CheckoutDir, "rev-parse", "HEAD") + g.Expect(shaErr).NotTo(HaveOccurred()) + g.Expect(strings.TrimSpace(headSHA)).To(Equal(reportedSHA)) + expected := filepath.Join(repo.CheckoutDir, basePath, + fmt.Sprintf("%s/apps/deployments/%s.yaml", testNs, deployName)) + _, statErr := os.Stat(expected) + g.Expect(statErr).NotTo(HaveOccurred(), "the edit is in the save's commit: %s", expected) + }, 2*time.Minute, 3*time.Second).Should(Succeed()) + Consistently(func(g Gomega) { + g.Expect(expectCommitRequestCommitted(g, testNs, requestName, "")).To(Equal(reportedSHA)) + pullLatestRepoState(g, repo.CheckoutDir) + g.Expect(mustCommitCount(repo.CheckoutDir)).To(Equal(1), + "the new controller makes no second commit for the save\n%s", + recentCommitDiagnostics(repo.CheckoutDir, basePath)) + }, 15*time.Second, 3*time.Second).Should(Succeed()) + + By("cleaning up the Deployment and CommitRequest") + _, _ = kubectlRunInNamespace(testNs, "delete", "deployment", deployName, "--ignore-not-found=true") + _, _ = kubectlRunInNamespace(testNs, "delete", "commitrequest", requestName, "--ignore-not-found=true") + }) + })