diff --git a/.coverage-baseline b/.coverage-baseline index 8d0afb62..e8398503 100644 --- a/.coverage-baseline +++ b/.coverage-baseline @@ -1 +1 @@ -82.6 +82.7 diff --git a/config/deployment.yaml b/config/deployment.yaml index 77d1516d..acc52a3e 100644 --- a/config/deployment.yaml +++ b/config/deployment.yaml @@ -134,7 +134,7 @@ spec: seccompProfile: type: RuntimeDefault serviceAccountName: gitops-reverser - terminationGracePeriodSeconds: 10 + terminationGracePeriodSeconds: 20 volumes: - emptyDir: {} name: tmp-dir diff --git a/docs/TODO.md b/docs/TODO.md index 3a62b133..1a69c5ab 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -13,7 +13,20 @@ should land before the next tag is cut (the last release is v0.55.0). Checked ag Before it is released, the reference model (`internal/git/window_model_test.go`) must cover both close orders, conflicting writes, delete and recreate, `attach_next`, zero-duration windows, a resync, shutdown, failed publication and replay, and buffer pressure, each with several windows - open. + open. The reviews of stages 1 and 2 add: + - **gate:** the model retains work across failed pushes and retries before several windows are + released. Today it publishes at once after every wake (`windowModelRun.wake`) and asserts no + pending writes after it (`drained`), so decided-but-uncommitted writes with windows open + (cooldown, failed push, replay) are never modeled; + - reach floors the model asserts: shutdown, flush and push hint with two or more windows open, + `attach_next` with others open, a reproof overlap, `path_conflict`; a concurrency-biased profile; + CI raised to about 20 seeds × 60 steps; + - model steps: a shutdown mid-run; a timer overdue but not yet delivered (the `!isOpen` guard in + `attachWaitingCommitRequests` survives its removal); a late stale applier write (the model + never withholds); + - the GitTarget-generation and checkout-reset parts of a footprint revision are untested; + - a counter for window fallbacks (`footprint_fallback` counts the windows it closes, not the + fallback). ## Applier claims and bi-directional editing @@ -31,6 +44,11 @@ should land before the next tag is cut (the last release is v0.55.0). Checked ag 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). +- [ ] **Gaps in the withheld-field record** (`internal/git/withheld_fields.go`): a field patch in the + apply window forgets the object's accepted version, so a withhold over it records nothing; + `objectKey` leaves out the GitTarget and the cluster (it predates #466); accepted versions held + for render fidelity (`recordUnpublishedEdits`) and resyncs with `Initial: false` are untested; + no counter shows how many fields are withheld. - [ ] **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 @@ -183,6 +201,9 @@ Their argument for each is in [gitops-api-open-asks.md](future/gitops-api-open-a four specs that look tested and are not, then the moves and merges. - [ ] `SetDefaultEventuallyTimeout` is global and set in 17 e2e files, so the last `Describe` body built wins; use per-spec timeouts (`Eventually(...).WithTimeout`). +- [ ] e2e for the commit-window stack: two authors editing disjoint files at once, then a + `path_conflict`; E5 extended to the withheld-field carry; a restart of a source cluster's + controller with a save in flight; `catchUp: AfterPush` at shutdown. - [ ] e2e Test A: a delete missed while the controller is scaled to 0; then Tests B and C. [e2e-coverage-gaps-and-improvements-plan.md](design/e2e-coverage-gaps-and-improvements-plan.md). - [ ] Re-enable `goconst` with a path-scoped exclusion (`test/`, `internal/git/commit.go`) instead diff --git a/docs/UPGRADING.md b/docs/UPGRADING.md index d57ecc6a..b3d49439 100644 --- a/docs/UPGRADING.md +++ b/docs/UPGRADING.md @@ -13,11 +13,14 @@ A request's progress and outcome are written as soon as its branch worker report at a two-second poll. Once the worker has registered a request, the poll is only a ten-second fallback. -A save's own writes now land in its window. A client that waits for its request to be registered -before it writes, as `SaveInProgress` asks, used to learn of the registration up to two seconds -late, which is the whole default `maxDuration`. The window then closed on the earlier writes under -the save's message, and the save's own writes committed under the generated message, or under the -next save's message. +A client that waits for its request to be registered before it writes, as `SaveInProgress` asks, +learns of the registration sooner. It used to learn of it up to two seconds late, which is the whole +default `maxDuration`, so the window often closed on the earlier writes under the save's message, +and the save's own writes committed under the generated message, or under the next save's message. +That is less likely now, not ruled out: the window's `maxDuration` still runs from the attach, +whatever the client does, so a write delivered late (its audit fact names its author late), or a +write of another author or `GitTarget` that closes the window, can still split a save. A save that +must hold every write gives its request a longer `spec.window.maxDuration`. `SaveInProgress` is decided when the worker registers the request, after its author was found. A client that relies on it to keep two of its saves apart waits for the `Reconciling` reason diff --git a/docs/configuration.md b/docs/configuration.md index 1cc485d8..950002b5 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -3157,9 +3157,9 @@ things split it: A write after the close commits on its own, under the generated message. For such a folder, give the request a longer window, `spec.window.maxDuration: "10s"` for example, or rolling collection (see [the request window](#the-request-window-specwindow)); the commit lands that much later. -- **Other writes on the branch.** A branch worker holds one window at a time, so a write by another - author, or for another `GitTarget` on the same branch, closes yours early, with the request's - message, and your later writes commit under the generated message. +- **Other writes on the branch.** A write for another `GitTarget` on the same branch, or by another + author to a file the branch worker cannot prove apart from your window's files, closes yours + early, with the request's message, and your later writes commit under the generated message. The request then ends `Committed` with the writes its window held. diff --git a/docs/design/applier-echoes.md b/docs/design/applier-echoes.md index 6a7f30bc..fea1de3d 100644 --- a/docs/design/applier-echoes.md +++ b/docs/design/applier-echoes.md @@ -683,14 +683,20 @@ the last on 2026-10-28: may be absent. Every later event of the object, mirrored or not and the `Repair`'s own landing included, and every resync snapshot of it, is restored against that record: a field still holding the applier's value was carried over and is set back to the accepted value; a field - holding any other value was changed by someone and leaves the record; a delete clears it. This is - the desired view of + holding any other value was changed by someone and leaves the record; a delete clears it. A + delete is accepted too, as the object's absence: an apply that makes the object again over it is + recorded as the whole object, and every later event of the object is kept out of Git, and every + snapshot of it left out of a resync, until a delete (the `Repair`'s own) clears it. The record + also ends when the applier claim is removed, and when a resync of the object's collection no + longer finds the object. This is the desired view of [`window-field-ownership-and-apply-hold.md`](window-field-ownership-and-apply-hold.md#convergence-is-a-separate-operation), limited to the fields a withheld apply reverted: commits stay whole objects, and no window owns - fields. Three limits are accepted. A list is one field, so an edit to another item of a list the + fields. Four limits are accepted. A list is one field, so an edit to another item of a list the apply reverted still carries the reverted item. An author setting a field to the applier's value - cannot be told from carrying it, and is set back. The record lives in memory, so a restart loses - it ([`withheld_fields.go`](../../internal/git/withheld_fields.go)). + cannot be told from carrying it, and is set back. An update of an object the apply made again + cannot be told from carrying it either, so it is kept out of Git until the object is deleted; + an applier that does not prune keeps it out for as long as the record lives. The record lives + in memory, so a restart loses it ([`withheld_fields.go`](../../internal/git/withheld_fields.go)). ## References diff --git a/docs/design/concurrent-author-windows.md b/docs/design/concurrent-author-windows.md index a5ab1a8e..07cebe7f 100644 --- a/docs/design/concurrent-author-windows.md +++ b/docs/design/concurrent-author-windows.md @@ -1,8 +1,9 @@ # Design: one commit window per author, open at the same time Status: **stages 1 and 2 built** (`internal/git/window_set.go`, `footprint.go`, -`window_admission.go`; see [what stage 2 built](#what-stage-2-built)). Stage 2 ships without the -startup flag of stage 4, so stage 3 lands before the next release (`docs/TODO.md`). Raised by +`window_admission.go`; see [what stage 2 built](#what-stage-2-built)); **stages 3 to 5 proposed**. +Stage 2 is on with no startup flag, so **stage 3 gates the next release**: it must land before the +next tag is cut (`docs/TODO.md`, "Before the next release"). Raised by gitops-api's review of the source-side save (PR #456): with saves created straight from a browser, a second person's edit splits the first person's change into two commits. This page sets out the change that fixes that, the rules it must keep, how to test @@ -189,6 +190,7 @@ unrelated push split a save that is protected today. | `shutdown` | every window | none | | `buffer_limit` | every window | none; closing only the largest is an open question | | `path_conflict` | the windows whose footprint the new write overlaps | none | +| `footprint_fallback` | every window but the newest, when a commit outside its footprint makes the worker hold one window at a time | none | | `identity_change` | every other window | stays for writes of another `GitTarget` and for unproven footprints | | `idle_timeout`, `max_duration`, `attach_next` | that window only | none | @@ -284,21 +286,25 @@ window, so they are proven against today's behavior before they are used to judg ## Stages -1. **Extract window management without changing behavior.** Move the window bookkeeping +Stages 1 and 2 are built; stages 3, 4 and 5 are proposed. Stage 3 gates the next release, because +stage 2 shipped without stage 4's flag. + +1. **Built. Extract window management without changing behavior.** Move the window bookkeeping (`openWindow`, its timers, attach and finalize) behind one type that holds a set of windows, with one window allowed. Add the controllable clock, the opening sequence number, and the reference model. Every existing test passes unchanged. -2. **Allow several windows for proven-independent destinations.** Writes keep whole-object commits +2. **Built. Allow several windows for proven-independent destinations.** Writes keep whole-object commits and apply behavior stays as it is. Writes of another `GitTarget` keep closing other windows (`identity_change`), and every layout in the footprint table's "no" rows stays serial. -3. **Test the failure boundaries.** Both close orders, conflicting writes, delete and recreate, - `attach_next`, zero-duration windows, resync, shutdown, failed publication and replay, and buffer - pressure, each against the reference model. -4. **Ship behind a startup flag**, off by default at first. The flag is read once, at start; changing +3. **Proposed; gates the next release. Test the failure boundaries.** Both close orders, + conflicting writes, delete and recreate, `attach_next`, zero-duration windows, resync, shutdown, + failed publication and replay, and buffer pressure, each against the reference model, which + first has to keep work pending across failed pushes and retries. +4. **Proposed. Ship behind a startup flag**, off by default at first. The flag is read once, at start; changing it takes a restart, and the restart drains every open window through the normal shutdown path (opening order, close reason `shutdown`), so there is no live transition to get right. Measure concurrent-window counts, forced splits by reason, save outcomes, and memory before turning it on by default. -5. **Reconsider the follow-ups** in +5. **Proposed. Reconsider the follow-ups** in [window-field-ownership-and-apply-hold.md](window-field-ownership-and-apply-hold.md), using those measurements. Field ownership and the apply hold are separate changes, each with its own state-machine review. @@ -320,7 +326,9 @@ The rules above, with these choices where the page left room: write before it. After a reset or a `GitTarget` change the proof is of another tree, and the commit is not checked. A refusal fails the write as terminal: the worktree is undone, its save fails with the paths, the worker logs `SingleWindowFallback` and holds one window at a time until - it restarts, and a resync of the `GitTarget` is asked for. + it restarts, closing every open window but the newest (`footprint_fallback`), and a resync of the + `GitTarget` is asked for. Each commit checked logs `FootprintChecked` with its tree, and the + reference model requires one for every commit it gave a footprint. - **A request's own author's window stands in no one's way.** A save waiting while another author's window is open records a mismatch only when no window of its own author is open. - **The window ID** (`window`) is in the open, close and finalized log lines, with the branch. diff --git a/docs/interpreting-metrics.md b/docs/interpreting-metrics.md index 8691e221..aa4e2a5b 100644 --- a/docs/interpreting-metrics.md +++ b/docs/interpreting-metrics.md @@ -206,7 +206,7 @@ boundary, the commit, the push. Background: | `git_queue_refusals_total` | counter | `gitbranch_namespace`, `gitbranch_name`, `provider_namespace`, `provider_name`, `kind` | Work refused at enqueue: the queue was full, intake was paused through an outage, or the worker was stopping. `kind` is `write` / `attach` / `resync` / `refresh`. The producer keeps a refused item and offers it again: the watch delivers a write again from its cursor, the controller re-sends a save, a resync is gathered again, and the next reconcile asks for a refresh. What it can cost, when its watch cursor expires before the write comes back, is the write's own commit and, under `prune.mode: OnEvent`, a deleted object's removal from Git. | | `git_commit_failures_total` | counter | `gitbranch_namespace`, `gitbranch_name`, `provider_namespace`, `provider_name`, `reason` | A live commit window that died between routing and pushing; a resync's failure is `git_resync_failures_total`. `reason` is `refused` (a Git path a human must fix) / `error`. Every increment is a window's events lost until the next resync. An unreachable remote does not count: a decided window waits for the publication retry instead. | | `git_writes_dropped_total` | counter | `gitbranch_namespace`, `gitbranch_name`, `provider_namespace`, `provider_name`, `reason` | A live event the branch worker dropped because its `GitTarget`'s render fidelity was not established. `reason` is `render_fidelity_diverged`: a rendered value does not match the live object, and every write is dropped until a fresh replay measures the folder clean. Or it is `render_fidelity_pending`: a collection of the `GitTarget` has not finished its replay, which closes every collection of that target until it does. A live event that meets a pending gate is held, not dropped, and committed with its author once the gate opens; it is counted here only when the hold gives up, 10 minutes after the first one was held or once the branch holds more than its retained-byte budget, and the worker then asks for a fresh snapshot, which commits the current state as the operator (a deletion comes back only under prune mode `Always`). Nothing replays a dropped write, so each one is an edit missing from Git until the object changes again or its collection replays. The worker logs the first drop under each gate status at Info, naming the collections the gate waits for, and each hold, release and give-up at Info. | -| `git_commit_windows_total` | counter | `gittarget_namespace`, `gittarget_name`, `close_reason`, `timer_source` | One per closed commit window, whether or not it then produced a commit. `close_reason` is `idle_timeout` / `max_duration` / `attach_next` / `identity_change` / `path_conflict` / `buffer_limit` / `resync_before_apply` / `foreign_commit` (a Git host said the branch may have moved, so the window is published now) / `apply_due` (an apply came due for an applier on the branch, so the window's edits are published before the apply is sent) / `shutdown`; `timer_source` is `target` or `commit_request`. See [tuning commit windows](#tuning-commit-windows). | +| `git_commit_windows_total` | counter | `gittarget_namespace`, `gittarget_name`, `close_reason`, `timer_source` | One per closed commit window, whether or not it then produced a commit. `close_reason` is `idle_timeout` / `max_duration` / `attach_next` / `identity_change` / `path_conflict` / `footprint_fallback` (a commit changed a file outside its window's footprint, so the worker holds one window at a time from then on, and closes all but the newest of the windows open) / `buffer_limit` / `resync_before_apply` / `foreign_commit` (a Git host said the branch may have moved, so the window is published now) / `apply_due` (an apply came due for an applier on the branch, so the window's edits are published before the apply is sent) / `shutdown`; `timer_source` is `target` or `commit_request`. See [tuning commit windows](#tuning-commit-windows). | | `git_commit_window_duration_seconds` | histogram | `gittarget_namespace`, `gittarget_name` | How long each window collected: from the write that opened it to the start of its finalize. Its count equals `git_commit_windows_total` summed over `close_reason` and `timer_source`. | | `git_queue_depth` | gauge | `gitbranch_namespace`, `gitbranch_name`, `provider_namespace`, `provider_name` | Pending + in-flight + committed-but-unpushed. Read at scrape time. | | `git_retained_bytes` | gauge | `gitbranch_namespace`, `gitbranch_name`, `provider_namespace`, `provider_name` | What the worker holds until a push lands, in the units of the retained-byte budget: the open window, the retained writes and waiting saves, each with a 1 KiB per-item charge. Serialized bytes, not heap. | diff --git a/internal/controller/commitrequest_decide.go b/internal/controller/commitrequest_decide.go index 95e31126..c973265e 100644 --- a/internal/controller/commitrequest_decide.go +++ b/internal/controller/commitrequest_decide.go @@ -91,6 +91,10 @@ func (c *commitRequestCore) decide( } current, written, err := c.writeOutcome(ctx, p, cr, o, afterClaim) if err != nil || !written { + if err == nil && current != nil && commitRequestIsTerminal(current) { + // Another write got there first, and the request is finished all the same. + c.inFlight.forget(p, client.ObjectKeyFromObject(current)) + } return commitRequestRequeueNow, err } c.inFlight.forget(p, client.ObjectKeyFromObject(current)) @@ -115,15 +119,19 @@ func (c *commitRequestCore) decide( return c.expireFinished(ctx, p, current) } -// writeOutcome writes o onto cr, re-reading it on a conflict. written is false when the fresh object -// already holds an outcome, or, before the claim, anything another path wrote; then the next step -// decides from that. +// writeOutcome writes o onto cr, re-reading it on a conflict. written is false when the object, cr or +// the fresh one, already holds an outcome, or, before the claim, anything another path wrote; then it +// returns that object, and the next step decides from it. It returns no object when the request is +// gone or was replaced. func (c *commitRequestCore) writeOutcome( ctx context.Context, p commitRequestPath, cr *configbutleraiv1alpha3.CommitRequest, o commitRequestOutcome, afterClaim bool, ) (*configbutleraiv1alpha3.CommitRequest, bool, error) { current := cr for attempt := 1; ; attempt++ { + if commitRequestIsTerminal(current) { + return current, false, nil + } o.apply(current) err := p.store.updateStatus(ctx, current) if err == nil { @@ -135,10 +143,10 @@ func (c *commitRequestCore) writeOutcome( } fresh, same, getErr := c.reread(ctx, p, current) if getErr != nil || !same { - return current, false, getErr + return nil, false, getErr } - if commitRequestIsTerminal(fresh) || (!afterClaim && len(fresh.Status.Conditions) > 0) { - return current, false, nil + if !afterClaim && len(fresh.Status.Conditions) > 0 { + return fresh, false, nil } current = fresh } diff --git a/internal/controller/commitrequest_server.go b/internal/controller/commitrequest_server.go index 885da10e..895f7379 100644 --- a/internal/controller/commitrequest_server.go +++ b/internal/controller/commitrequest_server.go @@ -120,6 +120,10 @@ type CommitRequestServer struct { streamsOnce sync.Once streams *commitRequestStreams + + // after is the timer a request's goroutine waits on between steps: time.After when nil. A test + // steps it, so only a wake or the test can take the next step. + after func(time.Duration) <-chan time.Time } // commitRequestWatcher is the watch of one KubeProvider's CommitRequests, in the namespaces it lends. @@ -576,6 +580,10 @@ func (s *CommitRequestServer) consider( } key := k8stypes.NamespacedName{Namespace: u.GetNamespace(), Name: u.GetName()} core := s.core() + wait := s.after + if wait == nil { + wait = time.After + } go func() { defer watcher.serving.Delete(u.GetUID()) woken, stop := s.Wakes.watch(u.GetUID()) @@ -592,7 +600,7 @@ func (s *CommitRequestServer) consider( } select { case <-ctx.Done(): - case <-time.After(after): + case <-wait(after): case <-woken: } } diff --git a/internal/controller/commitrequest_settle_test.go b/internal/controller/commitrequest_settle_test.go index 77c7310b..76084cf1 100644 --- a/internal/controller/commitrequest_settle_test.go +++ b/internal/controller/commitrequest_settle_test.go @@ -170,11 +170,74 @@ func TestCommitRequestsInFlight_AConflictReappliesTheOutcome(t *testing.T) { assert.Equal(t, "c0ffee", fetchCommitRequest(t, c, "save").Status.Commit) } +// A conflict whose fresh object holds an outcome already, written by a step that ran before the stop, +// leaves that outcome, and the request is finished: Settle forgets it rather than listing it as left +// without one. +func TestCommitRequestsInFlight_AnOutcomeWrittenFirstIsForgotten(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-- + first := &configv1alpha3.CommitRequest{} + if err := c.Get(ctx, client.ObjectKeyFromObject(cr), first); err != nil { + return err + } + applyFinalizeResultToStatus(first, git.FinalizeResult{ + Outcome: git.FinalizeCommitted, Commit: "beef", Branch: "main", + }, nil) + if err := c.SubResource(sub).Update(ctx, first, opts...); err != nil { + return err + } + 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, "beef", fetchCommitRequest(t, c, "save").Status.Commit, "the outcome written first stays") + assert.Empty(t, inFlight.remembered(), "a request that holds an outcome is finished") +} + +// A finished request is never written over: the write of an outcome checks for one before it applies +// its own. +func TestCommitRequestCore_AnOutcomeIsNotWrittenOverAFinishedRequest(t *testing.T) { + c := newCommitRequestClient(t, nil, withReadyCommitted(newCommitRequest("save"))) + before := fetchCommitRequest(t, c, "save") + core := &commitRequestCore{reader: c} + + _, written, err := core.writeOutcome(context.Background(), controlPath(c), before.DeepCopy(), + commitRequestOutcome{result: git.FinalizeResult{Branch: "main", Err: git.ErrBranchWorkerStopped}, + err: git.ErrBranchWorkerStopped}, true) + + require.NoError(t, err) + assert.False(t, written) + after := fetchCommitRequest(t, c, "save") + assert.Equal(t, before.ResourceVersion, after.ResourceVersion, "nothing is written") + assert.False(t, conditionIsTrue(after.Status.Conditions, ConditionTypeStalled)) +} + // 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. +// watch has stopped with the rest of the controllers. A control-cluster request of the same name is +// another request, and is settled too: a request is remembered by its path as well as its name. func TestCommitRequestsInFlight_ASourceClusterRequestIsSettledOnItsCluster(t *testing.T) { provider := kubeconfigProvider("edge", true) - c := newCommitRequestClient(t, nil, mirroringTarget("edge"), provider) + control := newCommitRequest("save") + control.Spec.GitTargetRef.Name = "control-config" + controlTarget := &configv1alpha3.GitTarget{ + ObjectMeta: metav1.ObjectMeta{Name: "control-config", Namespace: "default"}, + } + c := newCommitRequestClient(t, nil, mirroringTarget("edge"), provider, controlTarget, withInProgress(control)) gvr := commitRequestGVR() source := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{gvr: "CommitRequestList"}, sourceCommitRequest("default", "save")) @@ -199,13 +262,19 @@ func TestCommitRequestsInFlight_ASourceClusterRequestIsSettledOnItsCluster(t *te } 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") + serveStep(t, &commitRequestCore{reader: c, finalizer: worker, inFlight: inFlight}, controlPath(c), "save") + require.Equal(t, string(git.PhaseCollectingWindow), commitRequestReady(t, c, "save").Reason, + "the control-cluster request is attached too") cancel() s.stopAll() + require.Len(t, inFlight.remembered(), 2) worker.stop(shutdownCommit) inFlight.Settle(context.Background()) assert.Equal(t, crReasonCommitted, readReady(), "written on the source cluster") + assert.Equal(t, crReasonCommitted, commitRequestReady(t, c, "save").Reason, "and on the control cluster") + assert.Empty(t, inFlight.remembered()) } // A nil set remembers nothing, and settles nothing. diff --git a/internal/controller/commitrequest_wakes_test.go b/internal/controller/commitrequest_wakes_test.go index 476c81b8..c9ea6f5d 100644 --- a/internal/controller/commitrequest_wakes_test.go +++ b/internal/controller/commitrequest_wakes_test.go @@ -20,41 +20,24 @@ import ( "github.com/ConfigButler/gitops-reverser/internal/git" ) -// simulatedWorker answers like a branch worker that registers a request registerAfter after its first -// attach and pushes its commit resolveAfter after that attach, reporting each change as the worker's -// loop does. +// simulatedWorker answers like a branch worker, and moves only when the test says so: report changes +// its state and reports the change, as the worker's loop does. type simulatedWorker struct { - registerAfter, resolveAfter time.Duration - changes chan git.CommitRequestChange + change git.CommitRequestChange + changes chan git.CommitRequestChange mu sync.Mutex - started bool + steps int phase git.CommitRequestPhase resolved bool - pushedAt time.Time } func (w *simulatedWorker) ServiceCommitRequest( - _ context.Context, attach git.AttachCommitRequest, + context.Context, git.AttachCommitRequest, ) (git.FinalizeResult, bool, error) { w.mu.Lock() defer w.mu.Unlock() - if !w.started { - w.started = true - change := git.CommitRequestChange{Namespace: attach.Namespace, Name: attach.Name, UID: attach.UID} - time.AfterFunc(w.registerAfter, func() { - w.mu.Lock() - w.phase = git.PhaseWaitingForWindow - w.mu.Unlock() - w.changes <- change - }) - time.AfterFunc(w.resolveAfter, func() { - w.mu.Lock() - w.resolved, w.pushedAt = true, time.Now() - w.mu.Unlock() - w.changes <- change - }) - } + w.steps++ if w.resolved { return git.FinalizeResult{Outcome: git.FinalizeCommitted, Commit: "c0ffee", Branch: "main"}, true, nil } @@ -84,34 +67,78 @@ func (w *simulatedWorker) LookupCommitRequestApply( return git.SaveApply{}, false, nil } -// saveLatency is what a client waiting on a request sees: create to WaitingForWindow, and push to -// Ready. -type saveLatency struct{ registered, ready time.Duration } +// report applies a change to the worker's state and reports it. The channel is buffered: on the +// polled path nothing reads it. +func (w *simulatedWorker) report(change func(*simulatedWorker)) { + w.mu.Lock() + change(w) + w.mu.Unlock() + w.changes <- w.change +} + +func (w *simulatedWorker) stepsTaken() int { + w.mu.Lock() + defer w.mu.Unlock() + return w.steps +} + +// steppedTimers stands in for time.After between a request's steps: none fires until the test fires +// them, so a step taken without fire was taken by a wake. +type steppedTimers struct { + mu sync.Mutex + pending []chan time.Time +} + +func (s *steppedTimers) after(time.Duration) <-chan time.Time { + s.mu.Lock() + defer s.mu.Unlock() + ch := make(chan time.Time, 1) + s.pending = append(s.pending, ch) + return ch +} + +func (s *steppedTimers) armed() bool { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.pending) > 0 +} + +// fire runs out every timer armed so far. +func (s *steppedTimers) fire() { + s.mu.Lock() + defer s.mu.Unlock() + for _, ch := range s.pending { + ch <- time.Now() + } + s.pending = nil +} -// measureSaveLatency serves one request through a source cluster's watch, the production loop of -// CommitRequestServer.consider, with or without the worker's changes waking it. -func measureSaveLatency(t *testing.T, woken bool) saveLatency { +// servedSave is one request served through a source cluster's watch, the production loop of +// CommitRequestServer.consider, on stepped timers, with or without the worker's changes waking it. +type servedSave struct { + t *testing.T + c client.Client + cr *configv1alpha3.CommitRequest + worker *simulatedWorker + timers *steppedTimers +} + +func serveSave(t *testing.T, woken bool) *servedSave { t.Helper() cr := newCommitRequest("save") cr.Spec.GitTargetRef.Namespace = "default" c := newCommitRequestClient(t, nil, cr, mirroringTarget("edge"), kubeconfigProvider("edge", true)) worker := &simulatedWorker{ - registerAfter: 50 * time.Millisecond, resolveAfter: 3 * time.Second, + change: git.CommitRequestChange{Namespace: "default", Name: "save", UID: string(cr.UID)}, changes: make(chan git.CommitRequestChange, 4), } ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - server := &CommitRequestServer{Client: c, Finalizer: worker} + t.Cleanup(cancel) + timers := &steppedTimers{} + server := &CommitRequestServer{Client: c, Finalizer: worker, after: timers.after} if woken { server.Wakes = NewCommitRequestWakes(worker.changes) go func() { _ = server.Wakes.Start(ctx) }() - } else { - // Nothing listens to the worker's changes: the steps poll. - go func() { - for range worker.changes { - continue - } - }() } watcher := &commitRequestWatcher{ provider: types.NamespacedName{Namespace: "default", Name: "edge"}, path: watcherPath(c, "edge", false), @@ -120,45 +147,63 @@ func measureSaveLatency(t *testing.T, woken bool) saveLatency { u.SetNamespace("default") u.SetName("save") u.SetUID(cr.UID) - - created := time.Now() server.consider(ctx, watcher, u) - var latency saveLatency - require.Eventually(t, func() bool { - var got configv1alpha3.CommitRequest - if c.Get(ctx, client.ObjectKeyFromObject(cr), &got) != nil { - return false - } - reconciling := apimeta.FindStatusCondition(got.Status.Conditions, ConditionTypeReconciling) - if latency.registered == 0 && reconciling != nil && reconciling.Reason == string(git.PhaseWaitingForWindow) { - latency.registered = time.Since(created) - } - ready := apimeta.FindStatusCondition(got.Status.Conditions, ConditionTypeReady) - if ready == nil || ready.Status != metav1.ConditionTrue { - return false - } - worker.mu.Lock() - latency.ready = time.Since(worker.pushedAt) - worker.mu.Unlock() - return true - }, 10*time.Second, 5*time.Millisecond) - return latency + s := &servedSave{t: t, c: c, cr: cr, worker: worker, timers: timers} + require.Eventually(t, func() bool { return worker.stepsTaken() > 0 && timers.armed() }, 10*time.Second, + time.Millisecond, "the first step attaches the request and waits") + return s +} + +func (s *servedSave) conditions() []metav1.Condition { + var got configv1alpha3.CommitRequest + if s.c.Get(context.Background(), client.ObjectKeyFromObject(s.cr), &got) != nil { + return nil + } + return got.Status.Conditions +} + +func (s *servedSave) registered() bool { + reconciling := apimeta.FindStatusCondition(s.conditions(), ConditionTypeReconciling) + return reconciling != nil && reconciling.Reason == string(git.PhaseWaitingForWindow) +} + +func (s *servedSave) ready() bool { + return apimeta.IsStatusConditionTrue(s.conditions(), ConditionTypeReady) } +func registerRequest(w *simulatedWorker) { w.phase = git.PhaseWaitingForWindow } +func pushRequest(w *simulatedWorker) { w.resolved = true } + // TestCommitRequestWakes_AWorkersChangeTakesTheNextStepAtOnce pins gitops-api ask #40: a request's -// progress follows its worker's changes instead of a fixed poll. The worker here registers the request -// 50ms after its attach and pushes 3s after it; with the changes waking the steps, a client waiting -// for WaitingForWindow and then for Ready sees each within a few milliseconds, not the poll. +// progress follows its worker's changes instead of a fixed poll. No timer between steps fires unless +// the test fires it, so a step taken without one was taken by the worker's change. func TestCommitRequestWakes_AWorkersChangeTakesTheNextStepAtOnce(t *testing.T) { - polled := measureSaveLatency(t, false) - woken := measureSaveLatency(t, true) - t.Logf("create -> WaitingForWindow: polled %s, woken %s", polled.registered, woken.registered) - t.Logf("push -> Ready: polled %s, woken %s", polled.ready, woken.ready) - - assert.GreaterOrEqual(t, polled.registered, commitRequestPollInterval, "polled, the registration waits a poll") - assert.Less(t, woken.registered, 250*time.Millisecond, "woken, it is reported at once") - assert.Less(t, woken.ready, 250*time.Millisecond, "woken, the push is reported at once") + t.Run("woken, each change takes the next step", func(t *testing.T) { + s := serveSave(t, true) + s.worker.report(registerRequest) + require.Eventually(t, s.registered, 10*time.Second, time.Millisecond, + "the registration is reported with no poll fired") + s.worker.report(pushRequest) + require.Eventually(t, s.ready, 10*time.Second, time.Millisecond, + "the push is reported with no poll fired") + }) + + t.Run("polled, each change waits for the poll", func(t *testing.T) { + s := serveSave(t, false) + s.worker.report(registerRequest) + assert.Never(t, s.registered, 200*time.Millisecond, 5*time.Millisecond, + "nothing takes the step before the poll") + s.timers.fire() + require.Eventually(t, s.registered, 10*time.Second, time.Millisecond, "the poll takes it") + + require.Eventually(t, s.timers.armed, 10*time.Second, time.Millisecond) + s.worker.report(pushRequest) + assert.Never(t, s.ready, 200*time.Millisecond, 5*time.Millisecond, + "nothing takes the step before the poll") + s.timers.fire() + require.Eventually(t, s.ready, 10*time.Second, time.Millisecond, "the poll takes it") + }) } // TestCommitRequestWakes_AChangeMidStepIsKeptForTheNextWait pins that a change reported while the diff --git a/internal/git/applier_claim.go b/internal/git/applier_claim.go index 523e32ec..7b94db09 100644 --- a/internal/git/applier_claim.go +++ b/internal/git/applier_claim.go @@ -736,6 +736,7 @@ func (l *branchWorkerEventLoop) syncAppliers(targets []branchTarget) { for ref, state := range l.appliers { if _, ok := named[ref]; !ok { l.dropSaves(state) + l.dropWithheldOf(state) if state.inflight != nil { state.inflight.abandon() } diff --git a/internal/git/apply_window_carry_test.go b/internal/git/apply_window_carry_test.go index 3a6089f7..94aa364f 100644 --- a/internal/git/apply_window_carry_test.go +++ b/internal/git/apply_window_carry_test.go @@ -9,6 +9,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" "github.com/ConfigButler/gitops-reverser/internal/manifestanalyzer" @@ -73,7 +74,7 @@ func carryWithheld(t *testing.T, slug string) *echoFixture { f.writeWindow(carryApply("a1", "b1")) require.True(t, f.echoApplier().repair, "the apply over E1 is withheld, and a Repair is pending") - require.Len(t, f.loop.withheld[carryKey()], 1, "the field the apply reverted is recorded") + require.Len(t, f.loop.withheld[carryKey()].Fields, 1, "the field the apply reverted is recorded") return f } @@ -113,7 +114,7 @@ func TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit(t *testing.T) assert.Contains(t, f.onMain(echoFile("y")), "a: a2", "alice's commit of b keeps her own E1, not the withheld apply's a1") assert.Contains(t, f.onMain(echoFile("y")), "b: b2") - assert.Len(t, f.loop.withheld[carryKey()], 1, "the cluster still runs a1: the record stays") + assert.Len(t, f.loop.withheld[carryKey()].Fields, 1, "the cluster still runs a1: the record stays") }) t.Run("an edit of another author does not carry it either", func(t *testing.T) { @@ -164,6 +165,93 @@ func TestApplyWindow_TheWithheldRecordEnds(t *testing.T) { }) } +// carryDeleteThenRecreate publishes alice's delete of y inside the open apply window, then has the +// applier make y again, late: the apply is withheld, a Repair is pending, and the cluster runs y while +// Git holds none. +func carryDeleteThenRecreate(t *testing.T, slug string) *echoFixture { + t.Helper() + f := carryFlushOverAnEdit(t, slug) + f.publish() + deleted := carryEdit("alice", "a2", "b1") + deleted.Operation = string(itypes.OperationDelete) + f.writeWindow(deleted) + f.publish() + require.Empty(t, f.onMain(echoFile("y")), "precondition: alice's delete is published") + + recreated := carryApply("a1", "b1") + recreated.Operation = string(itypes.OperationCreate) + f.writeWindow(recreated) + require.True(t, f.echoApplier().repair, "the apply that made y again is withheld, and a Repair is pending") + return f +} + +// carryLabel is bob's kubectl label on y as the applier made it again: the whole object it carries +// is the apply's. +func carryLabel() Event { + event := carryApply("a1", "b1") + event.UserInfo.Username = "bob" + event.LastFieldManagers = []string{"kubectl-label"} + event.LastFieldManagedAt = time.Now().Add(2 * time.Hour) + event.Object.SetLabels(map[string]string{"team": "a"}) + return event +} + +// publishAny closes the open windows, if any, and pushes what is pending. +func (f *echoFixture) publishAny() { + f.t.Helper() + f.loop.lastPushAt = time.Time{} + f.loop.finalizeOpenWindow() + f.loop.pushPending() +} + +// TestApplyWindow_ADeleteInTheWindowStaysDeleted pins a delete as the last accepted version: an +// apply that makes the object again, late, over the delete is withheld, and until the Repair's own +// delete lands no later event of the object, by anyone, brings it back into Git. +func TestApplyWindow_ADeleteInTheWindowStaysDeleted(t *testing.T) { + t.Run("a later edit does not bring it back", func(t *testing.T) { + f := carryDeleteThenRecreate(t, "carry-delete-label") + f.writeWindow(carryLabel()) + f.publishAny() + assert.Empty(t, f.onMain(echoFile("y")), + "bob labelled the object the withheld apply made; committing it would undo alice's delete") + require.Len(t, f.loop.withheld[carryKey()].Fields, 1, "the record holds the object out") + + repaired := carryApply("a1", "b1") + repaired.Operation = string(itypes.OperationDelete) + f.writeWindow(repaired) // the Repair deletes y on the cluster + f.publishAny() + assert.Empty(t, f.loop.withheld, "the Repair's delete clears the record") + assert.Empty(t, f.onMain(echoFile("y"))) + + created := carryEdit("bob", "a9", "b9") + created.Operation = string(itypes.OperationCreate) + f.writeWindow(created) + f.publishAny() + assert.Contains(t, f.onMain(echoFile("y")), "a: a9", "an object made after the Repair is mirrored") + }) + + t.Run("a resync does not bring it back", func(t *testing.T) { + f := carryDeleteThenRecreate(t, "carry-delete-resync") + snapshot := carryLabel() + req := &ResyncRequest{ + Desired: []manifestanalyzer.DesiredResource{ + {Resource: snapshot.Identifier, Object: snapshot.Object}, + echoDesired("x", "x2"), + }, + GitTargetName: echoTarget, + GitTargetNamespace: "default", + Initial: true, + Result: make(chan ResyncResult, 1), + } + f.loop.handleQueueItem(resyncItem(req)) + require.NoError(t, (<-req.Result).Err) + f.publishAny() + + assert.Empty(t, f.onMain(echoFile("y")), "the snapshot holds y only because the apply made it again") + assert.Len(t, req.Desired, 2, "the snapshot itself is left as it was") + }) +} + // TestApplyWindow_AResyncAfterAWithholdKeepsTheEdit pins the record on a resync: its snapshot holds // what the cluster runs, the withheld apply's a1 included, and is set back as a live event is. func TestApplyWindow_AResyncAfterAWithholdKeepsTheEdit(t *testing.T) { @@ -217,7 +305,14 @@ func TestWithheldFields_RecordAndRestore(t *testing.T) { Accepted: fieldValue{Value: []interface{}{"one", "two"}}, Applier: fieldValue{Value: []interface{}{"one"}}, }, - }, loop.withheld[key]) + }, loop.withheld[key].Fields) + + // A later object that carries every value the apply wrote is set back whole: extra, which the + // accepted version did not have, is removed again. + carried := loop.restoreWithheld(key, withheld.Object.DeepCopy()) + assert.NotContains(t, carried.Object, "extra", "a field the accepted version lacked is removed") + assert.Equal(t, map[string]interface{}{"a": "a2", "c": "c1"}, carried.Object["data"]) + require.Len(t, loop.withheld[key].Fields, 4, "every field still holds the apply's value") // A later object: b added, data.a carried, data.c still missing, extra changed by someone, the // list carried. @@ -230,7 +325,7 @@ func TestWithheldFields_RecordAndRestore(t *testing.T) { assert.Equal(t, map[string]interface{}{"a": "a2", "b": "b2", "c": "c1"}, restored.Object["data"]) assert.Equal(t, "y", restored.Object["extra"], "a field someone changed is theirs") assert.Equal(t, []interface{}{"one", "two"}, restored.Object["spec"].(map[string]interface{})["list"]) - assert.Len(t, loop.withheld[key], 3, "extra left the record") + assert.Len(t, loop.withheld[key].Fields, 3, "extra left the record") // A second withhold, over a version without data at all, replaces the fields inside it. loop.keepAccepted(Event{Identifier: accepted.Identifier, Object: restored}) @@ -239,8 +334,8 @@ func TestWithheldFields_RecordAndRestore(t *testing.T) { again.Object.Object["spec"] = map[string]interface{}{"list": []interface{}{"one"}} again.Object.Object["extra"] = "y" loop.recordWithheld(again) - paths := make([][]string, 0, len(loop.withheld[key])) - for _, field := range loop.withheld[key] { + paths := make([][]string, 0, len(loop.withheld[key].Fields)) + for _, field := range loop.withheld[key].Fields { paths = append(paths, field.Path) } assert.ElementsMatch(t, [][]string{{"data"}, {"spec", "list"}}, paths) @@ -253,3 +348,126 @@ func TestWithheldFields_RecordAndRestore(t *testing.T) { assert.Same(t, unrelated, loop.restoreWithheld(key, unrelated)) assert.Empty(t, loop.withheld) } + +// carryData is author's live update of y, holding data. +func carryData(author string, data map[string]interface{}) Event { + event := carryEdit(author, "", "") + event.Object.Object["data"] = data + return event +} + +// TestApplyWindow_TheCommonOrderKeepsTheEdit is S1 in its most common order, on the live path: the +// attempt is out, the API edit lands, then the apply over it. The edit is accepted by the apply window +// alone, with no flush in progress, and a later edit of another field that carries the apply's value +// is set back to the edit's. +func TestApplyWindow_TheCommonOrderKeepsTheEdit(t *testing.T) { + loop, _ := sentAttempt(t) + loop.handleLiveEvents(&WriteRequest{Events: []Event{ + windowEvent("alpha", "3", "kubectl-edit"), + windowEvent("alpha", "2", FluxFieldManager), + }}) + label := windowEvent("alpha", "2", "kubectl-label") + label.Object.SetLabels(map[string]string{"team": "a"}) + loop.handleLiveEvents(&WriteRequest{Events: []Event{label}}) + + require.NotNil(t, loop.windows.only()) + require.Len(t, loop.windows.only().pathToEvent, 1) + for _, event := range loop.windows.only().pathToEvent { + replicas, _, _ := unstructured.NestedString(event.Object.Object, "spec", "replicas") + assert.Equal(t, "3", replicas, "the label edit carried the withheld apply's 2; Git keeps the edit's 3") + assert.Equal(t, map[string]string{"team": "a"}, event.Object.GetLabels()) + } +} + +// TestApplyWindow_AFieldTheEditRemovedStaysRemoved: alice removes data.b, the apply over her edit +// puts it back, and the next edit, which carries it, does not commit it. +func TestApplyWindow_AFieldTheEditRemovedStaysRemoved(t *testing.T) { + f := carryFlushOverAnEdit(t, "carry-removed-field") + f.writeWindow(carryData("alice", map[string]interface{}{"a": "a2"})) // alice removes b + f.publish() + require.NotContains(t, f.onMain(echoFile("y")), "b: b1", "precondition: her removal is published") + + f.writeWindow(carryApply("a1", "b1")) + require.True(t, f.echoApplier().repair, "the apply over her edit is withheld") + f.writeWindow(carryData("bob", map[string]interface{}{"a": "a1", "b": "b1", "c": "c1"})) + f.publish() + + got := f.onMain(echoFile("y")) + assert.Contains(t, got, "a: a2") + assert.Contains(t, got, "c: c1") + assert.NotContains(t, got, "b: b1", "alice removed b; only the withheld apply put it back") +} + +// TestApplyWindow_ChainedWithholdsKeepEveryEdit: a withhold, alice's edit of b, a second withheld +// apply, of a newer commit, over both, then bob's edit of c. Each withhold is compared with the +// event as restored, and the edit between them is accepted as restored, so Git keeps a2 and b2, and +// the record holds a and b, each against the value accepted for it. +func TestApplyWindow_ChainedWithholdsKeepEveryEdit(t *testing.T) { + f := carryWithheld(t, "carry-chained") + f.writeWindow(carryEdit("alice", "a1", "b2")) // alice edits b; the event carries the apply's a1 + + second := carryApply("a3", "b1") + second.LastFieldManagedAt = time.Now().Add(2 * time.Hour) + f.writeWindow(second) + f.writeWindow(carryData("bob", map[string]interface{}{"a": "a3", "b": "b1", "c": "c1"})) + f.publish() + + got := f.onMain(echoFile("y")) + assert.Contains(t, got, "a: a2", "alice's E1, accepted before both applies") + assert.Contains(t, got, "b: b2", "alice's edit of b") + assert.Contains(t, got, "c: c1") + assert.ElementsMatch(t, []withheldField{ + {Path: []string{"data", "a"}, Accepted: fieldValue{Value: "a2"}, Applier: fieldValue{Value: "a3"}}, + {Path: []string{"data", "b"}, Accepted: fieldValue{Value: "b2"}, Applier: fieldValue{Value: "b1"}}, + }, f.loop.withheld[carryKey()].Fields) +} + +// TestApplyWindow_TheRecordEndsWhenNothingWillRepairIt pins the ends of a record that no event of +// the object shows: the applier claim is gone, so no Repair will put the edit back; a resync of the +// object's collection no longer finds the object. +func TestApplyWindow_TheRecordEndsWhenNothingWillRepairIt(t *testing.T) { + t.Run("the applier claim is removed", func(t *testing.T) { + f := carryWithheld(t, "carry-claim-gone") + f.loop.syncAppliers(nil) + assert.Empty(t, f.loop.withheld) + }) + + t.Run("a resync no longer finds the object", func(t *testing.T) { + f := carryWithheld(t, "carry-resync-gone") + resync := func(scope *ResyncScope) { + req := &ResyncRequest{ + Desired: []manifestanalyzer.DesiredResource{echoDesired("x", "x2")}, + GitTargetName: echoTarget, + GitTargetNamespace: "default", + Scope: scope, + Initial: true, + Result: make(chan ResyncResult, 1), + } + f.loop.handleQueueItem(resyncItem(req)) + require.NoError(t, (<-req.Result).Err) + } + deployments := ResyncScopeFor( + schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}, "default") + resync(&deployments) + assert.Len(t, f.loop.withheld, 1, "a resync of another collection says nothing about y") + resync(nil) + assert.Empty(t, f.loop.withheld, "y is gone from the cluster") + }) +} + +// TestApplyWindow_AcceptedVersionsEndWithTheWindow: once the apply window has closed and the appliers +// are serviced, no last accepted version is kept, so a write to the object is never withheld again. +func TestApplyWindow_AcceptedVersionsEndWithTheWindow(t *testing.T) { + loop, req := sentAttempt(t) + loop.handleLiveEvents(&WriteRequest{Events: []Event{windowEvent("alpha", "3", "kubectl-edit")}}) + require.Len(t, loop.accepted, 1) + + report(req, ApplySucceeded) + loop.serviceAppliers() + require.Len(t, loop.accepted, 1, "the window stays open for its grace") + for _, state := range loop.appliers { + state.window.closesAt = time.Now().Add(-time.Millisecond) + } + loop.serviceAppliers() + assert.Empty(t, loop.accepted) +} diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index 03ec83c5..d7d175f6 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -443,6 +443,9 @@ const ( // newer revision, finds overlapping another window's: it commits before the write that overlaps // it (footprint.go). windowFinalizeReasonPathConflict windowFinalizeReason = "path-conflict" + // windowFinalizeReasonFootprintFallback closes the windows open side by side when a commit outside + // its footprint makes the worker fall back to one window at a time (fallBackToSingleWindow). + windowFinalizeReasonFootprintFallback windowFinalizeReason = "footprint-fallback" ) // NewBranchWorker creates a worker for a (provider, branch) combination. A zero @@ -1438,11 +1441,15 @@ func (l *branchWorkerEventLoop) appendLiveEvent(event Event) { // target earlier go first, once its gate opened. // // Every event is first restored against the fields withheld applies reverted, mirrored or not, so -// the event carries none of them into a commit (withheld_fields.go). +// the event carries none of them into a commit, and an event of an object only a withheld apply made +// again over a delete does not reach it (withheld_fields.go). func (l *branchWorkerEventLoop) mirrors(event *Event) bool { - l.restoreWithheldEvent(event) + keptOut := l.restoreWithheldEvent(event) target := itypes.NewResourceReference(event.GitTargetName, event.GitTargetNamespace) l.serviceHeld(target) + if keptOut { + return false + } status, closed := l.w.renderFidelityClosed(event.GitTargetName, event.GitTargetNamespace) if closed && status.State == RenderFidelityFalse { l.dropAtRenderFidelity(target, status, event.Identifier.String()) @@ -2976,31 +2983,17 @@ const ( ) // metricLabel is the commit_windows_total `close_reason` value for this reason. The log field keeps -// its hyphenated form; a label value follows the snake_case the other labels use. +// its hyphenated form; a label value follows the snake_case the other labels use. Only +// identity_change is named apart from its log form, author-or-target-change. func (r windowFinalizeReason) metricLabel() string { switch r { - case windowFinalizeReasonIdleTimeout: - return "idle_timeout" - case windowFinalizeReasonMaxDuration: - return "max_duration" - case windowFinalizeReasonAttachNext: - return "attach_next" case windowFinalizeReasonIdentityChange: return "identity_change" - case windowFinalizeReasonBufferLimit: - return "buffer_limit" - case windowFinalizeReasonResyncBeforeApply: - return "resync_before_apply" - case windowFinalizeReasonShutdown: - return "shutdown" - case windowFinalizeReasonForeignCommit: - return "foreign_commit" - case windowFinalizeReasonApplyDue: - return "apply_due" - case windowFinalizeReasonPathConflict: - return "path_conflict" - case windowFinalizeReasonUnspecified: - return "unspecified" + case windowFinalizeReasonIdleTimeout, windowFinalizeReasonMaxDuration, windowFinalizeReasonAttachNext, + windowFinalizeReasonBufferLimit, windowFinalizeReasonResyncBeforeApply, windowFinalizeReasonShutdown, + windowFinalizeReasonForeignCommit, windowFinalizeReasonApplyDue, windowFinalizeReasonPathConflict, + windowFinalizeReasonFootprintFallback, windowFinalizeReasonUnspecified: + return strings.ReplaceAll(string(r), "-", "_") default: return "unspecified" } diff --git a/internal/git/commit_executor.go b/internal/git/commit_executor.go index 5b9c4bf6..b60b8108 100644 --- a/internal/git/commit_executor.go +++ b/internal/git/commit_executor.go @@ -415,7 +415,8 @@ func (w *BranchWorker) executeEventWrite( // Before the commit, not after it: a window that changes a file outside its footprint may have // been open beside one that changes that file too, and committing it could put an older value // over a newer one. The caller undoes the worktree. - if err := w.checkFootprint(worktree, pendingWrite, first); err != nil { + checked, err := w.checkFootprint(worktree, pendingWrite, first) + if err != nil { return 0, plumbing.ZeroHash, messageResolutionUnsupported, err } @@ -432,6 +433,7 @@ func (w *BranchWorker) executeEventWrite( return 0, plumbing.ZeroHash, messageResolutionUnsupported, fmt.Errorf("failed to create commit: %w", err) } + w.logFootprintChecked(repo, hash, pendingWrite, checked) log.FromContext(ctx).Info( "git commit created", diff --git a/internal/git/commit_window_metrics_test.go b/internal/git/commit_window_metrics_test.go index 3defc365..0233fc08 100644 --- a/internal/git/commit_window_metrics_test.go +++ b/internal/git/commit_window_metrics_test.go @@ -102,6 +102,28 @@ func TestCommitWindowMetrics_EachCloseReasonIsCountedUnderItsOwnName(t *testing. requireOneObservationPerWindow(t, reader, 4) } +// TestCommitWindowMetrics_EveryCloseReasonHasItsLabel pins the close_reason value of every reason, as +// docs/interpreting-metrics.md lists them. +func TestCommitWindowMetrics_EveryCloseReasonHasItsLabel(t *testing.T) { + for reason, label := range map[windowFinalizeReason]string{ + windowFinalizeReasonIdleTimeout: "idle_timeout", + windowFinalizeReasonMaxDuration: "max_duration", + windowFinalizeReasonAttachNext: "attach_next", + windowFinalizeReasonIdentityChange: "identity_change", + windowFinalizeReasonBufferLimit: "buffer_limit", + windowFinalizeReasonResyncBeforeApply: "resync_before_apply", + windowFinalizeReasonShutdown: "shutdown", + windowFinalizeReasonForeignCommit: "foreign_commit", + windowFinalizeReasonApplyDue: "apply_due", + windowFinalizeReasonPathConflict: "path_conflict", + windowFinalizeReasonFootprintFallback: "footprint_fallback", + windowFinalizeReasonUnspecified: "unspecified", + windowFinalizeReason("made-up"): "unspecified", + } { + assert.Equal(t, label, reason.metricLabel(), string(reason)) + } +} + // TestCommitWindowMetrics_AnAttachKeepsTheOpeningTimeAndCountsTheRequestsTimers pins the two // things an attach must not do: move the moment the window opened, which the attach DOES reset for // the idle interval, and leave the window counted under the target's timers after it replaced them. diff --git a/internal/git/footprint.go b/internal/git/footprint.go index a3d532b3..08b240a5 100644 --- a/internal/git/footprint.go +++ b/internal/git/footprint.go @@ -12,6 +12,7 @@ import ( "strings" gogit "github.com/go-git/go-git/v6" + "github.com/go-git/go-git/v6/plumbing" "github.com/go-git/go-git/v6/plumbing/object" "github.com/ConfigButler/gitops-reverser/internal/git/manifestedit" @@ -469,25 +470,48 @@ func (l *branchWorkerEventLoop) proveBeforeCommit(i int) { // checkFootprint refuses the commit about to be made from the staged changes when it changes a file // outside the write's footprint. It checks a write's first commit only, against a footprint proven // on the tree the commit is planned on (proveBeforeCommit). A proof of another tree, after a reset -// or a change of the GitTarget, says nothing about this one, and the commit is not checked. -func (w *BranchWorker) checkFootprint(worktree *gogit.Worktree, pendingWrite *PendingWrite, first bool) error { +// or a change of the GitTarget, says nothing about this one, and the commit is not checked +// (FootprintCheckSkipped). It reports whether it checked the commit: the caller logs +// FootprintChecked once the commit is made. +func (w *BranchWorker) checkFootprint( + worktree *gogit.Worktree, pendingWrite *PendingWrite, first bool, +) (bool, error) { f := pendingWrite.footprint if f == nil || !f.known() || !first { - return nil + return false, nil } if !w.provenOnCheckout(pendingWrite.Target(), f.revision) { w.Log.V(1).Info("Not checking a commit against its window's footprint: the proof is of another tree", - "footprint", f.String(), "revision", f.revision.String()) - return nil + "branch", w.Branch, "footprint", f.String(), "revision", f.revision.String(), + "decision", decisionFootprintCheckSkipped) + return false, nil } changed, err := stagedPaths(worktree) if err != nil { - return fmt.Errorf("read the staged changes: %w", err) + return false, fmt.Errorf("read the staged changes: %w", err) } if outside := f.outside(changed); len(outside) > 0 { - return &footprintViolationError{outside: outside, footprint: *f} + return false, &footprintViolationError{outside: outside, footprint: *f} } - return nil + return true, nil +} + +// logFootprintChecked logs the commit checkFootprint checked, once it is made. The tree names what +// was checked across a rebuild of the commit onto the same parent, which makes a new commit of it and +// checks it no more: its actions were decided already. +func (w *BranchWorker) logFootprintChecked( + repo *gogit.Repository, hash plumbing.Hash, pendingWrite *PendingWrite, checked bool, +) { + if !checked { + return + } + var tree plumbing.Hash + if commit, err := repo.CommitObject(hash); err == nil { + tree = commit.TreeHash + } + w.Log.Info("Checked a commit against its window's footprint", "branch", w.Branch, + "commit", hash.String(), "tree", tree.String(), "footprint", pendingWrite.footprint.String(), + "decision", decisionFootprintChecked) } // provenOnCheckout reports whether rev is the revision the checkout is at now. diff --git a/internal/git/resync_flush.go b/internal/git/resync_flush.go index 3395f190..ca72d98e 100644 --- a/internal/git/resync_flush.go +++ b/internal/git/resync_flush.go @@ -58,8 +58,9 @@ func (l *branchWorkerEventLoop) applyResync(req *ResyncRequest) { } pendingWrite.Desired, pendingWrite.ownedElsewhere = l.withoutOwnedElsewhere(req) // The snapshot holds what the cluster runs, a withheld apply's values included: they are set - // back as in a live event (withheld_fields.go). - pendingWrite.Desired = l.restoreWithheldSnapshot(pendingWrite.Desired) + // back as in a live event, and an object only a withheld apply made again is left out + // (withheld_fields.go). + pendingWrite.Desired = l.restoreWithheldSnapshot(req, pendingWrite.Desired) pendingWrite.Committed = new(bool) pendingWrite.origin.resync = req l.decide(*pendingWrite) diff --git a/internal/git/shutdown_drain_test.go b/internal/git/shutdown_drain_test.go index f89d9489..63c549a9 100644 --- a/internal/git/shutdown_drain_test.go +++ b/internal/git/shutdown_drain_test.go @@ -4,6 +4,10 @@ package git import ( "context" + "os" + "path/filepath" + "regexp" + "strconv" "testing" "time" @@ -12,7 +16,10 @@ import ( "github.com/go-logr/logr" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" "github.com/ConfigButler/gitops-reverser/internal/types" ) @@ -70,6 +77,53 @@ func TestShutdown_AStalledLastPushEndsAtTheBudget(t *testing.T) { assert.Contains(t, res.Err.Error(), "worker stopped") } +// A worker retired or replaced while the process runs is stopped with its parent still live, and gets +// the same last push a shutdown does: Stop is a stop, not the end of the work. +func TestShutdown_AWorkerStoppedWhileTheProcessRunsPushesItsLastWindow(t *testing.T) { + f := newLedgerFixture(t, "stop-push", true) + worker := startDrainWorker(t, f) + require.NoError(t, worker.Start(context.Background())) + + const message = "save: the window open at retirement" + collectAttached(t, worker, "saved-at-retirement", message) + + 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) + 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 work context ends once the loop has exited, not workerShutdownBudget later, whatever made it +// exit: a Stop, or a loop that never ran because the GitProvider could not be read. +func TestShutdown_TheWorkEndsWhenTheLoopExits(t *testing.T) { + t.Run("after Stop", func(t *testing.T) { + f := newLedgerFixture(t, "stop-ends-work", true) + worker := startDrainWorker(t, f) + require.NoError(t, worker.Start(context.Background())) + + worker.Stop() + + assert.Error(t, worker.ctx.Err(), "nothing runs on the work context any more") + }) + + t.Run("when the GitProvider cannot be read", func(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, configv1alpha3.AddToScheme(scheme)) + worker := NewBranchWorker(fake.NewClientBuilder().WithScheme(scheme).Build(), logr.Discard(), + "missing", "default", "main", RepoIdentity{URL: "file:///nowhere"}, nil, BranchWorkerLimits{}) + require.NoError(t, worker.Start(context.Background())) + t.Cleanup(worker.Stop) + + require.Eventually(t, func() bool { return worker.ctx.Err() != nil }, 5*time.Second, 10*time.Millisecond, + "the loop exited at once, and its work context with it") + }) +} + // startDrainWorker readies the fixture's worker to run its own loop on team-a. func startDrainWorker(t *testing.T, f *ledgerFixture) *BranchWorker { t.Helper() @@ -135,3 +189,31 @@ func TestShutdown_TheManagerRunsTheSettleOnceTheWorkersStopped(t *testing.T) { require.NoError(t, ctxErr, "the step runs on a live context") assert.WithinDuration(t, time.Now().Add(afterWorkersStoppedBudget), deadline, afterWorkersStoppedBudget) } + +// shutdownGraceMargin is what a stop needs beyond the workers' drain and the settle step: the caches +// and the rest of the runnables stopping, and the process exiting. +const shutdownGraceMargin = 5 * time.Second + +// Both deployments give the pod time for the whole drain: the workers' last push +// (workerShutdownBudget), then the CommitRequest outcomes written after it (afterWorkersStoppedBudget). +// A shorter grace period has the kubelet kill the pod before the outcomes are written: the e2e +// manifest once set 10s. +func TestShutdown_EveryDeploymentWaitsForTheDrain(t *testing.T) { + need := workerShutdownBudget + afterWorkersStoppedBudget + shutdownGraceMargin + grace := regexp.MustCompile(`(?m)^\s*terminationGracePeriodSeconds:\s*(\d+)\s*$`) + for _, manifest := range []string{ + "../../config/deployment.yaml", + "../../charts/gitops-reverser/templates/deployment.yaml", + } { + t.Run(filepath.Base(filepath.Dir(manifest))+"/"+filepath.Base(manifest), func(t *testing.T) { + content, err := os.ReadFile(manifest) + require.NoError(t, err) + match := grace.FindSubmatch(content) + require.NotNil(t, match, "the manifest sets terminationGracePeriodSeconds") + seconds, err := strconv.Atoi(string(match[1])) + require.NoError(t, err) + assert.GreaterOrEqual(t, time.Duration(seconds)*time.Second, need, + "terminationGracePeriodSeconds covers the workers' drain, the settle step and a margin") + }) + } +} diff --git a/internal/git/window_admission.go b/internal/git/window_admission.go index f86fbe0f..02b23e23 100644 --- a/internal/git/window_admission.go +++ b/internal/git/window_admission.go @@ -22,15 +22,19 @@ import ( // (FootprintReproved); windows whose new proofs overlap close first, with path_conflict. // // The decisions have names in the log lines, and the reference model reads them: WindowOpened, -// WriteJoinedWindow, FootprintReproved, WindowClosed (with its reason), and SingleWindowFallback, -// which is about the worker rather than a window. They name diagnostics, not a journal. +// WriteJoinedWindow, FootprintReproved, WindowClosed (with its reason), FootprintChecked and +// FootprintCheckSkipped (a commit checked against its window's footprint, with its hash, or not, its +// proof being of another tree), and SingleWindowFallback, which is about the worker rather than a +// window. They name diagnostics, not a journal. const ( - decisionWindowOpened = "WindowOpened" - decisionWriteJoinedWindow = "WriteJoinedWindow" - decisionFootprintReproved = "FootprintReproved" - decisionWindowClosed = "WindowClosed" - decisionSingleWindowFallback = "SingleWindowFallback" + decisionWindowOpened = "WindowOpened" + decisionWriteJoinedWindow = "WriteJoinedWindow" + decisionFootprintReproved = "FootprintReproved" + decisionWindowClosed = "WindowClosed" + decisionFootprintChecked = "FootprintChecked" + decisionFootprintCheckSkipped = "FootprintCheckSkipped" + decisionSingleWindowFallback = "SingleWindowFallback" ) // admitWrite closes every open window the write cannot be admitted beside, and returns the footprint @@ -179,7 +183,8 @@ func (l *branchWorkerEventLoop) closeBeforeWrite( // fallBackToSingleWindow is the answer to a commit refused for changing a file outside its window's // footprint: the footprint rules are wrong somewhere, so the worker stops trusting them and holds one // window at a time for the rest of its life, and a resync of the GitTarget brings Git back to the -// cluster whatever order the windows closed in. +// cluster whatever order the windows closed in. The windows those rules admitted side by side close +// now, oldest first, all but the newest, as the next write of another author would close them. func (l *branchWorkerEventLoop) fallBackToSingleWindow(pendingWrite PendingWrite, violation *footprintViolationError) { name, namespace := pendingWrite.windowTarget() l.singleWindow = true @@ -190,6 +195,10 @@ func (l *branchWorkerEventLoop) fallBackToSingleWindow(pendingWrite PendingWrite "outside", violation.outside, "footprint", violation.footprint.String(), "decision", decisionSingleWindowFallback) + open := l.windows.all() + for _, window := range open[:max(len(open)-1, 0)] { + l.finalizeWindow(window, windowFinalizeReasonFootprintFallback) + } if l.w.resyncRecheck != nil { l.w.resyncRecheck(itypes.NewResourceReference(name, namespace)) } diff --git a/internal/git/window_admission_test.go b/internal/git/window_admission_test.go index fc2f75bb..975ca258 100644 --- a/internal/git/window_admission_test.go +++ b/internal/git/window_admission_test.go @@ -3,6 +3,9 @@ package git import ( + "fmt" + "maps" + "strings" "testing" "time" @@ -142,6 +145,208 @@ func TestWindowAdmission_ClosingOneWindowArmsTheTimerForTheOthers(t *testing.T) assert.Equal(t, bobAt, f.loop.commitTimerAt) } +func TestWindowAdmission_ShutdownClosesEveryWindowOldestFirst(t *testing.T) { + f := newAdmissionFixture(t, "admission-shutdown", nil) + f.writeBy("alice", "x", "x2") + f.writeBy("bob", "z", "z2") + require.Len(t, f.loop.windows.all(), 2) + before := len(splitLines(gitIn(t, f.repoDir, "log", "--format=%H", "main"))) + + f.loop.handleShutdown() + + assert.Empty(t, f.loop.windows.all()) + assert.Equal(t, []string{"alice", "bob"}, + splitLines(gitIn(t, f.repoDir, "log", "--reverse", "--format=%an", "-2", "main")), + "one commit per window, in the order the windows opened") + assert.Len(t, splitLines(gitIn(t, f.repoDir, "log", "--format=%H", "main")), before+3, + "the bootstrap write held behind the cooldown, then the two windows") + assert.Equal(t, echoDocument("x", "x2"), f.onMain(echoFile("x"))) + assert.Equal(t, echoDocument("z", "z2"), f.onMain(echoFile("z"))) +} + +func TestWindowAdmission_AWriteOfAnotherGitTargetClosesTheOpenWindows(t *testing.T) { + f := newAdmissionFixture(t, "admission-other-target", nil) + reader := captureInfo(f.worker) + createGitTargetWithPruneMode(t, f.worker, "team-b", "team-b", configv1alpha3.PruneOnEvent) + other := func(author, value string) Event { + event := configMapTargetEvent("b", author, "team-b") + event.Operation = string(itypes.OperationUpdate) + event.Object.Object["data"] = map[string]interface{}{"key": value} + return event + } + window := f.loop.defaultWindow + f.loop.defaultWindow = commitWindowDefaults{} + f.writeWindow(other("carol", "b1")) // team-b's folder is bootstrapped by its first commit + require.Empty(t, f.loop.windows.all()) + f.loop.defaultWindow = window + + f.writeBy("alice", "x", "x2") + f.writeWindow(other("bob", "b2")) + + assert.Equal(t, []string{"bob"}, authorsOf(f.loop.windows.all())) + assert.Len(t, reader.matching(`"msg"="Finalizing open commit window" "reason"="author-or-target-change" `+ + `"windowAuthor"="alice"`), 1, "per-target encryption and bootstrap stay serial") +} + +// TestWindowAdmission_AFootprintBugIsRefusedOnTheRealPath drives a footprint bug through the loop: two +// windows open side by side, the index is corrupted so x is proven at z's path, and alice's window +// closes on its timer. Her commit is checked before it is made, refused, and undone; her save fails, +// the worker falls back, and a resync is asked for. +func TestWindowAdmission_AFootprintBugIsRefusedOnTheRealPath(t *testing.T) { + f := newAdmissionFixture(t, "admission-violation-path", nil) + var asked []itypes.ResourceReference + f.worker.resyncRecheck = func(target itypes.ResourceReference) { asked = append(asked, target) } + f.writeBy("alice", "x", "x2") + serviceAttach(f.loop, attachReq("alice", time.Hour)) + f.writeBy("bob", "q", "q3") + open := f.loop.windows.all() + require.Len(t, open, 2) + require.NotNil(t, open[0].pendingCR, "alice's save rides her window") + head := gitIn(t, f.worker.repoPath(), "rev-parse", "HEAD") + + ix := f.loop.footprints[itypes.NewResourceReference(echoTarget, "default")] + require.NotNil(t, ix) + for dm, ref := range ix.docLoc { + if strings.HasSuffix(ref.FilePath, "configmaps/x.yaml") { + ref.FilePath = strings.TrimSuffix(ref.FilePath, "x.yaml") + "z.yaml" + ix.docLoc[dm] = ref + } + } + + open[0].timers.maxAt = time.Now().Add(-time.Millisecond) + f.loop.closeOrArmWindow() + f.loop.endWake(0) + + assert.Equal(t, head, gitIn(t, f.worker.repoPath(), "rev-parse", "HEAD"), "nothing was committed") + assert.Empty(t, gitIn(t, f.worker.repoPath(), "status", "--porcelain"), "and the worktree was undone") + assert.True(t, f.loop.singleWindow) + assert.Equal(t, []itypes.ResourceReference{itypes.NewResourceReference(echoTarget, "default")}, asked) + res, resolved := outcome(t, f.worker) + require.True(t, resolved) + require.Error(t, res.Err) + assert.Contains(t, res.Err.Error(), echoFile("x")) +} + +// TestWindowAdmission_ACommitThatAddsOrDeletesAFileOutsideItsFootprintIsRefused: the check reads +// every staged change, a file added or deleted as much as one modified. +func TestWindowAdmission_ACommitThatAddsOrDeletesAFileOutsideItsFootprintIsRefused(t *testing.T) { + deleted := echoEvent("x", "x1") + deleted.Operation = string(itypes.OperationDelete) + created := echoEvent("w", "w1") + created.Operation = string(itypes.OperationCreate) + for name, tc := range map[string]struct { + event Event + outside string + }{ + "a file added": {event: created, outside: echoFile("w")}, + "a file deleted": {event: deleted, outside: echoFile("x")}, + } { + t.Run(name, func(t *testing.T) { + f := newAdmissionFixture(t, "admission-violation-"+strings.ReplaceAll(name, " ", "-"), nil) + pendingWrite, err := f.worker.buildGroupedPendingWrite(f.worker.ctx, []Event{tc.event}) + require.NoError(t, err) + rev, ok := f.loop.currentRevision(echoTarget, "default") + require.True(t, ok) + pendingWrite.footprint = &footprint{paths: []string{echoFile("z")}, revision: rev} + + err = f.worker.commitPendingWrites([]PendingWrite{*pendingWrite}) + var violation *footprintViolationError + require.ErrorAs(t, err, &violation) + assert.Equal(t, []string{tc.outside}, violation.outside) + }) + } +} + +// TestFootprint_UnknownReasons pins every reason a write's footprint is unknown, and so serial, as +// proveEvents says it. +func TestFootprint_UnknownReasons(t *testing.T) { + kustomization := func(body string) string { + return "apiVersion: kustomize.config.k8s.io/v1beta1\nkind: Kustomization\n" + body + } + deleted := echoEvent("gone", "g") + deleted.Operation = string(itypes.OperationDelete) + created := echoEvent("fresh", "f1") + created.Operation = string(itypes.OperationCreate) + patch := deploymentScalePatch("api", 3) + patch.GitTargetName, patch.GitTargetNamespace = echoTarget, "default" + secret := echoEvent("s", "s1") + secret.Identifier.Resource = "secrets" + secret.Object.SetKind("Secret") + + for name, tc := range map[string]struct { + files map[string]string + placement *configv1alpha3.GitTargetPlacementSpec + event Event + why string + }{ + "a field patch": {event: patch, why: "a field patch"}, + "an encrypted type": {event: secret, why: "an encrypted type"}, + "a generated object": { + files: map[string]string{echoTarget + "/kustomization.yaml": kustomization( + "configMapGenerator:\n- name: fresh\n literals:\n - a=b\n options:\n disableNameSuffixHash: true\n")}, + event: created, why: "an object a configMapGenerator renders", + }, + "useKustomize": { + placement: &configv1alpha3.GitTargetPlacementSpec{UseKustomize: true}, + event: created, why: "placement.useKustomize is on", + }, + "placement.default": { + placement: &configv1alpha3.GitTargetPlacementSpec{Default: "{namespace}/all.yaml"}, + event: created, why: "placement.default is set", + }, + "a read outside spec.path": { + files: map[string]string{ + echoTarget + "/kustomization.yaml": kustomization("resources:\n- ../shared\n"), + "shared/kustomization.yaml": kustomization("resources:\n- cm.yaml\n"), + "shared/cm.yaml": echoDocument("shared", "s"), + }, + event: created, why: "the folder reads files outside spec.path", + }, + "a folder that feeds a generator": { + files: map[string]string{ + echoTarget + "/kustomization.yaml": kustomization("configMapGenerator:\n- name: g\n files:\n - data.txt\n"), + echoTarget + "/data.txt": "data\n", + }, + event: created, why: "the folder feeds a configMapGenerator", + }, + "an unsupported kustomization": { + files: map[string]string{echoTarget + "/kustomization.yaml": kustomization( + "secretGenerator:\n- name: s\n literals:\n - a=b\n")}, + event: created, why: "the folder holds a kustomization the writer does not model", + }, + "a delete of an object no document holds": { + files: map[string]string{ + echoTarget + "/kust/kustomization.yaml": kustomization("resources:\n- k.yaml\n"), + echoTarget + "/kust/k.yaml": echoDocument("k", "k1"), + }, + event: deleted, why: "a delete of an object no document holds, in a folder with a kustomization", + }, + "a new document at a taken path": { + files: map[string]string{echoFile("fresh"): "note: not a manifest\n"}, + event: created, why: "a new document at a path that is taken", + }, + } { + t.Run(name, func(t *testing.T) { + files := map[string]string{ + echoTarget + "/README.md": "readme\n", + echoTarget + "/" + gitTargetIgnoreFileName: "\n", + } + maps.Copy(files, tc.files) + f := newEchoFixtureWith(t, "footprint-"+strings.ReplaceAll(name, " ", "-"), + configv1alpha3.PruneOnEvent, map[string]string{"x": "x1"}, files) + if tc.placement != nil { + var target configv1alpha3.GitTarget + require.NoError(t, f.worker.Client.Get(f.worker.ctx, keyOf(echoTarget), &target)) + target.Spec.Placement = tc.placement + require.NoError(t, f.worker.Client.Update(f.worker.ctx, &target)) + } + got := f.loop.proveEvents(echoTarget, "default", []Event{tc.event}) + assert.False(t, got.known()) + assert.Equal(t, tc.why, got.unknown) + }) + } +} + func TestWindowAdmission_ACommitOutsideItsFootprintIsRefusedAndTheWorkerFallsBack(t *testing.T) { f := newAdmissionFixture(t, "admission-violation", nil) var asked []itypes.ResourceReference @@ -177,6 +382,31 @@ func TestWindowAdmission_ACommitOutsideItsFootprintIsRefusedAndTheWorkerFallsBac assert.Contains(t, res.Err.Error(), echoFile("x"), "the save says which path broke the footprint") } +// TestWindowAdmission_AFallbackLeavesOneWindowOpen pins the fallback closing the windows the wrong +// rules admitted side by side: every one but the newest, oldest first, with footprint_fallback. +func TestWindowAdmission_AFallbackLeavesOneWindowOpen(t *testing.T) { + f := newAdmissionFixture(t, "admission-fallback-closes", nil) + reader := captureInfo(f.worker) + f.writeBy("alice", "x", "x2") + f.writeBy("bob", "z", "z2") + f.writeBy("carol", "q", "q3") + require.Len(t, f.loop.windows.all(), 3) + decided := len(f.loop.pendingWrites) + + refused, err := f.worker.buildGroupedPendingWrite(f.worker.ctx, []Event{echoEvent("w", "w1")}) + require.NoError(t, err) + f.loop.pendingWrites = append(f.loop.pendingWrites, *refused) + violation := &footprintViolationError{outside: []string{echoFile("w")}, footprint: footprint{}} + f.loop.settleFailed(len(f.loop.pendingWrites)-1, fmt.Errorf("%w: %w", errWriteFailed, violation)) + + require.True(t, f.loop.singleWindow) + assert.Equal(t, []string{"carol"}, authorsOf(f.loop.windows.all()), "at most one window stays open") + require.Len(t, f.loop.pendingWrites, decided+2, "alice's and bob's windows are decided") + assert.Equal(t, "alice", f.loop.pendingWrites[decided].Author(), "oldest first") + assert.Equal(t, "bob", f.loop.pendingWrites[decided+1].Author()) + assert.Len(t, reader.matching(`"msg"="Finalizing open commit window" "reason"="footprint-fallback"`), 2) +} + func TestFootprint_OverlapUnionAndOutside(t *testing.T) { a := footprint{paths: []string{"a", "b"}} b := footprint{paths: []string{"c"}} diff --git a/internal/git/window_model_test.go b/internal/git/window_model_test.go index 7c40d71f..fe5e4ed6 100644 --- a/internal/git/window_model_test.go +++ b/internal/git/window_model_test.go @@ -1301,6 +1301,7 @@ func (r *windowModelRun) checkCommit(what, hash string, want modelCommit) { author, path, f) } } + r.checkFootprintChecked(what, hash, author, f) } for path, file := range want.files { r.checkFile(what, hash, path, file) @@ -1313,6 +1314,19 @@ func (r *windowModelRun) checkCommit(what, hash string, want modelCommit) { } } +// checkFootprintChecked fails unless the worker checked the commit against its window's footprint +// itself: one that skipped it would let a footprint bug through unrefused. It is matched by tree: a +// rebuild of the prefix onto the same parent (a resync's fetch) makes the checked commit again, under +// another hash. +func (r *windowModelRun) checkFootprintChecked(what, hash, author string, f *modelFootprint) { + r.t.Helper() + tree := gitIn(r.t, r.f.repoDir, "rev-parse", hash+"^{tree}") + if len(r.logs.matching(`"tree"="`+tree+`" "footprint"=`)) == 0 { + r.failf("%s: commit %s by %s had a footprint %s, and the worker did not check it (%s)", what, + hash[:8], author, f, decisionFootprintChecked) + } +} + // checkFile compares the file at path in rev, parsed, with what the model says it holds: the // kustomization's resources: entries, or each document whole, in order. YAML 1.2, so a bare y is the // name y. diff --git a/internal/git/withheld_fields.go b/internal/git/withheld_fields.go index 779642a0..89d27eb1 100644 --- a/internal/git/withheld_fields.go +++ b/internal/git/withheld_fields.go @@ -31,18 +31,37 @@ import ( // value was carried over, and is set back to the accepted value; a field holding any other value was // changed by someone since, and leaves the record. A delete clears the object's record. // +// A delete is accepted too, as the object's absence. A withheld apply that makes the object again is +// recorded as one field, the whole object (an empty Path): every later event of the object is kept +// out of Git, and every resync snapshot of it left out, until a delete, the Repair's own included, +// clears the record. No update can be told from carrying the apply's object: an author who changed +// some field of it still changed an object only the apply made. +// +// The record also ends when nothing will put the accepted version back: the applier claim that would +// repair it is gone (dropWithheldOf), or a resync of the object's collection no longer finds it. +// // Decided limits: a list is one field, so an edit to another item of a list the apply reverted still // carries the reverted item; an author setting a field to the apply's value cannot be told from -// carrying it, and is set back; and the record lives in memory, so a restart loses it. +// carrying it, and is set back; an object made again stays out of Git until it is deleted, so an +// applier that does not prune it keeps it out for as long as the record lives; and the record lives +// in memory, so a restart loses it. // withheldFields is, by objectKey, the fields withheld applies reverted that no event has changed // since. It is plain data. -type withheldFields map[string][]withheldField +type withheldFields map[string]withheldObject + +// withheldObject is the record of one object: the GitTarget it is mirrored for and the object, which +// a resync and an applier claim's removal match it by, and its fields. +type withheldObject struct { + GitTarget itypes.ResourceReference + Identifier itypes.ResourceIdentifier + Fields []withheldField +} // withheldField is one field a withheld apply reverted. type withheldField struct { // Path is the field's map keys from the object's root. A list is one field: a path never goes - // into one. + // into one. An empty Path is the whole object, which the apply made again over a delete. Path []string // Accepted is what the object's last accepted version held at Path, Applier what the withheld // apply wrote there. @@ -62,17 +81,23 @@ func (v fieldValue) equal(other fieldValue) bool { } // keepAccepted keeps a live event's object as its object's last accepted version, after an apply -// window or a flush recorded the event as an edit. A delete, and a field patch, which carries no -// whole object, forget it: with no version to compare with, a withhold records nothing. +// window or a flush recorded the event as an edit. A delete is kept as the object's absence, a nil +// version. A field patch, which carries no whole object, forgets it: with no version to compare with, +// a withhold records nothing. func (l *branchWorkerEventLoop) keepAccepted(event Event) { key := objectKey(event.Identifier) - if event.Object == nil || event.Operation == string(itypes.OperationDelete) { + deleted := event.Operation == string(itypes.OperationDelete) + if event.Object == nil && !deleted { delete(l.accepted, key) return } if l.accepted == nil { l.accepted = map[string]*unstructured.Unstructured{} } + if deleted { + l.accepted[key] = nil + return + } l.accepted[key] = event.Object.DeepCopy() } @@ -110,86 +135,138 @@ func (l *branchWorkerEventLoop) editRecorded(key string, now time.Time) bool { } // recordWithheld records the fields a withheld apply reverted: where its object differs from the -// object's last accepted version. A field recorded already is replaced, with any recorded inside it -// or around it. With no accepted version kept there is nothing to compare with, and nothing is -// recorded. +// object's last accepted version, or, when that version is the object's absence, the whole object. A +// field recorded already is replaced, with any recorded inside it or around it. With no accepted +// version kept there is nothing to compare with, and nothing is recorded. // // The event was restored first (restoreWithheld), so a field it still carried from an earlier // withheld apply holds the accepted value, differs in nothing, and keeps its record. func (l *branchWorkerEventLoop) recordWithheld(event Event) { key := objectKey(event.Identifier) - accepted := l.accepted[key] - if accepted == nil || event.Object == nil { + accepted, kept := l.accepted[key] + if !kept || event.Object == nil { return } - reverted := diffFields(nil, accepted.Object, event.Object.Object) + var reverted []withheldField + if accepted == nil { + reverted = []withheldField{{ + Accepted: fieldValue{Absent: true}, + Applier: fieldValue{Value: copyValue(event.Object.Object)}, + }} + } else { + reverted = diffFields(nil, accepted.Object, event.Object.Object) + } if len(reverted) == 0 { return } - fields := slices.DeleteFunc(slices.Clone(l.withheld[key]), func(field withheldField) bool { + fields := slices.DeleteFunc(slices.Clone(l.withheld[key].Fields), func(field withheldField) bool { return slices.ContainsFunc(reverted, func(other withheldField) bool { return overlaps(field.Path, other.Path) }) }) if l.withheld == nil { l.withheld = withheldFields{} } - l.withheld[key] = append(fields, reverted...) + l.withheld[key] = withheldObject{ + GitTarget: itypes.NewResourceReference(event.GitTargetName, event.GitTargetNamespace), + Identifier: event.Identifier, + Fields: append(fields, reverted...), + } l.w.Log.V(1).Info("Recorded the fields a withheld apply reverted, so the next event of the object does "+ "not carry them into Git", "branch", l.w.Branch, "object", event.Identifier.String(), "fields", len(reverted)) } // restoreWithheldEvent restores a live event against the record of its object, before it is // judged: a delete clears the record, and any other event with a whole object has each field it -// carried from a withheld apply set back (restoreWithheld). -func (l *branchWorkerEventLoop) restoreWithheldEvent(event *Event) { +// carried from a withheld apply set back (restoreWithheld). It reports whether the event is kept out +// of Git: its object was deleted, and exists again only because a withheld apply made it. +func (l *branchWorkerEventLoop) restoreWithheldEvent(event *Event) bool { key := objectKey(event.Identifier) if event.Operation == string(itypes.OperationDelete) { delete(l.withheld, key) - return + return false } - if restored := l.restoreWithheld(key, event.Object); restored != event.Object { + restored := l.restoreWithheld(key, event.Object) + switch { + case restored == nil && event.Object != nil: + l.w.Log.Info("Kept an event out of Git: its object was deleted, and only a withheld apply made it again", + "branch", l.w.Branch, "object", event.Identifier.String(), "author", event.UserInfo.Username) + return true + case restored != event.Object: l.w.Log.Info("Set back the fields a withheld apply reverted, which the event carried: the cluster "+ "still runs the apply's values, and the event's author did not change them", "branch", l.w.Branch, "object", event.Identifier.String(), "author", event.UserInfo.Username) event.Object = restored } + return false } // restoreWithheldSnapshot restores each object of a resync's snapshot against its record, as a live -// event is. The snapshot itself is left as it was: an object set back is a copy, in a copy of it. +// event is, and leaves out an object a record keeps out of Git. The request's snapshot is left as it +// was: an object set back is a copy, in a copy of it. +// +// The snapshot holds every object of its scope for its GitTarget, so a recorded object it does not +// hold is gone from the cluster, and its record ends. func (l *branchWorkerEventLoop) restoreWithheldSnapshot( - desired []manifestanalyzer.DesiredResource, + req *ResyncRequest, desired []manifestanalyzer.DesiredResource, ) []manifestanalyzer.DesiredResource { if len(l.withheld) == 0 { return desired } - var restored []manifestanalyzer.DesiredResource - for i, dr := range desired { + found := make(map[string]bool, len(req.Desired)) + for _, dr := range req.Desired { + found[objectKey(dr.Resource)] = true + } + target := itypes.NewResourceReference(req.GitTargetName, req.GitTargetNamespace) + for key, record := range l.withheld { + if record.GitTarget == target && req.Scope.Matches(record.Identifier) && !found[key] { + delete(l.withheld, key) + } + } + restored := make([]manifestanalyzer.DesiredResource, 0, len(desired)) + changed := false + for _, dr := range desired { object := l.restoreWithheld(objectKey(dr.Resource), dr.Object) - if object == dr.Object { - continue + if object != dr.Object { + changed = true } - if restored == nil { - restored = slices.Clone(desired) + if object == nil && dr.Object != nil { + continue } - restored[i].Object = object + dr.Object = object + restored = append(restored, dr) } - if restored == nil { + if !changed { return desired } return restored } +// dropWithheldOf ends the records of the objects of an applier claim that is gone: no Repair of its +// will put the accepted version back. +func (l *branchWorkerEventLoop) dropWithheldOf(state *applierState) { + for key, record := range l.withheld { + if slices.ContainsFunc(state.targets, func(target branchTarget) bool { + return target.ref.Name == record.GitTarget.Name && target.ref.Namespace == record.GitTarget.Namespace + }) { + delete(l.withheld, key) + } + } +} + // restoreWithheld applies the record of an object to one of its later versions, and updates the // record: a field still holding the withheld apply's value is set back to the accepted value, and // stays recorded; a field holding any other value leaves the record. It returns the object to -// mirror, a copy when a field was set back. +// mirror, a copy when a field was set back, and nil while the record keeps the whole object out of +// Git, which no later version ends. func (l *branchWorkerEventLoop) restoreWithheld( key string, object *unstructured.Unstructured, ) *unstructured.Unstructured { - fields := l.withheld[key] + fields := l.withheld[key].Fields if len(fields) == 0 || object == nil { return object } + if slices.ContainsFunc(fields, func(field withheldField) bool { return len(field.Path) == 0 }) { + return nil + } restored := object var kept []withheldField for _, field := range fields { @@ -206,7 +283,9 @@ func (l *branchWorkerEventLoop) restoreWithheld( if len(kept) == 0 { delete(l.withheld, key) } else { - l.withheld[key] = kept + record := l.withheld[key] + record.Fields = kept + l.withheld[key] = record } return restored } diff --git a/internal/watch/event_router_outcome_test.go b/internal/watch/event_router_outcome_test.go index b95dc539..7393c3ca 100644 --- a/internal/watch/event_router_outcome_test.go +++ b/internal/watch/event_router_outcome_test.go @@ -45,6 +45,26 @@ func TestLookupCommitRequestOutcome_ReadsWhatAStoppedWorkerDecided(t *testing.T) require.NoError(t, err) require.False(t, resolved, "no worker holds the request") + // The shutdown reads it from inside the manager's stop, which holds lifecycleMu: the lookup must not + // need that lock. + var ( + atStop git.FinalizeResult + resolvedAtStop bool + lookupReturned bool + ) + workerManager.SetAfterWorkersStopped(func(ctx context.Context) { + done := make(chan struct{}) + go func() { + defer close(done) + atStop, resolvedAtStop, _ = router.LookupCommitRequestOutcome(ctx, attach) + }() + select { + case <-done: + lookupReturned = true + case <-time.After(5 * time.Second): + } + }) + ctx, cancel := context.WithCancel(context.Background()) stopped := make(chan struct{}) go func() { @@ -71,6 +91,10 @@ func TestLookupCommitRequestOutcome_ReadsWhatAStoppedWorkerDecided(t *testing.T) cancel() <-stopped + require.True(t, lookupReturned, "the lookup inside the manager's stop does not deadlock") + require.True(t, resolvedAtStop, "the shutdown's settle step reads the outcome") + assert.Equal(t, git.FinalizeNoOpenWindow, atStop.Outcome) + 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") diff --git a/test/e2e/commit_request_e2e_test.go b/test/e2e/commit_request_e2e_test.go index d69a9b31..19be2131 100644 --- a/test/e2e/commit_request_e2e_test.go +++ b/test/e2e/commit_request_e2e_test.go @@ -387,12 +387,12 @@ var _ = Describe("Save Request Bundle (UC2)", Label("commit-request", "audit-con _, _ = kubectlRunInNamespace(testNs, "delete", "commitrequest", commitRequestName, "--ignore-not-found=true") }) - // The other order, with the schema's defaults and nothing tuned: a folder whose files kubectl - // applies in name order, two ConfigMaps and then the CommitRequest that saves them. The objects' - // window is open when the request registers, so only attach: CurrentOrNext saves it. A second - // apply of the same folder saves nothing new: an unchanged object is not written again, the - // finished request does not attach again, and an edited message is refused because the spec is - // immutable. The request itself is never mirrored. + // The other order: a folder whose files kubectl applies in name order, two ConfigMaps and then + // the CommitRequest that saves them, with the request's own window left at its defaults. This + // Describe's GitTarget window is 300s, so the objects' window is certainly open when the request + // registers, and the commit can come only from the request's finalize (A1): attach: CurrentOrNext + // joins the window the objects opened. On the schema's defaults that is not certain; see + // "Save Request Folder (defaults)" below. The request itself is never mirrored. It("saves a folder applied with kubectl apply -f (objects first) in one commit with the request's message", func() { const ( @@ -463,29 +463,6 @@ spec: Expect(strings.TrimSpace(commitAuthor)).To(Equal(author), "the person who applied the folder") } - By("applying the same folder again: nothing is written, so nothing is saved") - out, err := kubectlRunInNamespace(testNs, "apply", "-f", folder) - Expect(err).NotTo(HaveOccurred(), "an unchanged folder applies cleanly") - Expect(out).To(ContainSubstring("configmap/app-config unchanged")) - Expect(out).To(ContainSubstring("commitrequest.configbutler.ai/" + requestName + " unchanged")) - - By("applying it with an edited message: the request's spec is immutable") - saveFile := filepath.Join(folder, "30-save.yaml") - content, err := os.ReadFile(saveFile) - Expect(err).NotTo(HaveOccurred()) - Expect(os.WriteFile(saveFile, - []byte(strings.Replace(string(content), message, message+", again", 1)), 0o600)).To(Succeed()) - out, err = kubectlRunInNamespace(testNs, "apply", "-f", folder) - Expect(err).To(HaveOccurred(), "an edited request is refused") - Expect(out).To(ContainSubstring("CommitRequest spec is immutable after creation")) - - Consistently(func(g Gomega) { - g.Expect(remoteBranchHead(g, repo.CheckoutDir)).To(Equal(reportedSHA), - "neither apply adds a commit") - g.Expect(commitRequestField(g, testNs, requestName, "{.status.commit}")).To(Equal(reportedSHA), - "the finished request keeps its outcome") - }, 10*time.Second, 2*time.Second).Should(Succeed()) - By("cleaning up the folder's objects") for _, name := range []string{"app-config", "app-flags"} { _, _ = kubectlRunInNamespace(testNs, "delete", "configmap", name, "--ignore-not-found=true") @@ -494,6 +471,175 @@ spec: }) }) +// The folder save as a user who tunes nothing gets it: a GitTarget with no spec.commit.window (the +// schema's 5s idleTimeout and 1m maxDuration) and a request with no spec.window (CurrentOrNext, a +// 5s attachTimeout, a 2s maxDuration from the attach). kubectl applies the two ConfigMaps first, +// then the request. +// +// A folder save is not atomic (docs/configuration.md, "Saving a folder with kubectl apply"): the +// window closes 2s after the attach, a write that reaches the worker later commits on its own under +// the generated message, and a request that registers after the objects' window closed waits for a +// next window that never comes and ends NoWindow. So this spec asserts what the defaults guarantee, +// which is weaker than the Describe above: both ConfigMaps reach Git, every commit that holds them +// is the applier's and changes only them (the request is never mirrored), and the request ends +// Committed with its message on the commit it reports, or NoWindow with its message on no commit. +// Which of the two it was, and how many commits, is logged. +var _ = Describe( + "Save Request Folder (defaults)", + Label("commit-request", "folder-defaults", "audit-consumer"), + Ordered, + func() { + var p windowedDeploymentPipeline + + BeforeAll(func() { + p = setupWindowedDeploymentPipeline("commit-request-folder", "e2e/commit-request-folder", + e2eSchemaDefaultWindow) + }) + + AfterAll(func() { + cleanupPipeline(p.ns, p.gitProvider, p.gitTarget, p.watchRule) + cleanupNamespace(p.ns) + }) + + It( + "saves a folder applied with kubectl apply -f with no timing tuned, and a second apply saves nothing", + func() { + const ( + fixtureRoot = "test/e2e/fixtures/commit-request-folder" + requestName = "save-app-config" + message = "feat: add the app config from a folder" + ruleName = "commit-request-folder-defaults-configmaps" + ) + basePath := "e2e/commit-request-folder" + configMapPaths := []string{ + fmt.Sprintf("%s/%s/configmaps/app-config.yaml", basePath, p.ns), + fmt.Sprintf("%s/%s/configmaps/app-flags.yaml", basePath, p.ns), + } + + By("watching ConfigMaps, which the folder holds") + _, err := kubectlRunWithStdin(p.ns, fmt.Sprintf(`apiVersion: configbutler.ai/v1alpha3 +kind: WatchRule +metadata: + name: %s +spec: + gitTargetRef: + name: %s + rules: + - resources: ["configmaps"] +`, ruleName, p.gitTarget), "apply", "-f", "-") + Expect(err).NotTo(HaveOccurred(), "failed to apply the ConfigMap WatchRule") + DeferCleanup(func() { cleanupWatchRule(ruleName, p.ns) }) + verifyResourceStatus("watchrule", ruleName, p.ns, "True", "Succeeded", "") + waitForStreamsRunning(p.gitTarget, p.ns) + + folder := renderFixtureFolder(fixtureRoot, map[string]string{"__E2E_GITTARGET__": p.gitTarget}) + DeferCleanup(func() { _ = os.RemoveAll(folder) }) + + By("applying the folder: the ConfigMaps first, then the CommitRequest") + _, err = kubectlRunInNamespace(p.ns, "apply", "-f", folder) + Expect(err).NotTo(HaveOccurred(), "failed to apply the folder") + + By("waiting for the request to end, Committed or NoWindow") + var outcome commitRequestOutcome + Eventually(func(g Gomega) { + outcome = readCommitRequestOutcome(g, p.ns, requestName) + g.Expect(outcome.isTerminal()).To(BeTrue(), "the request has not ended yet\n%s", outcome) + }, 2*time.Minute, 2*time.Second).Should(Succeed()) + Expect(outcome.Stalled).NotTo(Equal("True"), "the save failed\n%s", outcome) + Expect(outcome.Reason).To(BeElementOf(commitRequestReasonCommitted, "NoWindow"), "%s", outcome) + + By("waiting for both ConfigMaps to reach Git") + var saved []string + Eventually(func(g Gomega) { + pullLatestRepoState(g, p.repo.CheckoutDir) + for _, path := range configMapPaths { + _, statErr := os.Stat(filepath.Join(p.repo.CheckoutDir, path)) + g.Expect(statErr).NotTo(HaveOccurred(), "%s is not in Git yet", path) + } + saved = commitsTouching(g, p.repo.CheckoutDir, configMapPaths) + }, 2*time.Minute, 2*time.Second).Should(Succeed()) + GinkgoWriter.Printf("folder save on the defaults: request %s, the folder in %d commit(s)\n", + outcome.Reason, len(saved)) + + By("checking every commit that holds the folder") + author := "" + if !configuredAuthorModeEnabled() { + author = commitRequestField(Default, p.ns, requestName, "{.status.author}") + } + for _, sha := range saved { + paths, err := gitRun( + p.repo.CheckoutDir, + "diff-tree", + "--root", + "-r", + "--name-only", + "--no-commit-id", + sha, + ) + Expect(err).NotTo(HaveOccurred()) + // The folder's first commit also adds the GitTarget's bootstrap files, beside the namespace. + var objects []string + for _, path := range strings.Fields(paths) { + if strings.HasPrefix(path, basePath+"/"+p.ns+"/") { + objects = append(objects, path) + } + } + Expect(objects).To(HaveEach(BeElementOf(configMapPaths)), + "a commit of the folder changes its ConfigMaps and no other object: the request is never mirrored") + subject, err := gitRun(p.repo.CheckoutDir, "log", "-1", "--pretty=%B", sha) + Expect(err).NotTo(HaveOccurred()) + if outcome.Reason == commitRequestReasonCommitted && sha == outcome.Commit { + Expect(strings.TrimSpace(subject)).To(Equal(message), "the commit the request reports") + } else { + Expect(strings.TrimSpace(subject)).NotTo(Equal(message), "only the request's commit carries it") + } + if author != "" { + commitAuthor, err := gitRun(p.repo.CheckoutDir, "log", "-1", "--pretty=%an", sha) + Expect(err).NotTo(HaveOccurred()) + Expect(strings.TrimSpace(commitAuthor)).To(Equal(author), "the person who applied the folder") + } + } + if outcome.Reason == commitRequestReasonCommitted { + Expect(saved).To(ContainElement(outcome.Commit), "the request reports a commit of the folder") + } + + head := "" + Eventually(func(g Gomega) { head = remoteBranchHead(g, p.repo.CheckoutDir) }).Should(Succeed()) + + By("applying the same folder again: nothing is written, so nothing is saved") + out, err := kubectlRunInNamespace(p.ns, "apply", "-f", folder) + Expect(err).NotTo(HaveOccurred(), "an unchanged folder applies cleanly") + Expect(out).To(ContainSubstring("configmap/app-config unchanged")) + Expect(out).To(ContainSubstring("commitrequest.configbutler.ai/" + requestName + " unchanged")) + + By("applying it with an edited message: the request's spec is immutable") + saveFile := filepath.Join(folder, "30-save.yaml") + content, err := os.ReadFile(saveFile) + Expect(err).NotTo(HaveOccurred()) + Expect(os.WriteFile(saveFile, + []byte(strings.Replace(string(content), message, message+", again", 1)), 0o600)).To(Succeed()) + out, err = kubectlRunInNamespace(p.ns, "apply", "-f", folder) + Expect(err).To(HaveOccurred(), "an edited request is refused") + Expect(out).To(ContainSubstring("CommitRequest spec is immutable after creation")) + + Consistently(func(g Gomega) { + g.Expect(remoteBranchHead(g, p.repo.CheckoutDir)).To(Equal(head), "neither apply adds a commit") + g.Expect(readCommitRequestOutcome(g, p.ns, requestName)).To(Equal(outcome), + "the finished request keeps its outcome") + }, 10*time.Second, 2*time.Second).Should(Succeed()) + }, + ) + }, +) + +// commitsTouching lists the commits on the checkout's HEAD, oldest first, that change any of paths. +func commitsTouching(g Gomega, checkoutDir string, paths []string) []string { + args := append([]string{"log", "--reverse", "--format=%H", "--"}, paths...) + out, err := gitRun(checkoutDir, args...) + g.Expect(err).NotTo(HaveOccurred()) + return strings.Fields(out) +} + // commitRequestManifest renders a single CommitRequest document with an explicit message and a // window that waits and then collects for the given Go duration each. It is used to build // multi-document `kubectl apply` bundles where the CommitRequest is the first document. @@ -697,8 +843,8 @@ func expectCommitRequestCommitted(g Gomega, namespace, name, diagnostics string) } // windowedDeploymentPipeline is the pipeline a CommitRequest suite saves through: a namespace and -// Gitea repo of its own, a GitProvider, a GitTarget with a long spec.commit.window, and a WatchRule -// on Deployments. +// Gitea repo of its own, a GitProvider, a GitTarget with the given spec.commit.window (none at all for +// e2eSchemaDefaultWindow), and a WatchRule on Deployments. type windowedDeploymentPipeline struct { ns string repo *RepoArtifacts diff --git a/test/e2e/helpers.go b/test/e2e/helpers.go index 064617f1..16bc98d1 100644 --- a/test/e2e/helpers.go +++ b/test/e2e/helpers.go @@ -197,6 +197,10 @@ const e2eCommitWindow = "0s" // cap would be exactly such a timer. const e2eCommitMaxDuration = "24h" +// e2eSchemaDefaultWindow asks for a GitTarget with no spec.commit.window at all, so the schema's +// defaults apply: what a user who tunes nothing gets. +const e2eSchemaDefaultWindow = "" + // gitBranchName is the GitBranch the e2e helpers create for a GitProvider's branch: one per // provider and branch, in the provider's namespace, so every GitTarget the helpers create on that // branch shares it. diff --git a/test/e2e/templates/gittarget.tmpl b/test/e2e/templates/gittarget.tmpl index b96a02df..f10ed9e3 100644 --- a/test/e2e/templates/gittarget.tmpl +++ b/test/e2e/templates/gittarget.tmpl @@ -25,10 +25,14 @@ spec: name: {{ .GitBranchName }} {{ .KubeProviderRef }} path: {{ .Path }} +{{- if or .CommitWindow .LiveTemplate .ReconcileTemplate }} commit: +{{- end }} +{{- if .CommitWindow }} window: idleTimeout: "{{ .CommitWindow }}" maxDuration: "{{ .CommitMaxDuration }}" +{{- end }} {{- if or .LiveTemplate .ReconcileTemplate }} message: {{- if .LiveTemplate }}