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 af65918b..3a62b133 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -17,9 +17,20 @@ should land before the next tag is cut (the last release is v0.55.0). Checked ag ## 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 3453bdd0..1cc485d8 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -3337,7 +3337,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 318490bc..03ec83c5 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 @@ -566,7 +569,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() @@ -575,6 +584,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() }() @@ -582,6 +592,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() @@ -1179,7 +1206,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") + }) + })