diff --git a/docs/TODO.md b/docs/TODO.md index 1a69c5ab..781f0e37 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -25,8 +25,12 @@ should land before the next tag is cut (the last release is v0.55.0). Checked ag `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). + - measure the footprint index on realistic repositories and commit rates before widening the + layouts it proves. It is rebuilt whenever its revision changes, every local commit included; + a review measured a rebuild at about 3.5 ms and 3.9 MB allocated for 100 small manifests, and + 33 ms and 37 MB for 1,000 (allocation, not retained heap), against 19 µs for a reuse. The + retained-byte budget does not charge the indexes, the accepted versions or the withheld-field + records, so a full budget does not bound the worker's memory. ## Applier claims and bi-directional editing @@ -46,7 +50,7 @@ should land before the next tag is cut (the last release is v0.55.0). Checked ag [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 + 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 diff --git a/docs/architecture.md b/docs/architecture.md index daba457e..51a52811 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -226,8 +226,10 @@ The solution, in the vocabulary used throughout this document: `unknown (attribution unresolved)`, so an installation that expects attribution can investigate its audit policy, ingress, Redis connection, or source identity instead of mistaking the result for configured-author mode. - **One `BranchWorker` per Git branch serializes all writes.** Every write to a branch funnels through a - single worker and a single commit window, which keeps concurrent GitTargets and authors from racing - each other into a corrupt tree. + single worker and its one event loop, which owns the checkout and orders every Git operation, so + concurrent GitTargets and authors cannot race each other into a corrupt tree. The loop may hold + several commit windows, one per author, only while it proves their files do not overlap + ([concurrent-author-windows.md](design/concurrent-author-windows.md)). *** @@ -756,9 +758,10 @@ flowchart TD PUSH --> DONE[Remote branch updated] ``` -The important boundary is the open window: it accepts only the same author and `GitTarget`. A -`CommitRequest` from another author never closes someone else's window; it resolves as a benign no-commit -instead. +The important boundary is the window: each accepts only the same author and `GitTarget`. Two +authors keep a window each, side by side, while their files are proven not to overlap; otherwise the +newcomer's write closes the other window first. A `CommitRequest` attaches only to its own author's +window and never closes someone else's; with none to attach to, it resolves as a benign no-commit. ### Remote moved while we were writing diff --git a/docs/configuration.md b/docs/configuration.md index 950002b5..31ff9f24 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -3311,8 +3311,9 @@ request produced a pushed commit: request is refused at once instead. Save again with a new request once the earlier one is `Ready`. A request for the `Next` window is not refused by one that is collecting: it closes that window under the earlier request's message first. Saves by different authors do not race for a - message: a write by another author closes the open window, and its message, before it opens one - of its own. + message: each author's writes collect in a window of their own, side by side while their files + are proven not to overlap, and otherwise a write by another author closes the open window, and + its message, before it opens one of its own. `SaveInProgress` is decided when the branch worker registers the request, which is after its author was found, not when it is created. A client that relies on it to keep two of its saves @@ -3339,8 +3340,10 @@ request produced a pushed commit: A request is served again after GitOps Reverser restarts: the path that owns it finds its own claim, and resumes with the author it recorded. A graceful stop, such as an upgrade or a rollout, first finishes the saves it holds: a request collecting its window is committed with its message, -pushed, and its outcome written before the pod exits. A request still waiting for a window is -picked up again by the next pod. A crash, or a stop that runs out of its grace period, loses the +pushed, and its outcome written before the pod exits. Edits the branch worker had already accepted +are committed too, each in its author's window; an edit still on its way from the cluster is not, +and the next pod's startup replay writes it under a generated message. A request still waiting for +a window is picked up again by the next pod. A crash, or a stop that runs out of its grace period, loses the open window: the next pod's startup replay writes its edits under a generated message, and the request attaches again as if new, so it can end `NoWindow`. diff --git a/docs/design/concurrent-author-windows.md b/docs/design/concurrent-author-windows.md index 07cebe7f..d94f92c6 100644 --- a/docs/design/concurrent-author-windows.md +++ b/docs/design/concurrent-author-windows.md @@ -18,7 +18,7 @@ Neither is needed for the benefit this page delivers. ## The problem -A branch worker holds **one** open commit window at a time (`branchWorkerEventLoop.openWindow`). A +Before stage 2, a branch worker held **one** open commit window at a time (`branchWorkerEventLoop.openWindow`). A window belongs to one author on one `GitTarget`, and a write that is anyone else's cannot join it (`openWindow.canAppend`). It closes the window first, with close reason `identity_change`, and opens its own. @@ -51,6 +51,14 @@ gitops-api avoids it today by serializing every save and its edits behind one lo (`commitGate`). A client that creates `CommitRequest`s on a source cluster from the browser has no such lock, so the split comes back. +What this page delivers is **commit grouping**: when different authors edit files it can prove +independent, each gets a commit of their own. It is not a transaction. A save still collects the +writes that reach its author's window before its timers close it; events and the save arrive +independently, so a deadline can still split a save, and nothing proves a save holds every edit its +author meant. A footprint proves that two windows change different files, not that two objects in +those files are consistent with each other. "This save is exactly these edits" needs explicit +membership and a completion signal, which longer timers cannot give. + Two saves by **one** author on one `GitTarget` are a different problem, and it is already handled. A window carries one message, so the second save is refused with `SaveInProgress` while the first still waits for a window or collects one. That stays true under this proposal, which changes only @@ -128,7 +136,8 @@ A write is **proven independent** only when all of these hold: is off, so the write registers nothing in a `kustomization.yaml` and creates none (a new file is always added to the kustomization governing it, and `useKustomize` creates one on the first new document); -- the object's destination is the built-in path, with no `placement.byType` or `placement.default` +- an object with a document already is written where that document lives, at any path; a new + object's destination is the built-in path, with no `placement.byType` or `placement.default` template for its type: a template can bundle several objects into one file (`all.yaml`); - the destination file holds only this object's document, or does not exist yet. The built-in path has no API version, so two versions of one object share a file: they count as one object. @@ -286,8 +295,8 @@ window, so they are proven against today's behavior before they are used to judg ## Stages -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. +Stages 1 and 2 are built; stage 3 is in progress, stage 4 was decided against, and stage 5 is +proposed. Stage 3 gates the next release, because stage 2 shipped without a 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 @@ -300,10 +309,14 @@ stage 2 shipped without stage 4's flag. 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. +4. **Decided against (2026-10-28): no startup flag.** Stage 2 is always on, so stage 3's tests gate + the release instead of a default-off rollout. What the flag's measurement period was for is read + from the metrics: `git_commit_windows_open` shows windows open side by side, + `git_commit_windows_single` a worker that fell back to one at a time, and + `git_commit_windows_total` by `close_reason` the forced splits (`identity_change`, + `path_conflict`, `footprint_fallback`). Memory is not measured yet: the retained-byte budget + charges windows and pending writes, but not the footprint indexes, the accepted versions or the + withheld-field records. 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 diff --git a/docs/design/event-pipeline-overview.md b/docs/design/event-pipeline-overview.md index 501a9719..ab354ceb 100644 --- a/docs/design/event-pipeline-overview.md +++ b/docs/design/event-pipeline-overview.md @@ -275,9 +275,11 @@ admission check reads. flowchart LR E["live event"] --> G{"render fidelity
established for
this target?"} G -- "no" --> DROP["dropped
(the pending replay repairs it)"] - G -- "open" --> CAN{"same author
and target as
the open window?"} - CAN -- "no" --> FIN1["close window
(identity change)"] --> NEW + G -- "open" --> CAN{"an open window of
the same author
and target?"} CAN -- "yes" --> ADD["add to window"] + CAN -- "no" --> PROVE{"footprints proven
and disjoint?"} + PROVE -- "yes" --> NEW + PROVE -- "no" --> FIN1["close the windows
it cannot sit beside
(identity change,
path conflict)"] --> NEW NEW["open window,
snapshot target's timers"] --> ATT["attach a waiting save"] --> ADD ADD --> CAP{"retained bytes
over budget?"} CAP -- "yes" --> FIN2["close window
(buffer limit)"] diff --git a/docs/interpreting-metrics.md b/docs/interpreting-metrics.md index aa4e2a5b..9d449866 100644 --- a/docs/interpreting-metrics.md +++ b/docs/interpreting-metrics.md @@ -208,6 +208,8 @@ boundary, the commit, the push. Background: | `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` / `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_commit_windows_open` | gauge | `gitbranch_namespace`, `gitbranch_name`, `provider_namespace`, `provider_name` | Commit windows the branch worker holds open. Above 1 means authors are collecting side by side, one window each, because their files are proven not to overlap. A branch that only ever reads 0 or 1 while several people edit it gets no concurrency: their writes are in shared files, kustomize folders, or another layout the worker cannot prove independent. Read at scrape time. | +| `git_commit_windows_single` | gauge | `gitbranch_namespace`, `gitbranch_name`, `provider_namespace`, `provider_name` | 1 once the branch worker has fallen back to one commit window at a time, after a commit would have changed a file outside its window's footprint (a bug: the worker refused the commit and asked for a resync). It stays 1 until the worker restarts. The worker logs the refused paths at Error (`SingleWindowFallback`). Read at scrape time. | | `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. | | `git_retained_writes` | gauge | `gitbranch_namespace`, `gitbranch_name`, `provider_namespace`, `provider_name` | Decided writes waiting for a push. | @@ -1421,6 +1423,7 @@ sum by (outcome) (rate(gitopsreverser_secret_encryptions_total[5m])) | `sum by (gitbranch_namespace, gitbranch_name, reason) (increase(gitopsreverser_git_writes_dropped_total[15m])) > 0` | Loss: writes met a closed render-vs-live gate and were dropped, and nothing replays them. `render_fidelity_diverged` goes with `RenderMatchesLive=False` on a `GitTarget`; `render_fidelity_pending` is a hold that gave up. | | `sum by (gittarget_namespace, gittarget_name, reason) (increase(gitopsreverser_placement_refusals_total[1h])) > 0` | Loss: new resources the writer declined to place are absent from the mirror until the placement policy is fixed. | | `sum by (gittarget_namespace, gittarget_name) (increase(gitopsreverser_placement_kustomization_entries_total{outcome="failed"}[1h])) > 0` | A new file is in Git without its `resources:` entry, so kustomize never builds it. | +| `max by (gitbranch_namespace, gitbranch_name) (gitopsreverser_git_commit_windows_single) == 1` | A branch worker found a commit that would have changed a file outside its window's footprint, refused it, and holds one commit window at a time until it restarts. It is a bug to report, with the `SingleWindowFallback` log line. | | `max by (gitbranch_namespace, gitbranch_name) (gitopsreverser_apply_operations_unresolved) > 0`, `for: 15m` | An applier is stuck mid-apply, and nothing more is asked of it until the operation completes. | | `max by (gitbranch_namespace, gitbranch_name) (gitopsreverser_apply_appliers_held) > 0`, `for: 15m` | An applier is held behind the initial-snapshot barrier: a `GitTarget` naming its claim watches nothing, or its folder is missing. | | `sum by (applier_kind, claim_namespace, claim_name, reason) (rate(gitopsreverser_apply_requests_total{outcome="failed"}[15m])) > 0`, `for: 30m` | Apply requests keep failing for the named applier claim; the worker retries at its backoff, and the cluster runs older content than the branch meanwhile. | diff --git a/docs/spec/commit-window-refactor.md b/docs/spec/commit-window-refactor.md index a7bf0ae9..f612f638 100644 --- a/docs/spec/commit-window-refactor.md +++ b/docs/spec/commit-window-refactor.md @@ -65,9 +65,9 @@ than planning the cluster's picture again onto the new tip. ```mermaid flowchart TD A[Live watch event or resync] --> B{Kind} - B -->|live event| C[Open window: same author and target] + B -->|live event| C[The author's window, or a new one beside proven-disjoint others] C --> E{Finalize trigger} - E -->|author or target change| F[PendingWriteCommit] + E -->|target change, or a write it cannot sit beside| F[PendingWriteCommit] E -->|idleTimeout silence| F E -->|maxDuration| F E -->|byte cap| F @@ -75,7 +75,7 @@ flowchart TD E -->|apply due: flush| F E -->|Git host push notice| F E -->|shutdown| F - B -->|resync| Q[Finalize open window if present] + B -->|resync| Q[Finalize the open windows, oldest first] Q -.-> F Q -->|then| D[PendingWriteResync] F --> G[pendingWrites] @@ -90,18 +90,23 @@ flowchart TD J -->|transient failure| N[keep local commits and pendingWrites for retry] ``` -`BranchWorker` owns the live open window, timers, repository preparation, push +`BranchWorker` owns the live open windows, timers, repository preparation, push cooldowns, and retry scheduling. It retains `[]PendingWrite` until a push succeeds. Local commit creation does not clear pending work. A write whose first commit changes nothing in Git is the exception: it settles at once and is neither retained nor pushed. A save (a write carrying a `CommitRequest`) is always retained, and the push decides its outcome. -Per-event writes are processed as a stream. The branch worker keeps one open -window at a time, and that window contains only one author and one target. The -window finalizes on its idle timeout or its maximum duration, and immediately on -author change, target change, the byte cap, an `attach: Next` request, a resync, -shutdown, an apply coming due, or a Git host's push notice. An apply coming due +Per-event writes are processed as a stream. Each open window contains only one +author and one target. The branch worker keeps several open side by side, one per +author of a target, while it can prove their footprints (the files their commits +will change) do not overlap; anything it cannot prove stays serial +([concurrent-author-windows.md](../design/concurrent-author-windows.md)). A +window finalizes on its idle timeout or its maximum duration, and immediately when +a write cannot sit beside it (another target, an unproven write, a write to a file +it holds), on the byte cap, an `attach: Next` request, a resync, shutdown, an +apply coming due, or a Git host's push notice. Windows that one trigger closes +close oldest first. An apply coming due closes the window so its edits are published before the apply request is sent ([the flush](../design/apply-trigger.md#flush-then-apply), `close_reason="apply_due"`). A push notice closes it so its writes are @@ -159,9 +164,13 @@ It appends to or finalizes the open window on event arrival. Rules: -- A new window starts when there is no open window. -- The current window finalizes when the author changes. -- The current window finalizes when the `GitTarget` changes. +- A write joins the open window of its author and `GitTarget`. +- Otherwise it opens a window of its own. It may open beside the windows already + open only when each footprint, its own included, is proven and none overlaps. + Every window it cannot sit beside finalizes first, oldest first. +- A write of another `GitTarget` finalizes every open window. +- A worker that refused a commit outside its window's footprint holds one window + at a time until it restarts. - Same-author, same-target edits of the same path stay in the same window. - Within a window, the latest event for a path wins. - First-seen path order is preserved for commit message resource ordering. diff --git a/internal/git/apply_window_carry_test.go b/internal/git/apply_window_carry_test.go index 94aa364f..774c7757 100644 --- a/internal/git/apply_window_carry_test.go +++ b/internal/git/apply_window_carry_test.go @@ -78,8 +78,8 @@ func carryWithheld(t *testing.T, slug string) *echoFixture { return f } -// carryKey is y's objectKey. -func carryKey() string { return objectKey(carryEdit("alice", "", "").Identifier) } +// carryKey is y's key in the record, as team-a mirrors it. +func carryKey() withheldKey { return withheldKeyOf(carryEdit("alice", "", "")) } // TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit pins the withheld-field record // (withheld_fields.go). @@ -328,7 +328,9 @@ func TestWithheldFields_RecordAndRestore(t *testing.T) { 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}) + restoredEvent := accepted + restoredEvent.Object = restored + loop.keepAccepted(restoredEvent) again := carryApply("", "") delete(again.Object.Object, "data") again.Object.Object["spec"] = map[string]interface{}{"list": []interface{}{"one"}} @@ -455,6 +457,37 @@ func TestApplyWindow_TheRecordEndsWhenNothingWillRepairIt(t *testing.T) { }) } +// TestApplyWindow_AWithheldApplyStaysWithItsGitTarget pins the record to the GitTarget the withheld +// apply was mirrored for. Another GitTarget on the branch can mirror an object of the same name, from +// another cluster or into another folder: its writes are neither restored against the record nor +// kept out of Git by it. +func TestApplyWindow_AWithheldApplyStaysWithItsGitTarget(t *testing.T) { + teamB := func(f *echoFixture, a, b string) Event { + createGitTargetWithPruneMode(t, f.worker, "team-b", "team-b", configv1alpha3.PruneOnEvent) + event := carryEdit("bob", a, b) + event.GitTargetName = "team-b" + return event + } + + t.Run("a reverted field", func(t *testing.T) { + f := carryWithheld(t, "carry-other-target-field") + f.writeWindow(teamB(f, "a1", "b2")) + f.publishAny() + assert.Contains(t, f.onMain("team-b/default/configmaps/y.yaml"), "a: a1", + "team-a's withheld apply does not set team-b's field back") + assert.Len(t, f.loop.withheld[carryKey()].Fields, 1, "team-a's record stays") + }) + + t.Run("an object made again", func(t *testing.T) { + f := carryDeleteThenRecreate(t, "carry-other-target-object") + f.writeWindow(teamB(f, "a9", "b9")) + f.publishAny() + assert.Contains(t, f.onMain("team-b/default/configmaps/y.yaml"), "a: a9", + "team-a's withheld apply does not keep team-b's object out of Git") + assert.Empty(t, f.onMain(echoFile("y")), "team-a's object stays out") + }) +} + // 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) { diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index d7d175f6..55dbeae2 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -159,6 +159,11 @@ type BranchWorker struct { // heldAppliers is how many of this branch's appliers the barrier holds a reason back for, as the // loop last published it, for the apply_appliers_held gauge (barrier.go). heldAppliers atomic.Int64 + // openWindows is how many commit windows the loop holds open, and singleWindow whether it holds + // one at a time since a footprint fallback, as the loop last published them, for the + // git_commit_windows_open and git_commit_windows_single gauges. + openWindows atomic.Int64 + singleWindow atomic.Bool // applyEpoch, appliersAhead and hasAppliers are the loop's ApplyState for the watch side, which // reads it when a snapshot's collection starts (apply_epoch.go). applyEpoch atomic.Uint64 @@ -199,17 +204,18 @@ type BranchWorker struct { // stoppingState closes the queue to new work from the moment Stop is called. // - // Shutdown drains whatever is still on the FIFO without processing it, so an item accepted - // after that point is silently thrown away, while Enqueue's "true" has already told a watch - // loop it may advance its durable cursor past the event. That cost nothing while Stop was + // Shutdown decides the live writes still on the FIFO and drains every other item without + // processing it (decideQueuedWrites), so an item accepted after that point is silently thrown + // away, while Enqueue's "true" has already told a watch loop it may advance its durable cursor + // past the event. That cost nothing while Stop was // only ever called at process exit. A worker REPLACED because its GitProvider names a // different repository is stopped with the process still running, and the sibling GitTargets // on its branch go on holding streams that point at it until their own next reconcile, which // is minutes of accepted-and-discarded events. // // Refusing them turns a silent loss into the drop path every producer already handles: the - // watch leaves its cursor where it is and redelivers to the replacement. Items already ON the - // queue when Stop is called are still drained, which is the behaviour shutdown always had. + // watch leaves its cursor where it is and redelivers to the replacement. Writes already ON the + // queue when Stop is called are still decided and pushed, and other items drained. // // It is written and read under pendingResyncsMu, the lock every enqueue already holds across // its send. Checking it outside that lock is not enough: an enqueue could pass the check, @@ -1152,10 +1158,10 @@ type branchWorkerEventLoop struct { flushing bool flushMark uint64 flushEdits map[itypes.ResourceReference]map[string]time.Time - // accepted is, by objectKey, the object of the last event an open apply window or the flush + // accepted is, by withheldKey, the object of the last event an open apply window or the flush // recorded as an edit, and withheld the fields withheld applies reverted that no event has // changed since: the next event of the object is restored against them (withheld_fields.go). - accepted map[string]*unstructured.Unstructured + accepted map[withheldKey]*unstructured.Unstructured withheld withheldFields // attachTimer fires at the earliest attach deadline of a waiting CommitRequest, so a request @@ -1289,6 +1295,8 @@ func (l *branchWorkerEventLoop) endWake(released int64) { // there would report a drained queue for a worker that has work. func (l *branchWorkerEventLoop) publishLoopState(released int64) { l.w.hasUnpushedWork.Store(l.windows.anyOpen() || len(l.pendingWrites) > 0 || len(l.held) > 0) + l.w.openWindows.Store(int64(l.windows.count())) + l.w.singleWindow.Store(l.singleWindow) l.syncAdmission(released) } @@ -1506,6 +1514,7 @@ func (l *branchWorkerEventLoop) armCommitTimer() { func (l *branchWorkerEventLoop) handleShutdown() { l.w.Log.Info("Handling shutdown, finalizing open window and pushing pending commits") + l.decideQueuedWrites() l.finalizeOpenWindowWithReason(windowFinalizeReasonShutdown) l.dropHeldAtShutdown() if len(l.pendingWrites) > 0 { @@ -1517,6 +1526,30 @@ func (l *branchWorkerEventLoop) handleShutdown() { l.drainUnhandledQueueItems() } +// decideQueuedWrites decides the live writes already on the FIFO when the shutdown began, before the +// windows close, so the last push carries them. Enqueue answered "accepted" for each, and a watch may +// have moved its cursor past it: dropped, the edit would come back only through a later resync, with +// no author and no window of its own. It takes only what the FIFO held when the shutdown began, so a +// producer still sending (the manager's cancel stops the worker before Stop closes admission) cannot +// keep it going. Writes are decided locally; every other item, which may read the remote or bind a +// save, is left to the drain, as before. +func (l *branchWorkerEventLoop) decideQueuedWrites() { + for range len(l.w.eventQueue) { + var item WorkItem + select { + case item = <-l.w.eventQueue: + default: + return + } + if item.Request == nil { + l.w.dropQueueItem(item) + continue + } + l.publishLoopState(l.dispatch(item)) + l.w.releaseInflightItem() + } +} + // failUnpushedCommitRequests settles every CommitRequest still riding a write the exiting loop // could not push. // @@ -1548,21 +1581,26 @@ func (w *BranchWorker) drainQueue() { for { select { case item := <-w.eventQueue: - if item.Attach != nil { - w.uncountQueuedAttach(item.Attach.id()) - } - charge := item.charge - if item.Resync != nil { - _, charge = w.takePendingResync(item.Resync) - } - w.intake.release(charge) - w.releaseInflightItem() + w.dropQueueItem(item) default: return } } } +// dropQueueItem clears one item no loop will handle: see drainQueue. +func (w *BranchWorker) dropQueueItem(item WorkItem) { + if item.Attach != nil { + w.uncountQueuedAttach(item.Attach.id()) + } + charge := item.charge + if item.Resync != nil { + _, charge = w.takePendingResync(item.Resync) + } + w.intake.release(charge) + w.releaseInflightItem() +} + func (l *branchWorkerEventLoop) resetCommitTimer(delay time.Duration) { l.commitTimerAt = l.w.now().Add(delay) if l.commitTimer == nil { diff --git a/internal/git/branch_worker_metrics_test.go b/internal/git/branch_worker_metrics_test.go index 02313b25..0a4ce40a 100644 --- a/internal/git/branch_worker_metrics_test.go +++ b/internal/git/branch_worker_metrics_test.go @@ -226,12 +226,13 @@ func TestEnqueueRequest_FullQueueCountsTheDrop(t *testing.T) { // On shutdown the loop must drain items still buffered on eventQueue so the depth reads 0. Each // buffered item was counted into inflightItems at enqueue; left undrained, the exiting worker would -// report a non-zero depth that never clears. +// report a non-zero depth that never clears. Queued writes are decided at shutdown instead +// (TestWindowAdmission_ShutdownDecidesTheWritesAlreadyQueued). func TestHandleShutdown_DrainsBufferedItemsToZeroDepth(t *testing.T) { w := newMetricsTestWorker() - // Two write requests accepted onto the queue but never handled by the loop. - w.enqueueRequest(&WriteRequest{Events: []Event{{}}}) - w.enqueueRequest(&WriteRequest{Events: []Event{{}}}) + // Two refreshes accepted onto the queue but never handled by the loop. + require.True(t, w.EnqueueRefresh(&RefreshRequest{})) + require.True(t, w.EnqueueRefresh(&RefreshRequest{})) require.Equal(t, int64(2), w.inflightItems.Load()) loop := newBranchWorkerEventLoop(w, time.Second) diff --git a/internal/git/footprint.go b/internal/git/footprint.go index 08b240a5..fd662518 100644 --- a/internal/git/footprint.go +++ b/internal/git/footprint.go @@ -496,22 +496,27 @@ func (w *BranchWorker) checkFootprint( 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. +// logFootprintChecked logs the commit checkFootprint checked, once it is made. The tree and its +// parent's tree name the change that 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. +// The tree alone is not enough: a later commit can return to the same tree from another parent, by +// a change nobody checked. func (w *BranchWorker) logFootprintChecked( repo *gogit.Repository, hash plumbing.Hash, pendingWrite *PendingWrite, checked bool, ) { if !checked { return } - var tree plumbing.Hash + var tree, parentTree plumbing.Hash if commit, err := repo.CommitObject(hash); err == nil { tree = commit.TreeHash + if parent, err := commit.Parent(0); err == nil { + parentTree = parent.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) + "commit", hash.String(), "tree", tree.String(), "parentTree", parentTree.String(), + "footprint", pendingWrite.footprint.String(), "decision", decisionFootprintChecked) } // provenOnCheckout reports whether rev is the revision the checkout is at now. diff --git a/internal/git/publication.go b/internal/git/publication.go index 19053f79..1ec5749c 100644 --- a/internal/git/publication.go +++ b/internal/git/publication.go @@ -157,6 +157,15 @@ func (m *WorkerManager) setPublicationGaugeSources() { telemetry.SetGaugeSource(telemetry.GaugeAppliersHeld, source(func(w *BranchWorker) (int64, bool) { return always(w.heldAppliers.Load()) })) + telemetry.SetGaugeSource(telemetry.GaugeGitCommitWindowsOpen, source(func(w *BranchWorker) (int64, bool) { + return always(w.openWindows.Load()) + })) + telemetry.SetGaugeSource(telemetry.GaugeGitCommitWindowsSingle, source(func(w *BranchWorker) (int64, bool) { + if w.singleWindow.Load() { + return 1, true + } + return 0, true + })) } // publicationGauges are the gauges setPublicationGaugeSources installs. @@ -170,6 +179,8 @@ var publicationGauges = []string{ telemetry.GaugeGitNextRetry, telemetry.GaugeApplyOperationsUnresolved, telemetry.GaugeAppliersHeld, + telemetry.GaugeGitCommitWindowsOpen, + telemetry.GaugeGitCommitWindowsSingle, } // publicationEventBuffer bounds the GitTarget events a burst of report changes can queue. A full diff --git a/internal/git/window_admission_test.go b/internal/git/window_admission_test.go index 975ca258..74481670 100644 --- a/internal/git/window_admission_test.go +++ b/internal/git/window_admission_test.go @@ -9,10 +9,12 @@ import ( "testing" "time" + "github.com/go-logr/logr" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + "github.com/ConfigButler/gitops-reverser/internal/telemetry" itypes "github.com/ConfigButler/gitops-reverser/internal/types" ) @@ -145,6 +147,84 @@ func TestWindowAdmission_ClosingOneWindowArmsTheTimerForTheOthers(t *testing.T) assert.Equal(t, bobAt, f.loop.commitTimerAt) } +// TestWindowAdmission_ADocumentAtItsOwnPathIsProvenWhereItLives pins an existing object outside the +// built-in path: alone in its file, that file is its footprint, and its author keeps a window beside +// another's. +func TestWindowAdmission_ADocumentAtItsOwnPathIsProvenWhereItLives(t *testing.T) { + custom := echoTarget + "/custom/named.yaml" + f := newAdmissionFixture(t, "admission-custom-path", map[string]string{custom: echoDocument("named", "v1")}) + f.writeBy("alice", "named", "v2") + f.writeBy("bob", "z", "z2") + + require.Equal(t, []string{"alice", "bob"}, authorsOf(f.loop.windows.all())) + assert.Equal(t, []string{custom}, f.loop.windows.all()[0].footprint.paths) + f.publishAny() + assert.Equal(t, echoDocument("named", "v2"), f.onMain(custom)) + assert.Equal(t, echoDocument("z", "z2"), f.onMain(echoFile("z"))) +} + +// TestWindowAdmission_ThePushCooldownKeepsWindowsSideBySide pins the commit of a decided write ahead of +// its push: while the cooldown holds the push, the checkout already holds the write, so a later write +// is still proven and opens beside the windows left open. +func TestWindowAdmission_ThePushCooldownKeepsWindowsSideBySide(t *testing.T) { + f := newAdmissionFixture(t, "admission-cooldown", nil) + f.loop.cooldownHoldsSaves = false + f.loop.pushPending() + f.loop.lastPushAt = time.Now() + f.writeBy("alice", "x", "x2") + f.writeBy("bob", "z", "z2") + f.loop.finalizeWindow(f.loop.windows.all()[0], windowFinalizeReasonIdleTimeout) + f.loop.endWake(0) + require.Len(t, f.loop.pendingWrites, 1) + require.NotNil(t, f.loop.pushTimer, "the cooldown holds alice's push") + require.True(t, f.loop.checkoutCurrent(), "and her write is committed already") + + f.writeBy("carol", "q", "q3") + f.writeBy("bob", "z", "z3") + assert.Equal(t, []string{"bob", "carol"}, authorsOf(f.loop.windows.all())) + + f.loop.finalizeOpenWindowWithReason(windowFinalizeReasonShutdown) + f.loop.endWake(0) + f.loop.pushPending() + assert.Equal(t, echoDocument("x", "x2"), f.onMain(echoFile("x"))) + assert.Equal(t, echoDocument("z", "z3"), f.onMain(echoFile("z"))) + assert.Equal(t, echoDocument("q", "q3"), f.onMain(echoFile("q"))) +} + +// TestWindowAdmission_AFailedPushKeepsItsWritesBesideTheOpenWindows pins an outage with windows open: +// the write the failed push carried stays committed and pending, the other windows go on collecting, +// and the retry publishes all of it without another edit. +func TestWindowAdmission_AFailedPushKeepsItsWritesBesideTheOpenWindows(t *testing.T) { + f := newAdmissionFixture(t, "admission-outage", nil) + f.loop.cooldownHoldsSaves = false + f.loop.pushPending() + restore := failPushes(t, f.worker) + f.writeBy("alice", "x", "x2") + f.writeBy("bob", "z", "z2") + f.loop.lastPushAt = time.Time{} + f.loop.finalizeWindow(f.loop.windows.all()[0], windowFinalizeReasonIdleTimeout) + f.loop.endWake(0) + require.True(t, f.loop.retry.pending()) + require.Len(t, f.loop.pendingWrites, 1) + + f.writeBy("carol", "q", "q3") + f.writeBy("bob", "z", "z3") + assert.Equal(t, []string{"bob", "carol"}, authorsOf(f.loop.windows.all())) + f.loop.finalizeOpenWindowWithReason(windowFinalizeReasonShutdown) + f.loop.endWake(0) + require.Len(t, f.loop.pendingWrites, 3) + assert.Equal(t, 3, f.loop.materializedPrefix(), "each is committed locally") + assert.Equal(t, echoDocument("x", "x1"), f.onMain(echoFile("x")), "and none is published") + + restore() + f.loop.retry.due = time.Now().Add(-time.Second) + fireRetry(f.loop) + assert.Empty(t, f.loop.pendingWrites) + assert.Equal(t, echoDocument("x", "x2"), f.onMain(echoFile("x"))) + assert.Equal(t, echoDocument("z", "z3"), f.onMain(echoFile("z"))) + assert.Equal(t, echoDocument("q", "q3"), f.onMain(echoFile("q"))) +} + func TestWindowAdmission_ShutdownClosesEveryWindowOldestFirst(t *testing.T) { f := newAdmissionFixture(t, "admission-shutdown", nil) f.writeBy("alice", "x", "x2") @@ -164,6 +244,34 @@ func TestWindowAdmission_ShutdownClosesEveryWindowOldestFirst(t *testing.T) { assert.Equal(t, echoDocument("z", "z2"), f.onMain(echoFile("z"))) } +// TestWindowAdmission_ShutdownDecidesTheWritesAlreadyQueued pins the writes the worker accepted and +// had not handled when the shutdown began: Enqueue told the watch they were taken, so the last push +// carries them, each in its author's window, and the items the loop does not decide at shutdown are +// still drained. +func TestWindowAdmission_ShutdownDecidesTheWritesAlreadyQueued(t *testing.T) { + f := newAdmissionFixture(t, "admission-shutdown-queued", nil) + f.loop.pushPending() + f.writeBy("alice", "x", "x2") + queued := func(author, name, value string) Event { + event := echoEvent(name, value) + event.UserInfo.Username = author + return event + } + require.True(t, f.worker.Enqueue(queued("alice", "x", "x3"))) + f.worker.EnqueueAttach(&AttachCommitRequest{Namespace: "default", Name: "save", Author: "carol"}) + require.True(t, f.worker.Enqueue(queued("bob", "z", "z2"))) + + f.loop.handleShutdown() + + assert.Empty(t, f.worker.eventQueue) + assert.Equal(t, int64(0), f.worker.inflightItems.Load(), "every queued item is released") + assert.Equal(t, echoDocument("x", "x3"), f.onMain(echoFile("x")), "alice's queued write is published") + assert.Equal(t, echoDocument("z", "z2"), f.onMain(echoFile("z")), "bob's queued write is published") + assert.Equal(t, []string{"alice", "bob"}, + splitLines(gitIn(t, f.repoDir, "log", "--reverse", "--format=%an", "-2", "main")), + "each in its author's window") +} + func TestWindowAdmission_AWriteOfAnotherGitTargetClosesTheOpenWindows(t *testing.T) { f := newAdmissionFixture(t, "admission-other-target", nil) reader := captureInfo(f.worker) @@ -407,6 +515,45 @@ func TestWindowAdmission_AFallbackLeavesOneWindowOpen(t *testing.T) { assert.Len(t, reader.matching(`"msg"="Finalizing open commit window" "reason"="footprint-fallback"`), 2) } +// TestWindowAdmission_TheGaugesShowConcurrencyAndAFallback pins what an operator sees of concurrent +// windows: how many are open side by side, and that a worker has fallen back to one at a time. +func TestWindowAdmission_TheGaugesShowConcurrencyAndAFallback(t *testing.T) { + reader, err := telemetry.InitTestExporter() + require.NoError(t, err) + f := newAdmissionFixture(t, "admission-gauges", nil) + manager := &WorkerManager{Log: logr.Discard(), workers: map[BranchKey]*BranchWorker{{Branch: "main"}: f.worker}} + manager.setPublicationGaugeSources() + t.Cleanup(func() { + for _, name := range publicationGauges { + telemetry.SetGaugeSource(name, nil) + } + }) + gauge := func(name string) int64 { + t.Helper() + value, ok := telemetry.CollectInt64Sum(reader, "gitopsreverser_"+name, map[string]string{ + "gitbranch_namespace": f.worker.GitBranch.Namespace, + "gitbranch_name": f.worker.GitBranch.Name, + }) + require.True(t, ok, name) + return value + } + + f.writeBy("alice", "x", "x2") + f.writeBy("bob", "z", "z2") + assert.Equal(t, int64(2), gauge(telemetry.GaugeGitCommitWindowsOpen)) + assert.Zero(t, gauge(telemetry.GaugeGitCommitWindowsSingle)) + + 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)) + f.loop.publishLoopState(0) + + assert.Equal(t, int64(1), gauge(telemetry.GaugeGitCommitWindowsOpen)) + assert.Equal(t, int64(1), gauge(telemetry.GaugeGitCommitWindowsSingle), "the fallback lasts") +} + 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 fe5e4ed6..01af0d4b 100644 --- a/internal/git/window_model_test.go +++ b/internal/git/window_model_test.go @@ -19,6 +19,7 @@ import ( "testing" "time" + "github.com/go-git/go-git/v6/plumbing" "github.com/stretchr/testify/require" "gopkg.in/yaml.v3" @@ -1314,19 +1315,64 @@ func (r *windowModelRun) checkCommit(what, hash string, want modelCommit) { } } +// TestWindowModel_AFootprintCheckVouchesOnlyForItsOwnChange pins the model's evidence that a commit +// was checked against its footprint. Alice's commit x2 is checked; bob's x3 and carol's return to x2 +// are made with the check skipped. Carol's commit has the tree alice's had, from another parent, so +// matching the log by tree alone took alice's check as carol's. +func TestWindowModel_AFootprintCheckVouchesOnlyForItsOwnChange(t *testing.T) { + f := newAdmissionFixture(t, "model-footprint-witness", nil) + f.loop.pushPending() + logs := captureInfo(f.worker) + f.writeBy("sentinel", "z", "z2") // company, so every window after it is proven + commit := func(author, value string, skipCheck bool) string { + f.writeBy(author, "x", value) + windows := f.loop.windows.all() + require.Len(t, windows, 2) + if skipCheck { + windows[1].footprint = nil + } + f.loop.lastPushAt = time.Time{} + f.loop.finalizeWindow(windows[1], windowFinalizeReasonIdleTimeout) + f.loop.endWake(0) + f.loop.pushPending() + return gitIn(t, f.repoDir, "rev-parse", "main") + } + checked := commit("alice", "x2", false) + commit("bob", "x3", true) + unchecked := commit("carol", "x2", true) + require.Equal(t, gitIn(t, f.repoDir, "rev-parse", checked+"^{tree}"), + gitIn(t, f.repoDir, "rev-parse", unchecked+"^{tree}"), "precondition: carol returns to alice's tree") + require.Len(t, logs.matching(decisionFootprintChecked), 1, "precondition: only alice's commit was checked") + + r := &windowModelRun{t: t, f: f, logs: logs} + require.True(t, r.footprintChecked(checked)) + require.False(t, r.footprintChecked(unchecked), "alice's check is not evidence for carol's commit") +} + // 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. +// itself: one that skipped it would let a footprint bug through unrefused. 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 { + if !r.footprintChecked(hash) { 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) } } +// footprintChecked reports whether the worker logged a footprint check of the commit's change. It is +// matched by the commit's tree and its parent's: a rebuild of the prefix onto the same parent (a +// resync's fetch) makes the checked commit again, under another hash, while a later commit that +// returns to a checked tree from another parent is another change. +func (r *windowModelRun) footprintChecked(hash string) bool { + r.t.Helper() + tree := gitIn(r.t, r.f.repoDir, "rev-parse", hash+"^{tree}") + parentTree := plumbing.ZeroHash.String() + if gitIn(r.t, r.f.repoDir, "rev-list", "--parents", "-n", "1", hash) != hash { + parentTree = gitIn(r.t, r.f.repoDir, "rev-parse", hash+"^^{tree}") + } + return len(r.logs.matching(`"tree"="`+tree+`" "parentTree"="`+parentTree+`" "footprint"=`)) > 0 +} + // 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 89d27eb1..681c7af7 100644 --- a/internal/git/withheld_fields.go +++ b/internal/git/withheld_fields.go @@ -46,9 +46,25 @@ import ( // 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 +// withheldFields is, by withheldKey, the fields withheld applies reverted that no event has changed // since. It is plain data. -type withheldFields map[string]withheldObject +type withheldFields map[withheldKey]withheldObject + +// withheldKey names an object as one GitTarget mirrors it. Two GitTargets on one branch can mirror +// objects of one name, from two clusters or into two folders: each has an accepted version and a +// record of its own, and an apply withheld for one never changes or holds out the other's. +type withheldKey struct { + GitTarget itypes.ResourceReference + Object string +} + +// withheldKeyOf is the withheldKey of a live event's object. +func withheldKeyOf(event Event) withheldKey { + return withheldKey{ + GitTarget: itypes.NewResourceReference(event.GitTargetName, event.GitTargetNamespace), + Object: objectKey(event.Identifier), + } +} // 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. @@ -85,14 +101,14 @@ func (v fieldValue) equal(other fieldValue) bool { // 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) + key := withheldKeyOf(event) 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{} + l.accepted = map[withheldKey]*unstructured.Unstructured{} } if deleted { l.accepted[key] = nil @@ -111,27 +127,25 @@ func (l *branchWorkerEventLoop) forgetAccepted(now time.Time) { } } -// editRecorded reports whether an open apply window, or the flush in progress, records an edit of -// the object. -func (l *branchWorkerEventLoop) editRecorded(key string, now time.Time) bool { +// editRecorded reports whether an open apply window of an applier that covers the object's +// GitTarget, or the flush in progress for that GitTarget, records an edit of the object. +func (l *branchWorkerEventLoop) editRecorded(key withheldKey, now time.Time) bool { for _, state := range l.appliers { window := state.window - if !window.open(now) { + if !window.open(now) || !slices.ContainsFunc(state.targets, func(target branchTarget) bool { + return target.ref == key.GitTarget + }) { continue } - if _, edited := window.edited[key]; edited { - return true - } - if _, carried := window.carried[key]; carried && window.carrying(now) { + if _, edited := window.edited[key.Object]; edited { return true } - } - for _, edits := range l.flushEdits { - if _, edited := edits[key]; edited { + if _, carried := window.carried[key.Object]; carried && window.carrying(now) { return true } } - return false + _, edited := l.flushEdits[key.GitTarget][key.Object] + return edited } // recordWithheld records the fields a withheld apply reverted: where its object differs from the @@ -142,7 +156,7 @@ func (l *branchWorkerEventLoop) editRecorded(key string, now time.Time) bool { // 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) + key := withheldKeyOf(event) accepted, kept := l.accepted[key] if !kept || event.Object == nil { return @@ -166,7 +180,7 @@ func (l *branchWorkerEventLoop) recordWithheld(event Event) { l.withheld = withheldFields{} } l.withheld[key] = withheldObject{ - GitTarget: itypes.NewResourceReference(event.GitTargetName, event.GitTargetNamespace), + GitTarget: key.GitTarget, Identifier: event.Identifier, Fields: append(fields, reverted...), } @@ -179,7 +193,7 @@ func (l *branchWorkerEventLoop) recordWithheld(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) + key := withheldKeyOf(*event) if event.Operation == string(itypes.OperationDelete) { delete(l.withheld, key) return false @@ -211,20 +225,20 @@ func (l *branchWorkerEventLoop) restoreWithheldSnapshot( if len(l.withheld) == 0 { return desired } - found := make(map[string]bool, len(req.Desired)) + target := itypes.NewResourceReference(req.GitTargetName, req.GitTargetNamespace) + found := make(map[withheldKey]bool, len(req.Desired)) for _, dr := range req.Desired { - found[objectKey(dr.Resource)] = true + found[withheldKey{GitTarget: target, Object: 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] { + if key.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) + object := l.restoreWithheld(withheldKey{GitTarget: target, Object: objectKey(dr.Resource)}, dr.Object) if object != dr.Object { changed = true } @@ -243,10 +257,8 @@ func (l *branchWorkerEventLoop) restoreWithheldSnapshot( // 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 - }) { + for key := range l.withheld { + if slices.ContainsFunc(state.targets, func(target branchTarget) bool { return target.ref == key.GitTarget }) { delete(l.withheld, key) } } @@ -258,7 +270,7 @@ func (l *branchWorkerEventLoop) dropWithheldOf(state *applierState) { // 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, + key withheldKey, object *unstructured.Unstructured, ) *unstructured.Unstructured { fields := l.withheld[key].Fields if len(fields) == 0 || object == nil { diff --git a/internal/telemetry/exporter.go b/internal/telemetry/exporter.go index 1cfcee50..345296fe 100644 --- a/internal/telemetry/exporter.go +++ b/internal/telemetry/exporter.go @@ -759,6 +759,8 @@ func registerObservableGauges() error { {"gitopsreverser_git_branch_targets", GaugeGitBranchTargets}, {"gitopsreverser_apply_operations_unresolved", GaugeApplyOperationsUnresolved}, {"gitopsreverser_apply_appliers_held", GaugeAppliersHeld}, + {"gitopsreverser_git_commit_windows_open", GaugeGitCommitWindowsOpen}, + {"gitopsreverser_git_commit_windows_single", GaugeGitCommitWindowsSingle}, } for _, o := range observable { if _, err := otelMeter.Int64ObservableGauge( diff --git a/internal/telemetry/gauges.go b/internal/telemetry/gauges.go index 72359323..b7098943 100644 --- a/internal/telemetry/gauges.go +++ b/internal/telemetry/gauges.go @@ -68,6 +68,13 @@ const ( // snapshot barrier holds back: a GitTarget they cover is not mirrored yet, or its folder is not on // the branch. Read from what the branch worker's loop last published. GaugeAppliersHeld = "apply_appliers_held" + // GaugeGitCommitWindowsOpen is how many commit windows a branch worker holds open, one per author + // whose files are proven not to overlap the others'. GaugeGitCommitWindowsSingle is 1 once the + // worker has fallen back to one window at a time, for the rest of its life, after a commit would + // have changed a file outside its window's footprint. Both are read from what the loop last + // published. + GaugeGitCommitWindowsOpen = "git_commit_windows_open" + GaugeGitCommitWindowsSingle = "git_commit_windows_single" ) var ( diff --git a/test/e2e/concurrent_authors_e2e_test.go b/test/e2e/concurrent_authors_e2e_test.go new file mode 100644 index 00000000..4c896093 --- /dev/null +++ b/test/e2e/concurrent_authors_e2e_test.go @@ -0,0 +1,179 @@ +// SPDX-License-Identifier: Apache-2.0 + +package e2e + +import ( + "fmt" + "path" + "slices" + "strings" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// Concurrent Authors is the flagship of docs/design/concurrent-author-windows.md: two people work on +// one GitTarget at the same time, each on objects of their own, and each saves when done. Every +// write goes through the real path (kubectl as that person → kube-apiserver → audit webhook and +// watch → attribution → branch worker), and each save must come out as one commit of its own: +// its author, its message, and exactly that person's files. +// +// The GitTarget's window is long, so no timer closes a window during the spec: only the saves do. +// Before concurrent windows, bob's first write closed alice's window (another author), so alice's +// first Deployment was committed alone under a generated message, and her save carried only what +// she wrote after bob. Both the quiet wait and the commit contents below would catch that. +// +// The branch is seeded first: a second window opens beside the first only once the folder's +// bootstrap files are committed, so on an empty repository the first save would be serial anyway. +// +// Not Serial: the spec owns its namespace and Gitea repository. +var _ = Describe("Concurrent Authors", Label("concurrent-authors", "commit-request", "audit-consumer"), Ordered, + func() { + const ( + basePath = "e2e/concurrent-authors" + commitWindow = "120s" + ) + var p windowedDeploymentPipeline + + BeforeAll(func() { + p = setupWindowedDeploymentPipeline("concurrent-authors", basePath, commitWindow) + }) + + AfterAll(func() { + cleanupPipeline(p.ns, p.gitProvider, p.gitTarget, p.watchRule) + cleanupNamespace(p.ns) + }) + + It("gives two people working at the same time one commit each", func() { + seed := GinkgoRandomSeed() + alice := author{user: "alice", message: "feat: add the storefront"} + bob := author{user: "bob", message: "feat: add the order queue"} + aliceAPI := fmt.Sprintf("storefront-api-%d", seed) + aliceWeb := fmt.Sprintf("storefront-web-%d", seed) + bobQueue := fmt.Sprintf("order-queue-%d", seed) + bobCron := fmt.Sprintf("order-cleanup-%d", seed) + file := func(name string) string { + return path.Join(basePath, p.ns, "apps/deployments", name+".yaml") + } + + By("seeding the branch, so the folder's bootstrap files are committed") + seedName := fmt.Sprintf("seed-%d", seed) + var bundle strings.Builder + bundle.WriteString(commitRequestManifest(p.ns, "save-seed", p.gitTarget, "chore: seed", "8s")) + bundle.WriteString("---\n") + bundle.WriteString(deploymentManifest(p.ns, seedName)) + _, err := kubectlRunWithStdin(p.ns, bundle.String(), "apply", "-f", "-") + Expect(err).NotTo(HaveOccurred()) + var seedSHA string + Eventually(func(g Gomega) { + seedSHA = expectCommitRequestCommitted(g, p.ns, "save-seed", + recentCommitDiagnostics(p.repo.CheckoutDir, basePath)) + }, 2*time.Minute, 2*time.Second).Should(Succeed()) + + By("alice and bob each start work, interleaved") + alice.create(p.ns, aliceAPI) + bob.create(p.ns, bobQueue) + alice.create(p.ns, aliceWeb) + bob.create(p.ns, bobCron) + alice.label(p.ns, aliceAPI, "tier=frontend") + + By("nothing is committed while both are still working") + Consistently(func(g Gomega) { + pullLatestRepoState(g, p.repo.CheckoutDir) + g.Expect(headOf(g, p.repo.CheckoutDir)).To(Equal(seedSHA), + "a window closed before its author saved\n%s", + recentCommitDiagnostics(p.repo.CheckoutDir, basePath)) + }, 10*time.Second, 2*time.Second).Should(Succeed()) + + By("alice saves, then bob saves") + alice.save(p.ns, "save-storefront", p.gitTarget) + bob.save(p.ns, "save-order-queue", p.gitTarget) + + var aliceSHA, bobSHA string + Eventually(func(g Gomega) { + diagnostics := recentCommitDiagnostics(p.repo.CheckoutDir, basePath) + aliceSHA = expectCommitRequestCommitted(g, p.ns, "save-storefront", diagnostics) + bobSHA = expectCommitRequestCommitted(g, p.ns, "save-order-queue", diagnostics) + }, 2*time.Minute, 2*time.Second).Should(Succeed()) + + By("each save is one commit: its author, its message, its files and nothing else") + Eventually(func(g Gomega) { + pullLatestRepoState(g, p.repo.CheckoutDir) + diagnostics := recentCommitDiagnostics(p.repo.CheckoutDir, basePath) + g.Expect(commitsSince(g, p.repo.CheckoutDir, seedSHA)).To(ConsistOf(aliceSHA, bobSHA), + "exactly the two saves were committed after the seed\n%s", diagnostics) + + alice.expectCommit(g, p.repo.CheckoutDir, aliceSHA, file(aliceAPI), file(aliceWeb)) + bob.expectCommit(g, p.repo.CheckoutDir, bobSHA, file(bobQueue), file(bobCron)) + + labelled, showErr := gitRun(p.repo.CheckoutDir, "show", aliceSHA+":"+file(aliceAPI)) + g.Expect(showErr).NotTo(HaveOccurred()) + g.Expect(labelled).To(ContainSubstring("tier: frontend"), + "alice's update after bob's writes is in her commit") + }, time.Minute, 2*time.Second).Should(Succeed()) + }) + }) + +// author is a person working through kubectl as themselves: every write is impersonated, so the +// audit event names them, and system:masters spares each spec its own RBAC. +type author struct { + user string + message string +} + +func (a author) kubectl(namespace, stdin string, args ...string) { + GinkgoHelper() + args = append(args, "--as="+a.user, "--as-group=system:masters") + out, err := kubectlRunWithStdin(namespace, stdin, args...) + Expect(err).NotTo(HaveOccurred(), "kubectl %v as %s: %s", args, a.user, out) +} + +func (a author) create(namespace, name string) { + GinkgoHelper() + a.kubectl(namespace, deploymentManifest(namespace, name), "create", "-f", "-") +} + +func (a author) label(namespace, name, label string) { + GinkgoHelper() + a.kubectl(namespace, "", "label", "deployment", name, label) +} + +// save creates the person's CommitRequest. Its window attaches to the window their writes opened +// (attach: CurrentOrNext) and closes it 8s later. +func (a author) save(namespace, name, gitTarget string) { + GinkgoHelper() + a.kubectl(namespace, commitRequestManifest(namespace, name, gitTarget, a.message, "8s"), "create", "-f", "-") +} + +// expectCommit asserts sha is this person's save: their name and address as the audit event gave +// them, their message verbatim, and exactly the files they wrote. +func (a author) expectCommit(g Gomega, checkoutDir, sha string, files ...string) { + out, err := gitRun(checkoutDir, "show", "-s", "--format=%an <%ae>%n%B", sha) + g.Expect(err).NotTo(HaveOccurred()) + authorLine, message, _ := strings.Cut(out, "\n") + g.Expect(authorLine).To(Equal(fmt.Sprintf("%s <%s@noreply.cluster.local>", a.user, a.user)), + "commit %s is %s's", sha[:8], a.user) + g.Expect(strings.TrimSpace(message)).To(Equal(a.message), "commit %s carries %s's message", sha[:8], a.user) + + changed, err := gitRun(checkoutDir, "show", "--name-only", "--format=", sha) + g.Expect(err).NotTo(HaveOccurred()) + paths := strings.Fields(changed) + slices.Sort(paths) + want := slices.Sorted(slices.Values(files)) + g.Expect(paths).To(Equal(want), "commit %s changes only %s's files", sha[:8], a.user) +} + +// headOf is the checkout's HEAD. +func headOf(g Gomega, checkoutDir string) string { + out, err := gitRun(checkoutDir, "rev-parse", "HEAD") + g.Expect(err).NotTo(HaveOccurred()) + return strings.TrimSpace(out) +} + +// commitsSince lists the commits on HEAD after base. +func commitsSince(g Gomega, checkoutDir, base string) []string { + out, err := gitRun(checkoutDir, "rev-list", base+"..HEAD") + g.Expect(err).NotTo(HaveOccurred()) + return strings.Fields(out) +}