diff --git a/cmd/main.go b/cmd/main.go index 7da0cc03..66457091 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -328,13 +328,18 @@ func main() { fatalIfErr(mgr.Add(watchMgr), "unable to add watch ingestion manager") // CommitRequests created on a source cluster, next to the edits they save: served through every // KubeProvider whose spec.commitRequests reconciles them, attributed from that cluster's audit facts. + // A worker's change to a CommitRequest takes the request's next step at once, on either path. + commitRequestWakes := controller.NewCommitRequestWakes(workerManager.CommitRequestChanges()) + fatalIfErr(mgr.Add(commitRequestWakes), "unable to add the CommitRequest wakes") commitRequestServer := &controller.CommitRequestServer{ Client: mgr.GetClient(), + APIReader: mgr.GetAPIReader(), Clusters: watchMgr, Finalizer: eventRouter, Authors: watchMgr.AuthorResolver, FactStreams: watchMgr.FactStreams, TTL: cfg.commitRequestTTL, + Wakes: commitRequestWakes, Log: ctrl.Log.WithName("commitrequests"), } fatalIfErr(mgr.Add(commitRequestServer), "unable to add the CommitRequest server") @@ -423,6 +428,7 @@ func main() { Authors: watchMgr.AuthorResolver, FactStreams: watchMgr.FactStreams, TTL: cfg.commitRequestTTL, + Wakes: commitRequestWakes, }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "unable to create controller", "controller", "CommitRequest") os.Exit(1) diff --git a/docs/TODO.md b/docs/TODO.md index 6d89effc..87820ea6 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -2,15 +2,11 @@ The open work, in one list. Each item is one or two lines with a code pointer or the design page that argues it. Ranked within each group, most urgent first. **Before the next release** marks what -should land before the next tag is cut (the last release is v0.53.0). Checked against `main` at +should land before the next tag is cut (the last release is v0.55.0). Checked against `main` at `1bc97904` on 2026-10-24. ## Before the next release -Nothing open. The 0.54.0 release PR's changelog was checked against what ships on 2026-10-24. - -## Applier claims and bi-directional editing - - [ ] **A withheld apply rides the next edit.** S1 withholds an apply that lands over an edit decided after the flush mark, and a `Repair` puts the edit back. But the apply did land on the cluster, so the next edit of that object carries the apply's value for every field it did not change, and the @@ -18,6 +14,9 @@ Nothing open. The 0.54.0 release PR's changelog was checked against what ships o `Repair` then applies the loss. The failing test is `TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit` (`internal/git/apply_window_carry_test.go`, skipped; `GITOPS_REVERSER_KNOWN_BUGS=1` runs it). + +## Applier claims and bi-directional editing + - [ ] **Reopen the render-fidelity gate after a Git repair, automatically.** Today a target on `RenderMatchesLive=False` clears only on an explicit recheck (the `reconcile.configbutler.ai/requestedAt` annotation, which restarts every scope; the render diff --git a/docs/UPGRADING.md b/docs/UPGRADING.md index e482c909..d57ecc6a 100644 --- a/docs/UPGRADING.md +++ b/docs/UPGRADING.md @@ -7,6 +7,36 @@ guidance that the changelog's breaking-change entries link to. We are pre-1.0, so breaking changes bump the **minor** version (release-please is configured with `bump-minor-pre-major`) rather than the major. Read the relevant entry before upgrading across it. +## A `CommitRequest` reports each phase as the worker reaches it + +A request's progress and outcome are written as soon as its branch worker reports them, instead of +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. + +`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 +`WaitingForWindow` or `CollectingWindow` before it makes the save's edits, not only for +`Ready=Unknown`. + +## A source-cluster request no path can serve ends `NotAllowed` + +A `CommitRequest` on a source cluster whose `GitTarget` mirrors another cluster, either the cluster +GitOps Reverser runs in or one whose `KubeProvider` serves no `CommitRequest`s, ends `Ready=False` +with reason `NotAllowed`, naming both clusters. It used to keep no status at all. It finishes, so on +a `KubeProvider` with `commitRequests: ReconcileAndDeleteFinished` it is deleted after the TTL. + +## A save is pushed at once + +A push that a `CommitRequest` waits on no longer waits out the five-second push cooldown, as a push +an apply waits on already did not. Two saves in a row each resolve as soon as their commit is +pushed. Commits that no save waits on still share one push per cooldown. + ## Upgrading from 0.54.0: drain the `CommitRequest`s first Existing providers, branches, targets and claims keep running across this upgrade; none of the diff --git a/docs/api-first-publication.md b/docs/api-first-publication.md index 2ef82ddf..d7926290 100644 --- a/docs/api-first-publication.md +++ b/docs/api-first-publication.md @@ -131,6 +131,7 @@ a publication succeeds. These clocks overlap. | Retained-byte threshold | Default `8Mi` | Estimated open plus retained bytes reach the threshold | Close the open window under memory pressure | | Shutdown | Immediate attempt | The worker exits | Attempt to publish finalized work without waiting for cooldown | | Apply due | Immediate | An applier on the branch has an apply due | Close the open window and publish every write decided so far without waiting for cooldown, so the apply includes them | +| Save | Immediate | A pending write carries a `CommitRequest` | Publish without waiting for cooldown, because a person waits for the save; one save per author at a time keeps this from becoming a burst | The first publication can push as soon as a commit is ready. There is no initial five-second push wait. Setting `commit.window.idleTimeout: 0s` makes local commits immediately; subsequent publications @@ -282,11 +283,11 @@ are never held by the same worker. ## Story 3: save now, and know what reached Git -A `CommitRequest` attaches to a matching author-and-target window and closes it. It does not -bypass the push cooldown. Two settings with separate clocks set the timing: `attachTimeout` runs -from the worker's registration of the request and bounds the wait for a window, and `maxDuration` -runs from the attach and bounds the collection that follows. Both default to `2s` and accept `0s` -to `5m`. A repeat attach restarts neither clock. +A `CommitRequest` attaches to a matching author-and-target window and closes it. Its commit is +pushed at once, without waiting for the push cooldown. Two settings with separate clocks set the +timing: `attachTimeout` runs from the worker's registration of the request and bounds the wait for +a window, and `maxDuration` runs from the attach and bounds the collection that follows. They +default to `5s` and `2s` and accept `0s` to `5m`. A repeat attach restarts neither clock. ```mermaid sequenceDiagram diff --git a/docs/architecture.md b/docs/architecture.md index 7769a22d..daba457e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1553,8 +1553,10 @@ out, the retained buffer reaches `--branch-buffer-max-size` (default `8Mi`), an or an apply comes due for an applier on the branch. The timers are the GitTarget's `spec.commit.window` until a `CommitRequest` attaches and replaces them. Successful local commits are retained until a fixed push cooldown (`5s`) allows a push, which prevents remote push storms during -bursts. The one exception is an apply's flush: the writes decided before the apply came due are -pushed at once, cooldown or not, because the apply waits for them. +bursts. Two things push at once, cooldown or not, because something waits for the push: an apply's +flush, for the writes decided before the apply came due, and a `CommitRequest`'s commit, for the +person who saved. Saves cannot burst the way plain writes can, since an author has one save in +progress at a time, so the cooldown still paces every stream of plain writes. ### Local clones and conflict retry diff --git a/docs/bi-directional-argocd.md b/docs/bi-directional-argocd.md index abc18d9d..22bc3ac7 100644 --- a/docs/bi-directional-argocd.md +++ b/docs/bi-directional-argocd.md @@ -430,15 +430,16 @@ applier on the branch, whether or not the push touched its folder, and a `CatchU create in the folder and when a `GitTarget` first names the applier claim. **The apply window.** Until 10 seconds after the sync's operation completes, however long that -takes, Argo CD's own write to an object nobody edited is the sync landing, and is not mirrored. An -API edit to a field the `Application` sets, one made after the flush included, can be reverted on -the cluster by the sync. GitOps Reverser recognizes that write by Argo CD's field manager -(`argocd-controller`, for a client-side and a server-side apply alike, or -`argocd-application-controller` for a `Replace=true` sync), withholds it from Git, and asks for a -`Repair` sync pinned to the tip that holds the edit, but only when Argo CD's entry is as new as the -edit or newer: undoing an edit can leave an older entry of Argo CD's as the newest, and that undo -is mirrored. When Argo CD and another manager wrote in the same second, the write is mirrored and -counted as `ambiguous`. +takes, and until the watch has delivered every write the cluster held then, for at most two minutes +([the life of an apply](bi-directional.md#the-life-of-an-apply)), Argo CD's own write to an object +nobody edited is the sync landing, and is not mirrored. An API edit to a field the `Application` +sets, one made after the flush included, can be reverted on the cluster by the sync. GitOps Reverser +recognizes that write by Argo CD's field manager (`argocd-controller`, for a client-side and a +server-side apply alike, or `argocd-application-controller` for a `Replace=true` sync), withholds it +from Git, and asks for a `Repair` sync pinned to the tip that holds the edit, but only when Argo +CD's entry is as new as the edit or newer: undoing an edit can leave an older entry of Argo CD's as +the newest, and that undo is mirrored. When Argo CD and another manager wrote in the same second, +the write is mirrored and counted as `ambiguous`. A request that fails, and a sync that ends `Failed` or `Error`, is decided again from scratch, after that `Application`'s own backoff. A sync still running after 5 minutes is `Unresolved`: GitOps diff --git a/docs/bi-directional-flux.md b/docs/bi-directional-flux.md index 13efc35a..7d1d2119 100644 --- a/docs/bi-directional-flux.md +++ b/docs/bi-directional-flux.md @@ -461,11 +461,13 @@ as failed, and records an `OperationUnfollowed` Event. A resumed `Kustomization` nothing watching it. **The apply window.** From the moment an operation is asked for until 10 seconds after it completes, -a write by `kustomize-controller` alone to an object nobody edited is the apply landing, and is not -mirrored. A write by `kustomize-controller` over a newer API edit, one made after the flush included, -is withheld from Git, and GitOps Reverser asks for a `Repair`; the applier claim shows `repairPending` -until that apply succeeds. Only a `kustomize-controller` entry as new as the edit or newer counts: -undoing an edit can leave an older one as the newest entry, and that undo is mirrored. When +and until the watch has delivered every write the cluster held then, for at most two minutes ([the +life of an apply](bi-directional.md#the-life-of-an-apply)), a write by `kustomize-controller` alone +to an object nobody edited is the apply landing, and is not mirrored. A write by +`kustomize-controller` over a newer API edit, one made after the flush included, is withheld from +Git, and GitOps Reverser asks for a `Repair`; the applier claim shows `repairPending` until that +apply succeeds. Only a `kustomize-controller` entry as new as the edit or newer counts: undoing an +edit can leave an older one as the newest entry, and that undo is mirrored. When `kustomize-controller` and another manager wrote in the same second, the write is mirrored and counted as `ambiguous`. diff --git a/docs/configuration.md b/docs/configuration.md index c310b351..fd44a76d 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -3023,6 +3023,14 @@ The `GitTarget` a request names decides where it belongs, through the `KubeProvi other one's. Create it on the other cluster instead. When that `KubeProvider` does serve them, the request is left alone, since the kubeconfig may point back at this cluster and then its own watch serves the request. If it points elsewhere, nothing serves the request and it gets no status. +- **A request on a source cluster, for a `GitTarget` that mirrors another cluster, is refused** with + reason `NotAllowed`, and the message names both clusters, when nothing can serve it from there: + the `GitTarget` mirrors the cluster GitOps Reverser runs in, or its `KubeProvider` serves no + `CommitRequest`s. Create it on the cluster the `GitTarget` mirrors. A refused request finishes, + so a `KubeProvider` with `ReconcileAndDeleteFinished` deletes it after the TTL. A kubeconfig that + points back at the cluster GitOps Reverser runs in sees that cluster's own requests, and leaves + them to it. A request whose `GitTarget`'s `KubeProvider` does serve `CommitRequest`s is left + alone, since that `KubeProvider` may connect to the same cluster under another identity. Only the path the `GitTarget`'s `KubeProvider` picks writes a request's progress, so several `KubeProvider`s that watch one cluster never fight over a request. Any of them may report @@ -3085,12 +3093,13 @@ A request is registered with the branch worker only once its own audit event has and its edit is attributed the same way, so a request created just before its edit does not always reach the worker first. On a `GitTarget` whose windows close on the first silence (`idleTimeout: "0s"`), such an edit can commit alone before the request is there to attach to it. Make the edit -once the request is claimed (`Ready=Unknown`), or give the `GitTarget` an `idleTimeout` that covers -the gap. +once the request is registered, when its `Reconciling` reason is `WaitingForWindow` (see +[its status](#its-status)), or give the `GitTarget` an `idleTimeout` that covers the gap. +`Ready=Unknown` alone is too early: it starts at reason `Saving`, before the registration. `maxDuration`, and `idleTimeout` when set, cover the span of the save's own writes after the attach. The worst case from registration to the finalize is `attachTimeout` plus `maxDuration`; the -push follows, with its cooldown and any retries. +push follows at once, without the push cooldown, with any retries a remote that refuses it needs. The two directions are not symmetric. Overshooting costs a few seconds of latency on the commit; undershooting `attachTimeout` resolves the request `Ready=True` with reason `NoWindow` while the @@ -3104,6 +3113,59 @@ request never resolves `NoWindow` while its commit waits out the push cooldown. and a request together gives no ordering guarantee between controllers and watch streams. See the [request contract](spec/commitrequest-design.md). +#### Saving a folder with `kubectl apply` + +To save the objects of a folder in one commit with your message, put a `CommitRequest` in the +folder, next to them, and give it a new name for every apply: + +```yaml +# save-2026-10-28-checkout-limits.yaml +apiVersion: configbutler.ai/v1alpha3 +kind: CommitRequest +metadata: + name: save-2026-10-28-checkout-limits +spec: + gitTargetRef: + name: example-target + message: "feat: raise the checkout limits" +``` + +`kubectl apply -f folder/` then commits the folder's changes, authored by you, with the request's +message. kubectl applies the files in name order, so the request can land before or after the +objects: with `attach: CurrentOrNext` the request attaches to the window the objects already opened, +or waits for the one they open. Do not use `attach: Next` here: it would close the objects' window +under the generated message, then wait for a write that never comes. Another unclaimed edit of yours +on the same `GitTarget`, still in its window, shares the commit. + +It is one commit only when the window holds every write, and a window is not a transaction. Two +things split it: + +- **Time.** With the defaults the window closes 2 seconds after the attach. Writes reach the branch + worker once their audit facts have named their author, so on a large folder, a loaded API server or + a source cluster with slow audit delivery, the last writes can arrive later than kubectl sent them. + 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. + +The request then ends `Committed` with the writes its window held. + +The name is what makes the next apply a save. Applying the same folder again does this: + +- **Nothing changed:** kubectl sends no write for an unchanged object, so there is nothing to save, + and the finished request keeps its outcome. If the request was already deleted (see + [finished requests](#finished-requests-are-deleted-after-48-hours)), the apply creates it again, + and it ends `NoWindow`. +- **Objects changed, same request:** the request is unchanged and finished, so it saves nothing. + The edits commit on the `GitTarget`'s own timers, under the generated message. +- **Message edited, same name:** the API server refuses the request, because its spec is + immutable, and kubectl exits with an error after applying the other files. Their edits commit + under the generated message. + +The request itself is never mirrored, even when a `WatchRule` covers its folder or its type. + ### Who may save: `spec.allowedCommitRequests` A `GitTarget` says which `CommitRequest`s may save its windows in `spec.allowedCommitRequests`, a @@ -3237,8 +3299,15 @@ request produced a pushed commit: message, so the two would race for it and one message would be lost without a word; the later 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 never race, since each - author has windows of its own. + 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. + + `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 + apart waits until the `Reconciling` reason is `WaitingForWindow` or `CollectingWindow` before it + makes the save's edits, not only for `Ready=Unknown`: a request still `Saving` is not registered + yet, and its edits could reach the window of the save before it. `AlreadyPresent` means **the remote confirmed there was nothing to add**, so it is reported after the push rather than when the local plan found no difference. That costs a few seconds and buys a @@ -3248,7 +3317,8 @@ request produced a pushed commit: - **Reconciling** / **Stalled**: the kstatus progress/blocked pair. `Reconciling=True` while the request is in progress, with the phase the branch worker reports as its reason: `Saving` (claimed, not yet registered), `WaitingForWorker` (the GitTarget has no branch worker yet), - `WaitingForWindow`, `CollectingWindow`, or `WaitingForPush`. + `WaitingForWindow`, `CollectingWindow`, or `WaitingForPush`. Each phase, and the outcome, is + written as soon as the branch worker reports it. `Stalled=True` when the request failed and needs attention (kstatus reports the object Failed). - **Pushed**: `True` once the commit, empty or not, is in the remote repository. - **Applied**: only with [`catchUp: AfterPush`](#applying-a-save-at-once), once `Ready=True`: diff --git a/docs/design/push-cooldown.md b/docs/design/push-cooldown.md index d390a3c6..44d98721 100644 --- a/docs/design/push-cooldown.md +++ b/docs/design/push-cooldown.md @@ -46,6 +46,7 @@ if l.awaitingRetry() { return } // a failed attempt has if l.lastPushAt.IsZero() || l.publishesAtOnce() { // never pushed, or an apply waits: go now l.pushPending(); return } +if l.pendingCarriesCommitRequest() { l.pushPending(); return } // a save waits: go now if time.Since(l.lastPushAt) >= PushCooldown { l.pushPending(); return } if l.pushTimer == nil { l.pushTimer = time.NewTimer(...) } // otherwise wait out the remainder ``` @@ -55,7 +56,9 @@ and does not pace retries. The publication backoff does that separately. Parent its attempts while active, and new commits wait for its deadline; see the [worker event model](branch-worker-event-model.md). A flush for an apply skips the cooldown: the writes up to the flush mark are pushed at once (`publishesAtOnce`), because the apply waits for -them ([the apply trigger](apply-trigger.md#flush-then-apply)). +them ([the apply trigger](apply-trigger.md#flush-then-apply)). A push that a `CommitRequest` waits on +skips it too: a person waits for the save, and an author has one save in progress at a time, so +saves cannot make the burst the cooldown exists to batch. Plain writes keep it. ## 2. Ordinary timer closures already space a single identity diff --git a/docs/spec/commitrequest-design.md b/docs/spec/commitrequest-design.md index 50d16c1e..b4ce2fc5 100644 --- a/docs/spec/commitrequest-design.md +++ b/docs/spec/commitrequest-design.md @@ -39,15 +39,19 @@ The named `GitTarget`'s `KubeProvider` alone picks the path that serves a reques | the control cluster | a kubeconfig, with a `Reconcile` value | left alone: the kubeconfig may point back at the control cluster, and then its watch serves it | | the control cluster | a kubeconfig, with `commitRequests: Ignore` | `NotAllowed`, saying to create the request on that cluster | | a source cluster that provider P watches | P | P's watch serves it | -| a source cluster that provider P watches | any other provider | left alone | +| a source cluster that provider P watches | `inCluster`, and the request is not the control cluster's own object (by UID) | `NotAllowed`, naming both clusters | +| a source cluster that provider P watches | `inCluster`, and the request is the control cluster's own object | left alone: P's kubeconfig points back at the control cluster, whose path serves it | +| a source cluster that provider P watches | another provider, with `commitRequests: Ignore` | `NotAllowed`, naming both clusters | +| a source cluster that provider P watches | another provider, with a `Reconcile` value | left alone: it may watch the same cluster | | a source cluster | `gitTargetRef` names no namespace | not picked up, and no status | | either | the `GitTarget` does not exist | `GitTargetNotFound` once `attachTimeout` has passed since creation | Three rules keep several observers of one request safe: 1. **A path writes only what it knows to be true.** Only the owner serves a request and writes its - progress. A path that is not the owner writes only `GitTargetNotFound`, and only into an empty - status, because nothing else is true for every observer. + progress. A path that is not the owner writes only `GitTargetNotFound`, or `NotAllowed` when it + can show no path serves the request, and only into an empty status, because nothing else is true + for every observer. 2. **Every status write is conditional on the resourceVersion its decision was made from.** On a conflict the writer reads the request again and decides again, so a `GitTargetNotFound` decided from an older version never overwrites an owner that has started. @@ -66,7 +70,10 @@ served, because its author would come from the control cluster's audit and its e other cluster's. A request on a source cluster must name the `GitTarget`'s namespace, because that namespace belongs to the control cluster; the path cannot refuse a request without one, since a kubeconfig pointing back at the control cluster sees the control cluster's requests, where an -omitted namespace is valid. Known gap: a control-cluster request whose `GitTarget`'s provider has a +omitted namespace is valid. A source-cluster request for a `GitTarget` that mirrors the control +cluster is refused only once an uncached read of the control cluster finds no request of that UID: +the same UID means the watch's kubeconfig points back at the control cluster, whose path serves it. +Known gap: a control-cluster request whose `GitTarget`'s provider has a `Reconcile` value and points somewhere else gets no status from anyone. ### Admission @@ -230,9 +237,12 @@ the remote, so the remote settles it: a no-diff request resolves `AlreadyPresent nothing to add, and `Committed` when the replay produced a commit after all. **A request whose write is decided stays identifiable until publication.** It remains attached to -that write, and the controller keeps re-sending its attach every couple of seconds until it reads -an outcome. The worker keeps that attachment even when local materialization is deferred, so -the re-send is recognized as the same request: it cannot register afresh and expire into +that write, and the controller keeps re-sending its attach until it reads an outcome: at once on +every change the worker reports about the request (`CommitRequestChanges`), and at the latest every +ten seconds, the fallback for a change that was dropped. Before the worker registers the request, +the controller polls every two seconds, since nothing would report that the worker exists yet. The +worker keeps that attachment even when local materialization is deferred, so the re-send is +recognized as the same request: it cannot register afresh and expire into `NoWindow` while its commit waits out the push cooldown, and it cannot claim the next same-author window and stamp its message on somebody else's commit. Its attach deadline is spent and no longer arms anything. diff --git a/internal/controller/commitrequest_apply.go b/internal/controller/commitrequest_apply.go index 6c919749..09a4c721 100644 --- a/internal/controller/commitrequest_apply.go +++ b/internal/controller/commitrequest_apply.go @@ -59,12 +59,12 @@ func (c *commitRequestCore) followApply( ctx context.Context, p commitRequestPath, cr *configbutleraiv1alpha3.CommitRequest, ) (time.Duration, error) { apply := nothingToApply() + known := false if cr.Status.Commit != "" { attach := git.AttachCommitRequest{ Namespace: cr.Namespace, Name: cr.Name, UID: string(cr.UID), GitTargetName: cr.GitTargetKey().Name, GitTargetNamespace: cr.GitTargetKey().Namespace, } - var known bool var err error apply, known, err = c.finalizer.LookupCommitRequestApply(ctx, attach) if err != nil { @@ -94,6 +94,9 @@ func (c *commitRequestCore) followApply( if applyReported(cr) { return c.expireFinished(ctx, p, cr) } + if known { + return c.registeredPoll(), nil // the worker reports each move of the apply + } return commitRequestPollInterval, nil } diff --git a/internal/controller/commitrequest_controller.go b/internal/controller/commitrequest_controller.go index a7676365..00f2fee7 100644 --- a/internal/controller/commitrequest_controller.go +++ b/internal/controller/commitrequest_controller.go @@ -12,7 +12,9 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/handler" logf "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/source" configbutleraiv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" "github.com/ConfigButler/gitops-reverser/internal/git" @@ -163,6 +165,10 @@ type CommitRequestReconciler struct { // CommitRequestDeleteAfterAnnotation at that moment; zero writes nothing, so the request stays. TTL time.Duration + // Wakes reconciles a request as soon as its branch worker reports a change, so a registered + // request is polled only as the fallback. Nil polls every request. + Wakes *CommitRequestWakes + streamsOnce sync.Once streams *commitRequestStreams } @@ -187,6 +193,7 @@ func (r *CommitRequestReconciler) core() *commitRequestCore { r.streamsOnce.Do(func() { r.streams = &commitRequestStreams{set: r.FactStreams} }) return &commitRequestCore{ reader: r.Client, finalizer: r.Finalizer, authors: r.Authors, streams: r.streams, ttl: r.TTL, + wakeFallback: r.Wakes.fallback(), } } @@ -245,11 +252,14 @@ func (s typedCommitRequestStore) remove(ctx context.Context, cr *configbutleraiv // a step can wait out the attribution grace for its request's author, so they must, or one request's // wait would hold up every other. func (r *CommitRequestReconciler) SetupWithManager(mgr ctrl.Manager) error { - return ctrl.NewControllerManagedBy(mgr). + b := ctrl.NewControllerManagedBy(mgr). For(&configbutleraiv1alpha3.CommitRequest{}). WithOptions(controller.Options{MaxConcurrentReconciles: commitRequestConcurrency}). - Named("commitrequest"). - Complete(r) + Named("commitrequest") + if events := r.Wakes.Events(); events != nil { + b = b.WatchesRawSource(source.Channel(events, &handler.EnqueueRequestForObject{})) + } + return b.Complete(r) } // commitRequestConcurrency is how many CommitRequest steps run at once. diff --git a/internal/controller/commitrequest_core.go b/internal/controller/commitrequest_core.go index d6bd7741..76841ed5 100644 --- a/internal/controller/commitrequest_core.go +++ b/internal/controller/commitrequest_core.go @@ -96,8 +96,14 @@ func (p commitRequestPath) gitTargetKey(cr *configbutleraiv1alpha3.CommitRequest // commitRequestCore serves a CommitRequest for either path, one step at a time. type commitRequestCore struct { // reader reads GitTargets, KubeProviders and Namespaces on the control cluster. - reader client.Reader - finalizer CommitRequestFinalizer + reader client.Reader + // controlRequests reads CommitRequests on the control cluster, uncached, for a source-cluster + // path to tell its own cluster from the control cluster (sourceRefusal); nil on the control path. + controlRequests client.Reader + // wakeFallback is how often a request its worker registered is looked at, when the worker's + // changes wake its steps (CommitRequestWakes); zero polls it at commitRequestPollInterval. + wakeFallback time.Duration + finalizer CommitRequestFinalizer // authors finds a request's creator from audit facts; nil when author attribution is off. authors watch.AuthorResolver // streams holds the fact stream of commitrequests on each audit route a creator is looked up on. @@ -152,7 +158,7 @@ func (c *commitRequestCore) serveOpen( } return c.awaitGitTarget(ctx, p, cr, targetKey) } - provider, owner, refusal, err := c.owns(ctx, p, target) + provider, owner, refusal, err := c.owns(ctx, p, cr, target) if err != nil { return 0, err } @@ -187,9 +193,10 @@ func (c *commitRequestCore) gitTarget( // path. On the control cluster that is a provider with inCluster. A provider with a kubeconfig may point // back at the control cluster, and then its own watch serves the request, so it is left alone, unless // that provider serves no CommitRequests: then refusal says why, and where to create the request -// instead. A provider that cannot be found picks no path yet. +// instead. A provider that cannot be found picks no path yet. On a source cluster, see sourceRefusal. func (c *commitRequestCore) owns( - ctx context.Context, p commitRequestPath, target *configbutleraiv1alpha3.GitTarget, + ctx context.Context, p commitRequestPath, cr *configbutleraiv1alpha3.CommitRequest, + target *configbutleraiv1alpha3.GitTarget, ) (*configbutleraiv1alpha3.KubeProvider, bool, string, error) { key := target.KubeProviderKey() var provider configbutleraiv1alpha3.KubeProvider @@ -200,7 +207,11 @@ func (c *commitRequestCore) owns( return nil, false, "", fmt.Errorf("read KubeProvider %s: %w", key, err) } if !p.controlCluster() { - return &provider, key == p.provider, "", nil + if key == p.provider { + return &provider, true, "", nil + } + refusal, err := c.sourceRefusal(ctx, p, cr, target, &provider) + return &provider, false, refusal, err } if provider.IsInCluster() { return &provider, true, "", nil @@ -213,6 +224,48 @@ func (c *commitRequestCore) owns( client.ObjectKeyFromObject(target), key, key), nil } +// sourceRefusal says why no path can serve a request on the source cluster of path p, which names a +// GitTarget whose KubeProvider is another: empty while some path may still serve it. +// +// - The GitTarget mirrors the cluster GitOps Reverser runs in. The control-cluster path serves it +// only when p's kubeconfig points back at that cluster, and then the request is the same object +// there, with the same UID. Read uncached, so a request created a moment ago is not mistaken for +// one that does not exist. Without the reader nothing is refused. +// - The GitTarget's KubeProvider serves no CommitRequests, so no watch picks the request up. +// +// A KubeProvider that does serve them may connect to this same cluster under another identity, and +// then its watch serves the request: it is left alone. +func (c *commitRequestCore) sourceRefusal( + ctx context.Context, p commitRequestPath, cr *configbutleraiv1alpha3.CommitRequest, + target *configbutleraiv1alpha3.GitTarget, provider *configbutleraiv1alpha3.KubeProvider, +) (string, error) { + here := fmt.Sprintf("the cluster KubeProvider %s connects to", p.provider) + targetKey := client.ObjectKeyFromObject(target) + if provider.IsInCluster() { + if c.controlRequests == nil { + return "", nil + } + var local configbutleraiv1alpha3.CommitRequest + err := c.controlRequests.Get(ctx, client.ObjectKeyFromObject(cr), &local) + switch { + case err == nil && local.UID == cr.UID: + return "", nil // the same object: this cluster is the control cluster, whose path serves it + case err != nil && !apierrors.IsNotFound(err): + return "", fmt.Errorf("read CommitRequest %s on the cluster GitOps Reverser runs in: %w", + client.ObjectKeyFromObject(cr), err) + } + return fmt.Sprintf("this CommitRequest is on %s, and GitTarget %s mirrors the cluster GitOps Reverser "+ + "runs in, through KubeProvider %s; create the CommitRequest on the cluster GitOps Reverser runs in", + here, targetKey, provider.Key()), nil + } + if !provider.ServesCommitRequests() { + return fmt.Sprintf("this CommitRequest is on %s, and GitTarget %s mirrors the cluster KubeProvider %s "+ + "connects to, which serves no CommitRequests; create the CommitRequest on that cluster, and set "+ + "spec.commitRequests: Reconcile on KubeProvider %s", here, targetKey, provider.Key(), provider.Key()), nil + } + return "", nil +} + // awaitGitTarget waits attachTimeout from the request's creation for the GitTarget it names, which may // be created alongside it, and then reports GitTargetNotFound. That outcome is true for every observer, // so any path may write it, but only into an empty status: a path that claimed the request in the @@ -522,14 +575,34 @@ func (c *commitRequestCore) recordPhase( ctx context.Context, p commitRequestPath, cr *configbutleraiv1alpha3.CommitRequest, phase git.CommitRequestPhase, held string, ) (time.Duration, error) { + poll := commitRequestPollInterval + if registered(phase) { + poll = c.registeredPoll() + } if progressRecorded(cr, phase, held) { - return commitRequestPollInterval, nil + return poll, nil } markCommitRequestProgressing(cr, phase, held) if err := p.store.updateStatus(ctx, cr); err != nil { - return commitRequestPollInterval, ignoreConflict(err) + return poll, ignoreConflict(err) + } + return poll, nil +} + +// registered reports whether the worker has registered a request in this phase, so it reports the +// request's every change from now on. Before that, while there is no worker or the attach is still +// on its way, nothing would wake the step. +func registered(phase git.CommitRequestPhase) bool { + return phase == git.PhaseWaitingForWindow || phase.Held() +} + +// registeredPoll is how soon a step looks again at a request its worker has registered: the +// fallback for a dropped change when the worker's changes wake the steps, the poll otherwise. +func (c *commitRequestCore) registeredPoll() time.Duration { + if c.wakeFallback > 0 { + return c.wakeFallback } - return commitRequestPollInterval, nil + return commitRequestPollInterval } // refusePrunedGitTargetRef fails a request whose spec names no GitTarget, which can only be one diff --git a/internal/controller/commitrequest_core_test.go b/internal/controller/commitrequest_core_test.go index 138eb362..2660b872 100644 --- a/internal/controller/commitrequest_core_test.go +++ b/internal/controller/commitrequest_core_test.go @@ -628,3 +628,82 @@ func TestCommitRequestCore_AWatcherCarriesNextAndCommitEmpty(t *testing.T) { assert.Equal(t, configv1alpha3.AttachNext, f.calls[0].Attach) assert.True(t, f.calls[0].CommitEmpty) } + +// A request on a source cluster that names a GitTarget no path can serve from there ends NotAllowed, +// naming both clusters, instead of waiting with no status: gitops-api ask #38. It is never attached. +// A kubeconfig that points back at the control cluster sees the control cluster's own request, the +// same object, and leaves it to the control-cluster path; so does a request whose GitTarget's +// KubeProvider serves CommitRequests, since it may watch this same cluster. +func TestCommitRequestCore_ASourceRequestForAGitTargetOfAnotherCluster(t *testing.T) { + for _, tc := range []struct { + name string + target string + providers []client.Object + pointsBack bool + wantIn []string + }{ + {name: "the GitTarget mirrors the control cluster", target: localProvider, + providers: []client.Object{kubeconfigProvider("edge", true)}, + wantIn: []string{"the cluster KubeProvider default/edge connects to", + "mirrors the cluster GitOps Reverser runs in"}}, + {name: "the GitTarget's KubeProvider serves no CommitRequests", target: "other", + providers: []client.Object{kubeconfigProvider("edge", true), kubeconfigProvider("other", false)}, + wantIn: []string{"the cluster KubeProvider default/edge connects to", + "the cluster KubeProvider default/other connects to, which serves no CommitRequests"}}, + {name: "a kubeconfig that points back at the control cluster", target: localProvider, + providers: []client.Object{kubeconfigProvider("edge", true)}, pointsBack: true}, + {name: "the GitTarget's KubeProvider serves CommitRequests", target: "other", + providers: []client.Object{kubeconfigProvider("edge", true), kubeconfigProvider("other", true)}}, + } { + t.Run(tc.name, func(t *testing.T) { + cr := newCommitRequest("save") + cr.Spec.GitTargetRef.Namespace = "default" + objects := append([]client.Object{mirroringTarget(tc.target)}, tc.providers...) + if tc.pointsBack { + objects = append(objects, cr.DeepCopy()) + } + // The control cluster, and a source cluster of its own, whose requests are not on it. + control := newCommitRequestClient(t, nil, objects...) + src := newCommitRequestClient(t, nil, cr) + f := committed() + core := &commitRequestCore{reader: control, controlRequests: control, finalizer: f} + + after := serveStep(t, core, watcherPath(src, "edge", false), "save") + + assert.Empty(t, f.calls, "never attached") + ready := commitRequestReady(t, src, "save") + if len(tc.wantIn) == 0 { + assert.Nil(t, ready, "left alone") + assert.Equal(t, commitRequestUnownedRecheck, after) + return + } + require.NotNil(t, ready) + assert.Equal(t, metav1.ConditionFalse, ready.Status) + assert.Equal(t, crReasonNotAllowed, ready.Reason) + for _, want := range tc.wantIn { + assert.Contains(t, ready.Message, want) + } + }) + } +} + +// A refused source request finishes, so a watcher that deletes finished requests stamps and then +// deletes it: a tenant without delete on the type no longer keeps it for ever. +func TestCommitRequestCore_ARefusedSourceRequestIsCollected(t *testing.T) { + cr := newCommitRequest("save") + cr.Spec.GitTargetRef.Namespace = "default" + control := newCommitRequestClient(t, nil, mirroringTarget(localProvider), kubeconfigProvider("edge", true)) + src := newCommitRequestClient(t, nil, cr) + core := &commitRequestCore{reader: control, controlRequests: control, finalizer: committed(), ttl: time.Hour} + p := watcherPath(src, "edge", true) + + serveStep(t, core, p, "save") + got := fetchCommitRequest(t, src, "save") + require.True(t, commitRequestIsTerminal(&got)) + require.Contains(t, got.Annotations, CommitRequestDeleteAfterAnnotation, "stamped when it was refused") + + got.Annotations[CommitRequestDeleteAfterAnnotation] = time.Now().Add(-time.Minute).UTC().Format(time.RFC3339) + require.NoError(t, src.Update(context.Background(), &got)) + serveStep(t, core, p, "save") + assert.False(t, commitRequestExists(t, src, "save"), "deleted once its time passed") +} diff --git a/internal/controller/commitrequest_decide.go b/internal/controller/commitrequest_decide.go index cf07c63a..2600a4cf 100644 --- a/internal/controller/commitrequest_decide.go +++ b/internal/controller/commitrequest_decide.go @@ -185,7 +185,7 @@ func (c *commitRequestCore) keeps( if err != nil || !found { return !found, err } - _, owner, refusal, err := c.owns(ctx, p, target) + _, owner, refusal, err := c.owns(ctx, p, cr, target) return owner || refusal != "", err } diff --git a/internal/controller/commitrequest_server.go b/internal/controller/commitrequest_server.go index e01cdfde..4d13e2da 100644 --- a/internal/controller/commitrequest_server.go +++ b/internal/controller/commitrequest_server.go @@ -92,7 +92,10 @@ type CommitRequestsReport struct { // the leader only, as every status writer does. type CommitRequestServer struct { // Client reads KubeProviders, GitTargets and Namespaces on the control cluster. - Client client.Client + Client client.Client + // APIReader reads CommitRequests on the control cluster, uncached, to tell a source cluster from a + // kubeconfig that points back at the control cluster. Nil refuses no request for that reason. + APIReader client.Reader Clusters CommitRequestClusters Finalizer CommitRequestFinalizer // Authors finds a request's creator from the source cluster's audit facts; nil with attribution @@ -102,7 +105,10 @@ type CommitRequestServer struct { FactStreams *queue.FactStreamSet // TTL is how long a finished request is kept on a provider that deletes them. TTL time.Duration - Log logr.Logger + // Wakes wakes a request's steps as soon as its branch worker reports a change, so a registered + // request is polled only as the fallback. Nil polls every request. + Wakes *CommitRequestWakes + Log logr.Logger mu sync.Mutex watchers map[k8stypes.NamespacedName]*commitRequestWatcher @@ -569,6 +575,8 @@ func (s *CommitRequestServer) consider( core := s.core() go func() { defer watcher.serving.Delete(u.GetUID()) + woken, stop := s.Wakes.watch(u.GetUID()) + defer stop() for ctx.Err() == nil { after, err := core.serve(ctx, watcher.path, key) if err != nil { @@ -582,6 +590,7 @@ func (s *CommitRequestServer) consider( select { case <-ctx.Done(): case <-time.After(after): + case <-woken: } } }() @@ -590,7 +599,8 @@ func (s *CommitRequestServer) consider( func (s *CommitRequestServer) core() *commitRequestCore { s.streamsOnce.Do(func() { s.streams = &commitRequestStreams{set: s.FactStreams} }) return &commitRequestCore{ - reader: s.Client, finalizer: s.Finalizer, authors: s.Authors, streams: s.streams, ttl: s.TTL, + reader: s.Client, controlRequests: s.APIReader, finalizer: s.Finalizer, authors: s.Authors, + streams: s.streams, ttl: s.TTL, wakeFallback: s.Wakes.fallback(), } } diff --git a/internal/controller/commitrequest_wakes.go b/internal/controller/commitrequest_wakes.go new file mode 100644 index 00000000..e05e5563 --- /dev/null +++ b/internal/controller/commitrequest_wakes.go @@ -0,0 +1,125 @@ +// SPDX-License-Identifier: Apache-2.0 + +package controller + +import ( + "context" + "sync" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/event" + + configbutleraiv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + "github.com/ConfigButler/gitops-reverser/internal/git" +) + +// commitRequestWakeFallback is how often a request the worker has registered is looked at when its +// steps are woken by the worker's changes: only a change that was dropped waits this long. +const commitRequestWakeFallback = 10 * time.Second + +// commitRequestWakeBuffer bounds the reconciles a burst of changes can queue; a full buffer drops the +// arriving one, and the fallback poll takes its step. +const commitRequestWakeBuffer = 256 + +// CommitRequestWakes turns each change a branch worker reports about a CommitRequest into the +// request's next step, at once instead of at its next poll: a reconcile for a request on the control +// cluster, and a wake of the goroutine that serves a request on a source cluster. Nil wakes nothing, +// and the steps poll. +type CommitRequestWakes struct { + changes <-chan git.CommitRequestChange + events chan event.GenericEvent + + mu sync.Mutex + // served holds, by UID, the wake of every goroutine that serves a source-cluster request now. A + // request can have several: each KubeProvider watching its namespace on that cluster looks at it, + // and one that does not own it must not take the owner's wake away. + served map[types.UID]map[chan struct{}]struct{} +} + +// NewCommitRequestWakes reads changes; nil changes make wakes that wake nothing. +func NewCommitRequestWakes(changes <-chan git.CommitRequestChange) *CommitRequestWakes { + return &CommitRequestWakes{ + changes: changes, + events: make(chan event.GenericEvent, commitRequestWakeBuffer), + served: map[types.UID]map[chan struct{}]struct{}{}, + } +} + +// Start hands each change on until ctx ends. +func (w *CommitRequestWakes) Start(ctx context.Context) error { + for { + select { + case <-ctx.Done(): + return nil + case change, ok := <-w.changes: + if !ok { + return nil + } + w.wake(change) + } + } +} + +// wake queues a reconcile of the request on the control cluster, and wakes every goroutine serving +// it on a source cluster. A change names one request, on whichever cluster it lives, and +// a reconcile of a name the control cluster does not hold reads nothing and ends. +func (w *CommitRequestWakes) wake(change git.CommitRequestChange) { + request := &configbutleraiv1alpha3.CommitRequest{ + ObjectMeta: metav1.ObjectMeta{Namespace: change.Namespace, Name: change.Name}, + } + select { + case w.events <- event.GenericEvent{Object: request}: + default: + } + w.mu.Lock() + defer w.mu.Unlock() + for ch := range w.served[types.UID(change.UID)] { + select { + case ch <- struct{}{}: + default: + } + } +} + +// Events carries the control-cluster requests to reconcile. Nil on nil wakes. +func (w *CommitRequestWakes) Events() <-chan event.GenericEvent { + if w == nil { + return nil + } + return w.events +} + +// watch returns the wake of the request uid while a goroutine serves it, and the func that stops it. +// A change while the goroutine is mid-step is kept for its next wait. Nil wakes return a channel that +// never fires. +func (w *CommitRequestWakes) watch(uid types.UID) (<-chan struct{}, func()) { + if w == nil || uid == "" { + return nil, func() {} + } + ch := make(chan struct{}, 1) + w.mu.Lock() + if w.served[uid] == nil { + w.served[uid] = map[chan struct{}]struct{}{} + } + w.served[uid][ch] = struct{}{} + w.mu.Unlock() + return ch, func() { + w.mu.Lock() + delete(w.served[uid], ch) + if len(w.served[uid]) == 0 { + delete(w.served, uid) + } + w.mu.Unlock() + } +} + +// fallback is how often a step looks at a request its worker registered: the slow fallback when +// the worker's changes wake the steps, zero (the plain poll) when nothing does. +func (w *CommitRequestWakes) fallback() time.Duration { + if w == nil || w.changes == nil { + return 0 + } + return commitRequestWakeFallback +} diff --git a/internal/controller/commitrequest_wakes_test.go b/internal/controller/commitrequest_wakes_test.go new file mode 100644 index 00000000..10364841 --- /dev/null +++ b/internal/controller/commitrequest_wakes_test.go @@ -0,0 +1,230 @@ +// SPDX-License-Identifier: Apache-2.0 + +package controller + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + "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. +type simulatedWorker struct { + registerAfter, resolveAfter time.Duration + changes chan git.CommitRequestChange + + mu sync.Mutex + started bool + phase git.CommitRequestPhase + resolved bool + pushedAt time.Time +} + +func (w *simulatedWorker) ServiceCommitRequest( + _ context.Context, attach 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 + }) + } + if w.resolved { + return git.FinalizeResult{Outcome: git.FinalizeCommitted, Commit: "c0ffee", Branch: "main"}, true, nil + } + return git.FinalizeResult{Phase: w.phase}, false, nil +} + +func (w *simulatedWorker) WithdrawCommitRequest( + ctx context.Context, attach git.AttachCommitRequest, +) (git.FinalizeResult, bool, error) { + return w.ServiceCommitRequest(ctx, attach) +} + +func (w *simulatedWorker) LookupCommitRequestApply( + context.Context, git.AttachCommitRequest, +) (git.SaveApply, bool, error) { + 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 } + +// 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 { + 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, + changes: make(chan git.CommitRequestChange, 4), + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + server := &CommitRequestServer{Client: c, Finalizer: worker} + 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), + } + u := &unstructured.Unstructured{} + 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 +} + +// 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. +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") +} + +// TestCommitRequestWakes_AChangeMidStepIsKeptForTheNextWait pins that a change reported while the +// request's goroutine is between waits is not lost: its next wait ends at once. +func TestCommitRequestWakes_AChangeMidStepIsKeptForTheNextWait(t *testing.T) { + changes := make(chan git.CommitRequestChange, 1) + wakes := NewCommitRequestWakes(changes) + woken, stop := wakes.watch("uid-save") + defer stop() + + wakes.wake(git.CommitRequestChange{Namespace: "default", Name: "save", UID: "uid-save"}) + select { + case <-woken: + default: + t.Fatal("the change was lost") + } + select { + case ev := <-wakes.Events(): + assert.Equal(t, "save", ev.Object.GetName(), "the control-cluster request is reconciled too") + default: + t.Fatal("no reconcile was queued") + } + + stop() + wakes.wake(git.CommitRequestChange{Namespace: "default", Name: "save", UID: "uid-save"}) + select { + case <-woken: + t.Fatal("a request no goroutine serves any more is not woken") + default: + } +} + +// TestCommitRequestWakes_AnotherWatcherLeavesTheOwnersWake pins that two KubeProviders watching +// the same source cluster each keep their own wake of a request. The one that does not own it looks +// at it after every write the owner makes, and leaves at once; neither its watch nor its leaving may +// take the owner's wake away, or the owner falls back to the slow poll. +func TestCommitRequestWakes_AnotherWatcherLeavesTheOwnersWake(t *testing.T) { + wakes := NewCommitRequestWakes(make(chan git.CommitRequestChange)) + owner, stopOwner := wakes.watch("uid-save") + defer stopOwner() + change := git.CommitRequestChange{Namespace: "default", Name: "save", UID: "uid-save"} + + for range 2 { + other, stopOther := wakes.watch("uid-save") + wakes.wake(change) + select { + case <-other: + default: + t.Fatal("the other watcher was not woken") + } + stopOther() + stopOther() + select { + case <-owner: + default: + t.Fatal("the owner was not woken while another watcher looked at its request") + } + } + + wakes.wake(change) + select { + case <-owner: + default: + t.Fatal("the owner was not woken after the other watcher left") + } + stopOwner() + wakes.mu.Lock() + assert.Empty(t, wakes.served, "a request nobody serves leaves nothing behind") + wakes.mu.Unlock() +} + +// TestCommitRequestWakes_NilWakesPoll pins that without wakes every step polls as it always did. +func TestCommitRequestWakes_NilWakesPoll(t *testing.T) { + var wakes *CommitRequestWakes + assert.Nil(t, wakes.Events()) + assert.Zero(t, wakes.fallback()) + woken, stop := wakes.watch("uid") + stop() + assert.Nil(t, woken) +} diff --git a/internal/git/applier_echoes_test.go b/internal/git/applier_echoes_test.go index aa2c9cd8..c5a21016 100644 --- a/internal/git/applier_echoes_test.go +++ b/internal/git/applier_echoes_test.go @@ -143,7 +143,7 @@ func (f *echoFixture) writeWindow(events ...Event) { func TestApplierEchoes_AnUnchangedObjectIsNoAction(t *testing.T) { t.Run("alone, it is never retained", func(t *testing.T) { f := newEchoFixture(t, "echo-unchanged-alone", configv1alpha3.PruneOnEvent, map[string]string{"x": "x1"}) - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) f.writeWindow(echoEvent("x", "x1")) // the applier's echo of what Git holds require.Empty(t, f.loop.pendingWrites, "a write that changes nothing settles at once") @@ -158,7 +158,7 @@ func TestApplierEchoes_AnUnchangedObjectIsNoAction(t *testing.T) { t.Run("beside a changed object, only the changed one is committed", func(t *testing.T) { f := newEchoFixture(t, "echo-unchanged-batch", configv1alpha3.PruneOnEvent, map[string]string{"x": "x1", "y": "y1"}) - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) f.writeWindow(echoEvent("x", "x1"), echoEvent("y", "y2")) require.Len(t, f.loop.pendingWrites, 1) @@ -176,7 +176,7 @@ func TestApplierEchoes_AnUnchangedObjectIsNoAction(t *testing.T) { t.Run("a save keeps every object it captured", func(t *testing.T) { f := newEchoFixture(t, "echo-unchanged-save", configv1alpha3.PruneOnEvent, map[string]string{"x": "x1"}) - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) save := attachReq("alice", time.Hour) save.GitTargetName = echoTarget save.MaxDuration = 0 @@ -203,7 +203,7 @@ func TestApplierEchoes_AnUnchangedObjectIsNoAction(t *testing.T) { func TestApplierEchoes_AResyncReplayAppliesItsDecidedActions(t *testing.T) { f := newEchoFixture(t, "echo-resync-replay", configv1alpha3.PruneAlways, map[string]string{"x": "x1", "y": "y1", "gone": "g1"}) - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) req := &ResyncRequest{ Desired: []manifestanalyzer.DesiredResource{echoDesired("x", "x1"), echoDesired("y", "y2")}, @@ -243,7 +243,7 @@ func TestApplierEchoes_AResyncReplayAppliesItsDecidedActions(t *testing.T) { func TestApplierEchoes_AReplayedResyncObeysATightenedPruneMode(t *testing.T) { f := newEchoFixture(t, "echo-resync-tightened", configv1alpha3.PruneAlways, map[string]string{"y": "y1", "gone": "g1"}) - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) req := &ResyncRequest{ Desired: []manifestanalyzer.DesiredResource{echoDesired("y", "y2")}, @@ -314,7 +314,7 @@ func (f *echoFixture) echoApplier() *applierState { func TestApplierEchoes_FlushThenApply(t *testing.T) { f := newEchoFixture(t, "echo-flush", configv1alpha3.PruneOnEvent, map[string]string{"x": "x1", "y": "y1"}) requester := f.withClaim() - f.loop.lastPushAt = time.Now() // the push cooldown runs: a flush does not wait for it + holdPushes(f.loop) // the push cooldown runs: a flush does not wait for it f.writeWindow(echoEvent("x", "x2")) // E0, in the open window when the apply comes due require.NotNil(t, f.loop.openWindow) @@ -375,7 +375,7 @@ func TestApplierEchoes_AnApplierThatJoinsAFlushHoldsTheEditsSinceItsMark(t *test t.Fatalf("no applier covers %s", target) return nil } - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) f.writeWindow(echoEvent("x", "x2")) restore := failPushes(t, f.worker) diff --git a/internal/git/apply_window_carry_test.go b/internal/git/apply_window_carry_test.go index fa02370d..bd024da7 100644 --- a/internal/git/apply_window_carry_test.go +++ b/internal/git/apply_window_carry_test.go @@ -49,7 +49,7 @@ func TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit(t *testing.T) f := newEchoFixture(t, slug, configv1alpha3.PruneOnEvent, map[string]string{"x": "x1"}) f.writeWindow(edit("a1", "b1")) // y as Git holds it before anyone edits requester := f.withClaim() - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) f.writeWindow(echoEvent("x", "x2")) restore := failPushes(t, f.worker) diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index f52d0959..65c73fb7 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -399,6 +399,10 @@ type BranchWorker struct { // publicationReporter is told when the worker's publication report changes. Set by the // WorkerManager before Start; nil in tests that do not assert on it. See publication.go. publicationReporter func() + // commitRequestReporter is told when a CommitRequest's standing on this worker changes. Set by + // the WorkerManager before Start; nil in tests that do not assert on it. See + // commit_request_changes.go. + commitRequestReporter func(CommitRequestChange) // publication is the loop's latest publication report, and backlog what it keeps, for readers // outside the loop. See publication.go. publication atomic.Pointer[PublicationStatus] @@ -1030,8 +1034,11 @@ type branchWorkerEventLoop struct { pendingWrites []PendingWrite pendingWritesBytes int64 - lastPushAt time.Time - commitTimer *time.Timer + lastPushAt time.Time + // cooldownHoldsSaves keeps a save's push behind the cooldown, like any other push. A test seam, + // for tests that inspect a save's write before it is published; never set otherwise. + cooldownHoldsSaves bool + commitTimer *time.Timer // commitTimerAt is the worker-clock time commitTimer is armed for, while it is armed. The timer // itself runs on real time; tests that step the clock deliver its wake at this time. commitTimerAt time.Time @@ -1643,6 +1650,14 @@ func (l *branchWorkerEventLoop) maybeSchedulePush() { l.pushPending() return } + if !l.cooldownHoldsSaves && l.pendingCarriesCommitRequest() { + // A save waits for this publication, and a person waits for the save: it is made at once too. + // Saves cannot burst the way plain writes can, since an author has one save in progress at a + // time, so the cooldown still paces every stream of plain writes. + l.w.Log.V(1).Info("Pushing at once: a CommitRequest waits for this push") + l.pushPending() + return + } elapsed := time.Since(l.lastPushAt) if elapsed >= PushCooldown { l.pushPending() @@ -1653,6 +1668,17 @@ func (l *branchWorkerEventLoop) maybeSchedulePush() { } } +// pendingCarriesCommitRequest reports whether a pending write carries a CommitRequest, which resolves +// only once the write is pushed. +func (l *branchWorkerEventLoop) pendingCarriesCommitRequest() bool { + for i := range l.pendingWrites { + if l.pendingWrites[i].CommitRequest != nil { + return true + } + } + return false +} + // pushPending materializes the pending writes and publishes them. On success, pendingWrites is cleared, // lastPushAt advances, and the outage, if there was one, is over. On failure (transient or after // exhausting replay retries), pendingWrites stays in place, safe but unpublished, and the next diff --git a/internal/git/branch_worker_empty_repo_test.go b/internal/git/branch_worker_empty_repo_test.go index 3795235d..b94dff43 100644 --- a/internal/git/branch_worker_empty_repo_test.go +++ b/internal/git/branch_worker_empty_repo_test.go @@ -43,9 +43,12 @@ func newRealServerNewBranch(t *testing.T, slug string, seed func(repoDir string) return &realServerNewBranch{ledgerFixture: f, loop: loop} } +// gitIn runs git in dir. A fixture's own commits are never signed, whatever the developer's global +// commit.gpgSign says: CI signs nothing, and a signing agent that went away must not fail a test. func gitIn(t *testing.T, dir string, args ...string) string { t.Helper() - out, err := exec.Command("git", append([]string{"-C", dir}, args...)...).CombinedOutput() + unsigned := []string{"-C", dir, "-c", "commit.gpgSign=false"} + out, err := exec.Command("git", append(unsigned, args...)...).CombinedOutput() require.NoError(t, err, "git %v: %s", args, out) return strings.TrimSpace(string(out)) } @@ -122,7 +125,7 @@ func TestBranchWorker_AnEmptyRepositoryThatGainsABranchIsNotOrphaned(t *testing. require.NoError(t, err) f := newRealServerNewBranch(t, "empty-gains-main", nil) - f.loop.lastPushAt = time.Now() // cooldown: the write is retained + holdPushes(f.loop) // cooldown: the write is retained req := standbyRequest() serviceAttach(f.loop, req) liveWrite(f.loop, "cm1") @@ -148,7 +151,7 @@ func TestBranchWorker_AnEmptyRepositoryThatGainsABranchIsNotOrphaned(t *testing. func TestBranchWorker_AnEmptyRepositoryThatGainsABranchKeepsANoOpAbsent(t *testing.T) { f := newRealServerNewBranch(t, "empty-gains-main-noop", nil) - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) deleted := configMapTargetEvent("never-written", "alice", newBranchTarget) deleted.Operation = "DELETE" deleted.Object = nil @@ -167,7 +170,7 @@ func TestBranchWorker_AnEmptyRepositoryThatGainsABranchKeepsANoOpAbsent(t *testi func TestBranchWorker_AnEmptyRepositoryThatGainsOnlyATagWaits(t *testing.T) { f := newRealServerNewBranch(t, "empty-gains-tag", nil) - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) liveWrite(f.loop, "cm1") require.Len(t, f.loop.pendingWrites, 1) diff --git a/internal/git/branch_worker_fetch_metric_test.go b/internal/git/branch_worker_fetch_metric_test.go index f564e201..deb05df4 100644 --- a/internal/git/branch_worker_fetch_metric_test.go +++ b/internal/git/branch_worker_fetch_metric_test.go @@ -284,7 +284,7 @@ func TestGitFetchesTotal_ResyncWithRetainedWritesStillReadsTheRemote(t *testing. // A live write is committed and retained: the push cooldown has not elapsed, so it is sitting // in pendingWrites when the resync arrives. loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ Events: []Event{configMapEvent("retained", "alice", "live")}, @@ -326,7 +326,7 @@ func TestGitFetchesTotal_RecoveryIsOneSeriesWhetherOrNotWritesAreRetained(t *tes f.publish("prime") loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() // Retained work, so the loop's replay path is the one that recovers. diff --git a/internal/git/branch_worker_loop_test.go b/internal/git/branch_worker_loop_test.go index e7b5346e..4a06f792 100644 --- a/internal/git/branch_worker_loop_test.go +++ b/internal/git/branch_worker_loop_test.go @@ -97,7 +97,7 @@ func TestEventLoop_MaybeSchedulePush_CooldownGate(t *testing.T) { // pushTimer rather than push immediately. loop.pendingWrites = []PendingWrite{{Kind: PendingWriteCommit}} loop.pendingWritesBytes = 1 - loop.lastPushAt = time.Now() // pretend we just pushed + holdPushes(loop) // pretend we just pushed loop.maybeSchedulePush() require.NotNil(t, loop.pushTimer, "cooldown active → pushTimer scheduled") diff --git a/internal/git/branch_worker_new_branch_test.go b/internal/git/branch_worker_new_branch_test.go index 1fe970f0..dae1f33c 100644 --- a/internal/git/branch_worker_new_branch_test.go +++ b/internal/git/branch_worker_new_branch_test.go @@ -243,7 +243,7 @@ func TestBranchWorker_LiveWriteOnAbsentBranchStartsFromTheParentsCurrentTip(t *t hashB := f.pushToMain(map[string]string{"LATEST.md": "from main at B\n"}) // Hold the push so the cycle's recorded root can be read before it is sent. - loop.lastPushAt = time.Now() + holdPushes(loop) liveWrite(loop, "cm1") require.Len(t, loop.pendingWrites, 1) assert.Equal(t, plumbing.NewBranchReferenceName("main"), f.worker.pushCycleRootBranch, @@ -363,7 +363,7 @@ func TestBranchWorker_NoDiffLiveWriteOnAbsentBranchStaysOnStandby(t *testing.T) defer loop.stopTimers() f.warmOnParent() - loop.lastPushAt = time.Now() + holdPushes(loop) liveWrite(loop, "cm1") require.Len(t, loop.pendingWrites, 1) require.False(t, loop.pendingWrites[0].CommitSHA.IsZero(), "planned against A, the write is a commit") @@ -480,7 +480,7 @@ func TestBranchWorker_NewBranchIsNotPublishedOnAnUncheckedParent(t *testing.T) { return report, err } - loop.lastPushAt = time.Now() + holdPushes(loop) liveWrite(loop, "cm1") f.pushToMain(map[string]string{"FIRST.md": "moved before the first attempt\n"}) loop.pushPending() @@ -522,7 +522,7 @@ func TestBranchWorker_WriteBranchCreatedConcurrentlyIsBuiltOn(t *testing.T) { defer loop.stopTimers() f.warmOnParent() - loop.lastPushAt = time.Now() + holdPushes(loop) liveWrite(loop, "cm1") require.Len(t, loop.pendingWrites, 1) diff --git a/internal/git/branch_worker_parent_branch_test.go b/internal/git/branch_worker_parent_branch_test.go index e7178bf8..99e12929 100644 --- a/internal/git/branch_worker_parent_branch_test.go +++ b/internal/git/branch_worker_parent_branch_test.go @@ -248,7 +248,7 @@ func TestBranchWorker_AParentChangeRebuildsRetainedWrites(t *testing.T) { loop := newBranchWorkerEventLoop(f.worker, 0) defer loop.stopTimers() - loop.lastPushAt = time.Now() + holdPushes(loop) noOpSave(loop) require.Len(t, loop.pendingWrites, 1) require.True(t, loop.pendingWrites[0].CommitSHA.IsZero(), "a no-op save is retained during the cooldown") @@ -283,7 +283,7 @@ func TestBranchWorker_AParentChangeSurvivesAFailedRebuild(t *testing.T) { loop := newBranchWorkerEventLoop(f.worker, 0) defer loop.stopTimers() - loop.lastPushAt = time.Now() + holdPushes(loop) liveWrite(loop, "cm1") require.Len(t, loop.pendingWrites, 1) require.True(t, f.worker.SetParentBranch(parent)) diff --git a/internal/git/branch_worker_parent_change_test.go b/internal/git/branch_worker_parent_change_test.go index 2d1576c7..438109e4 100644 --- a/internal/git/branch_worker_parent_change_test.go +++ b/internal/git/branch_worker_parent_change_test.go @@ -158,7 +158,7 @@ func TestBranchWorker_AParentChangeBeforeAdmissionReplaysOntoTheNewParent(t *tes loop := newBranchWorkerEventLoop(f.worker, 0) defer loop.stopTimers() - loop.lastPushAt = time.Now() // cooldown: the write is retained + holdPushes(loop) // cooldown: the write is retained liveWrite(loop, "cm1") require.Len(t, loop.pendingWrites, 1) @@ -240,7 +240,7 @@ func TestBranchWorker_AParentChangeToAMissingParentKeepsTheWork(t *testing.T) { f.warmOnParent() loop := newBranchWorkerEventLoop(f.worker, 0) defer loop.stopTimers() - loop.lastPushAt = time.Now() + holdPushes(loop) liveWrite(loop, "cm1") require.Len(t, loop.pendingWrites, 1) beforePushAdmission = func() { f.worker.SetParentBranch("release") } diff --git a/internal/git/branch_worker_split_test.go b/internal/git/branch_worker_split_test.go index d6e7139d..61464cf3 100644 --- a/internal/git/branch_worker_split_test.go +++ b/internal/git/branch_worker_split_test.go @@ -769,7 +769,7 @@ func TestEventLoop_CommitWindowZero_HonestPerEvent(t *testing.T) { loop := newBranchWorkerEventLoop(worker, 0) // Pretend a push just happened so the cooldown gate is active; the // scheduling logic must defer the next push rather than firing it. - loop.lastPushAt = time.Now() + holdPushes(loop) loop.handleQueueItem(WorkItem{Request: &WriteRequest{ Events: []Event{ @@ -805,7 +805,7 @@ func TestEventLoop_CommitWindowZero_HonestPerEvent(t *testing.T) { func TestEventLoop_AuthorChangeFinalizesOpenWindow(t *testing.T) { worker, _, _ := setupCommitPushSplitWorker(t) loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) loop.handleQueueItem(WorkItem{Request: &WriteRequest{ Events: []Event{ @@ -828,7 +828,7 @@ func TestEventLoop_TargetChangeFinalizesOpenWindow(t *testing.T) { createPlainGitTarget(t, worker, "team-b", "team-b") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) loop.handleQueueItem(WorkItem{Request: &WriteRequest{ Events: []Event{ @@ -848,7 +848,7 @@ func TestEventLoop_TargetChangeFinalizesOpenWindow(t *testing.T) { func TestEventLoop_RepeatedPathEditsCollapseInOpenWindow(t *testing.T) { worker, _, _ := setupCommitPushSplitWorker(t) loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) first := configMapEvent("same", "alice", "team-a") second := configMapEvent("same", "alice", "team-a") @@ -877,7 +877,7 @@ func TestEventLoop_ByteCapFinalizesIncludingTrippingEvent(t *testing.T) { worker, _, _ := setupCommitPushSplitWorker(t) worker.branchBufferMaxBytes = 1 loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) loop.handleQueueItem(WorkItem{Request: &WriteRequest{ Events: []Event{configMapEvent("oversized", "alice", "team-a")}, @@ -894,7 +894,7 @@ func TestEventLoop_ByteCapFinalizesIncludingTrippingEvent(t *testing.T) { func TestEventLoop_AuthorChangeStartsFreshCommitTimer(t *testing.T) { worker, _, _ := setupCommitPushSplitWorker(t) loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) loop.handleQueueItem(WorkItem{Request: &WriteRequest{ Events: []Event{configMapEvent("first", "alice", "team-a")}, @@ -917,7 +917,7 @@ func TestEventLoop_AuthorChangeStartsFreshCommitTimer(t *testing.T) { func TestEventLoop_CrossAuthorSamePathPingPongUsesOnlyAuthorBoundaries(t *testing.T) { worker, _, _ := setupCommitPushSplitWorker(t) loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) for i, event := range []Event{ configMapEvent("F", "alice", "team-a"), @@ -959,7 +959,7 @@ func TestEventLoop_DeferredEventCommitsDuringCooldownPushTogether(t *testing.T) initialHash := initialRef.Hash() loop := newBranchWorkerEventLoop(worker, 0) - loop.lastPushAt = time.Now() + holdPushes(loop) loop.handleQueueItem(WorkItem{Request: &WriteRequest{ Events: []Event{configMapTargetEvent("live-a", "alice", "target-a")}, @@ -1018,7 +1018,7 @@ func TestResync_WorkerAppliesMarkAndSweepAndCommits(t *testing.T) { initialHash := initialRef.Hash() loop := newBranchWorkerEventLoop(worker, 0) - loop.lastPushAt = time.Now() + holdPushes(loop) // Seed a managed ConfigMap under the GitTarget path with a normal per-event commit. loop.handleQueueItem(WorkItem{Request: &WriteRequest{ @@ -1091,7 +1091,7 @@ func TestResync_WorkerEmptyDesiredSweepsManagedResource(t *testing.T) { createGitTargetWithPruneMode(t, worker, "target-a", "live", configv1alpha3.PruneAlways) loop := newBranchWorkerEventLoop(worker, 0) - loop.lastPushAt = time.Now() + holdPushes(loop) loop.handleQueueItem(WorkItem{Request: &WriteRequest{ Events: []Event{configMapTargetEvent("orphan", "alice", "target-a")}, }}) diff --git a/internal/git/commit_request_attach_loop.go b/internal/git/commit_request_attach_loop.go index 2301e621..85c8acf1 100644 --- a/internal/git/commit_request_attach_loop.go +++ b/internal/git/commit_request_attach_loop.go @@ -54,6 +54,7 @@ func (w *BranchWorker) recordCommitRequestOutcomeLocked(id commitRequestID, resu } now := time.Now() w.crOutcomes[id] = commitRequestOutcomeEntry{result: result, resolvedAt: now} + w.reportCommitRequestChange(id) for k, entry := range w.crOutcomes { if now.Sub(entry.resolvedAt) > commitRequestOutcomeTTL && w.crQueuedAttaches[k] == 0 { delete(w.crOutcomes, k) @@ -162,7 +163,10 @@ func (w *BranchWorker) setCommitRequestPhase(id commitRequestID, phase CommitReq if w.crPhases == nil { w.crPhases = map[commitRequestID]CommitRequestPhase{} } - w.crPhases[id] = phase + if w.crPhases[id] != phase { + w.crPhases[id] = phase + w.reportCommitRequestChange(id) + } } // LookupCommitRequestPhase returns where an unresolved request stands on this worker, or "" when diff --git a/internal/git/commit_request_attach_test.go b/internal/git/commit_request_attach_test.go index b68e7473..74b9e926 100644 --- a/internal/git/commit_request_attach_test.go +++ b/internal/git/commit_request_attach_test.go @@ -69,6 +69,13 @@ func forceDue(loop *branchWorkerEventLoop) { } } +// holdPushes starts the push cooldown now and keeps every push behind it, a save's too, so the local +// commits stay inspectable until the test pushes. +func holdPushes(loop *branchWorkerEventLoop) { + loop.lastPushAt = time.Now() + loop.cooldownHoldsSaves = true +} + func outcome(t *testing.T, w *BranchWorker) (FinalizeResult, bool) { t.Helper() return w.LookupCommitRequestOutcome("default", crName, "uid-"+crName) @@ -348,7 +355,7 @@ func TestAttach_ForeignWindowIsNotStolen(t *testing.T) { createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ @@ -404,7 +411,7 @@ func TestAttach_ForeignWindowClosingBeforeExpiryIsStillAMismatch(t *testing.T) { createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ @@ -447,7 +454,7 @@ func TestAttach_ASecondSaveOfTheSameAuthorIsRefusedWhileTheFirstIsInProgress(t * worker, _, _ := setupCommitPushSplitWorker(t) createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ Events: []Event{configMapTargetEvent("cm", "alice", "team-a")}, @@ -497,7 +504,7 @@ func TestAttach_ASecondSaveOfTheSameAuthorIsRefusedWhileTheFirstIsInProgress(t * worker, _, _ := setupCommitPushSplitWorker(t) createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ Events: []Event{configMapTargetEvent("cm", "alice", "team-a")}, @@ -525,7 +532,7 @@ func TestAttach_ForeignWindowOpeningAfterExpiryIsNotAMismatch(t *testing.T) { createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() // Bob parks with nothing open, and his grace runs out before anyone else starts work. @@ -625,7 +632,7 @@ func TestFinalizeOpenWindow_ReturnsCommittedFlag(t *testing.T) { createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() assert.False(t, loop.finalizeOpenWindow(), "no open window → false") @@ -669,7 +676,7 @@ func TestAttach_NoDiffResolvesAlreadyPresentOnceTheRemoteConfirms(t *testing.T) require.Empty(t, loop.pendingWrites) // Hold the push back, so the next assertion is about the finalize and nothing else. - loop.lastPushAt = time.Now() + holdPushes(loop) // A second window re-asserts the SAME object: no diff. Attach a CommitRequest and // finalize it (maxDuration 0s). @@ -776,7 +783,7 @@ func TestAttach_ConflictReplayResolvesToPostReplaySHA(t *testing.T) { // lastPushAt set so the finalize's push is deferred (a cooldown timer): the // window commits locally, then we move the remote, then push explicitly. loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ @@ -958,7 +965,7 @@ func TestAttach_ReSentAttachDuringPushCooldownDoesNotResolve(t *testing.T) { // The cooldown is active, so the finalize below commits locally and does not push. loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ @@ -1003,7 +1010,7 @@ func TestAttach_ReSentAttachDuringAFailedPushDoesNotResolve(t *testing.T) { createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ @@ -1048,7 +1055,7 @@ func TestAttach_ACommittedRequestNeverClaimsAnotherWindow(t *testing.T) { createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() const message = "save: only my own window" @@ -1083,7 +1090,7 @@ func TestAttach_ShutdownFailsACommittedRequestThatNeverPushed(t *testing.T) { createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ diff --git a/internal/git/commit_request_changes.go b/internal/git/commit_request_changes.go new file mode 100644 index 00000000..27220097 --- /dev/null +++ b/internal/git/commit_request_changes.go @@ -0,0 +1,43 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +// CommitRequestChange names a CommitRequest whose standing on its branch worker changed: it was +// registered, attached, committed or resolved, or its apply moved. It carries only the request's +// identity: the controller reads the standing from the worker, as it always does, so a change that is +// dropped or arrives twice costs at most a step. +type CommitRequestChange struct { + Namespace string + Name string + UID string +} + +// commitRequestChangeBuffer bounds the changes a burst can queue. A full buffer drops the arriving +// change: the controller's fallback poll reads the request later. +const commitRequestChangeBuffer = 256 + +// CommitRequestChanges carries each change of a CommitRequest's standing on a branch worker, so the +// controller takes the request's next step at once instead of at its next poll. Nil for a manager +// not built by NewWorkerManager. +func (m *WorkerManager) CommitRequestChanges() <-chan CommitRequestChange { + return m.commitRequestChanges +} + +// notifyCommitRequestChange queues a change, and never blocks: the worker's loop calls it, holding +// the lock over its outcome tables. +func (m *WorkerManager) notifyCommitRequestChange(change CommitRequestChange) { + if m.commitRequestChanges == nil { + return + } + select { + case m.commitRequestChanges <- change: + default: + } +} + +// reportCommitRequestChange tells the manager that the request's standing changed. +func (w *BranchWorker) reportCommitRequestChange(id commitRequestID) { + if w.commitRequestReporter != nil { + w.commitRequestReporter(CommitRequestChange(id)) + } +} diff --git a/internal/git/commit_request_empty_test.go b/internal/git/commit_request_empty_test.go index 9e78f3e0..881bbfe9 100644 --- a/internal/git/commit_request_empty_test.go +++ b/internal/git/commit_request_empty_test.go @@ -460,7 +460,7 @@ func TestCommitEmpty_ABufferFlushStillServesTheWaitingSaves(t *testing.T) { createPlainGitTarget(t, worker, "team-a", "team-a") worker.branchBufferMaxBytes = 1 loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() // hold the pushes, so the retained writes can be counted + holdPushes(loop) // hold the pushes, so the retained writes can be counted defer loop.stopTimers() serviceAttach(loop, commitEmptyReq("bob", "bob's save")) diff --git a/internal/git/commit_request_folder_test.go b/internal/git/commit_request_folder_test.go new file mode 100644 index 00000000..dca004ca --- /dev/null +++ b/internal/git/commit_request_folder_test.go @@ -0,0 +1,251 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "slices" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" +) + +// These tests drive one branch worker through a CommitRequest saved the way a person saves one: on the +// worker's clock, with the wakes its timers would deliver, and with the gaps a real cluster has +// between writes and between a request's creation and its registration (its own audit event lags). +// Every push is made at once, so the commits on the remote are the ones the windows closed into. + +// saveTarget is the commit window a GitTarget that declares none gets: the schema's defaults. +const ( + saveTargetIdle = 5 * time.Second + saveTargetMax = time.Minute +) + +// saveRun is one worker driven on a test clock. +type saveRun struct { + t *testing.T + worker *BranchWorker + loop *branchWorkerEventLoop + clock *testClock + remote string + base string + start time.Time +} + +func newSaveRun(t *testing.T) *saveRun { + t.Helper() + worker, _, remoteURL := setupCommitPushSplitWorker(t) + createPlainGitTarget(t, worker, crTarget, crTarget) + clock := useTestClock(worker) + loop := newBranchWorkerEventLoop(worker, saveTargetIdle) + loop.defaultWindow.maxDuration = saveTargetMax + t.Cleanup(loop.stopTimers) + remote := strings.TrimPrefix(remoteURL, "file://") + return &saveRun{ + t: t, worker: worker, loop: loop, clock: clock, remote: remote, + base: gitIn(t, remote, "rev-parse", "main"), start: clock.Now(), + } +} + +// until steps the clock to at, from the run's start, delivering each timer wake as it comes due. +func (r *saveRun) until(at time.Duration) { + const step = 50 * time.Millisecond + for r.clock.Now().Sub(r.start)+step <= at { + r.clock.advance(step) + r.wake() + } + r.clock.advance(at - r.clock.Now().Sub(r.start)) + r.wake() +} + +// untilResolved steps the clock until the request named name is resolved, for at most 30s. +func (r *saveRun) untilResolved(name string) { + for range 600 { + if _, ok := r.outcome(name); ok { + return + } + r.until(r.clock.Now().Sub(r.start) + 50*time.Millisecond) + } +} + +func (r *saveRun) wake() { + r.loop.lastPushAt = time.Time{} + timerWake(r.loop) +} + +func (r *saveRun) write(name string) { + r.loop.lastPushAt = time.Time{} + r.loop.handleQueueItem(WorkItem{Request: &WriteRequest{ + Events: []Event{configMapTargetEvent(name, "alice", crTarget)}, + }}) +} + +// register registers a request with the schema's defaults: CurrentOrNext, attachTimeout 5s, +// maxDuration 2s and no idleTimeout. +func (r *saveRun) register(name, message string) { + r.loop.lastPushAt = time.Time{} + req := saveRequest(name, message) + r.loop.handleQueueItem(WorkItem{Attach: req}) +} + +func saveRequest(name, message string) *AttachCommitRequest { + return &AttachCommitRequest{ + Namespace: "default", Name: name, UID: "uid-" + name, + Author: "alice", GitTargetName: crTarget, GitTargetNamespace: "default", + Message: message, + Attach: configv1alpha3.AttachCurrentOrNext, + AttachTimeout: 5 * time.Second, + MaxDuration: 2 * time.Second, + } +} + +func (r *saveRun) outcome(name string) (FinalizeResult, bool) { + return r.worker.LookupCommitRequestOutcome("default", name, "uid-"+name) +} + +// savedCommit is one commit on the remote: its message and the ConfigMaps it changes, by name. +type savedCommit struct { + hash, author, message string + configMaps []string +} + +// commits are the commits on the remote since the run started, oldest first. +func (r *saveRun) commits() []savedCommit { + var out []savedCommit + for _, hash := range splitLines(gitIn(r.t, r.remote, "rev-list", "--reverse", r.base+"..main")) { + c := savedCommit{ + hash: hash, + author: gitIn(r.t, r.remote, "log", "-1", "--format=%an", hash), + message: gitIn(r.t, r.remote, "log", "-1", "--format=%B", hash), + } + for _, path := range splitLines(gitIn(r.t, r.remote, "diff-tree", "-r", "--name-only", "--no-commit-id", hash)) { + if name, ok := strings.CutPrefix(path, crTarget+"/default/configmaps/"); ok { + c.configMaps = append(c.configMaps, strings.TrimSuffix(name, ".yaml")) + } + } + slices.Sort(c.configMaps) + out = append(out, c) + } + return out +} + +// folderStep is one thing kubectl apply -f folder/ makes reach the worker: a ConfigMap's write, or, +// with request set, the CommitRequest's registration. +type folderStep struct { + at time.Duration + write string + request bool +} + +// TestSaveFolder_AnAppliedFolderMakesOneCommitUnderTheRequestsMessage pins the folder scenario: a +// CommitRequest beside the objects it saves, applied with kubectl apply -f folder/, makes one commit +// of those objects, authored by the person, carrying the request's message, in both orders kubectl +// can apply them in. +func TestSaveFolder_AnAppliedFolderMakesOneCommitUnderTheRequestsMessage(t *testing.T) { + const message = "feat: raise the checkout limits" + cases := []struct { + name string + steps []folderStep + }{ + { + // The request's own audit event lags, so it registers once both objects are in the window: + // only CurrentOrNext attaches to it. + name: "objects first, the request registers after both", + steps: []folderStep{ + {at: 0, write: "app-config"}, + {at: 800 * time.Millisecond, write: "app-flags"}, + {at: 2 * time.Second, request: true}, + }, + }, + { + name: "objects first, the request registers between them", + steps: []folderStep{ + {at: 0, write: "app-config"}, + {at: 900 * time.Millisecond, request: true}, + {at: 1700 * time.Millisecond, write: "app-flags"}, + }, + }, + { + name: "request first", + steps: []folderStep{ + {at: 0, request: true}, + {at: 600 * time.Millisecond, write: "app-config"}, + {at: 1900 * time.Millisecond, write: "app-flags"}, + }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + r := newSaveRun(t) + for _, step := range tc.steps { + r.until(step.at) + if step.request { + r.register(crName, message) + } else { + r.write(step.write) + } + } + r.until(30 * time.Second) + + res, ok := r.outcome(crName) + require.True(t, ok, "the request is resolved") + require.NoError(t, res.Err) + assert.Equal(t, FinalizeCommitted, res.Outcome) + commits := r.commits() + require.Len(t, commits, 1, "one commit: %+v", commits) + assert.Equal(t, res.Commit, commits[0].hash) + assert.Equal(t, "alice", commits[0].author) + assert.Equal(t, message, commits[0].message) + assert.Equal(t, []string{"app-config", "app-flags"}, commits[0].configMaps) + }) + } +} + +// TestSaveFolder_ThreeSavesInARowKeepEachMessageOnItsOwnChange pins gitops-api ask #39: a write of the +// same author that no request claimed leaves a window open, and three saves follow one another, each +// filed, then written once the worker reports it registered, then waited on. The report reaches the +// client within milliseconds (CommitRequestChanges); the 100ms here also covers the client's write. +// Each save's message must land on the commit holding its own write: the earlier write may share +// the first save's commit (the folder scenario), but no save's write is committed under the +// generated message. The shift by one the ask measured came from a 2s poll for that report, as long +// as the default maxDuration, so the write arrived after the request's window had closed. +func TestSaveFolder_ThreeSavesInARowKeepEachMessageOnItsOwnChange(t *testing.T) { + r := newSaveRun(t) + r.write("unclaimed") + + at := 300 * time.Millisecond + for _, name := range []string{"save-1", "save-2", "save-3"} { + r.until(at) + r.register(name, "feat: "+name) + r.until(at + 100*time.Millisecond) + r.write("edit-of-" + name) + r.untilResolved(name) + res, ok := r.outcome(name) + require.True(t, ok, "%s is resolved", name) + require.NoError(t, res.Err) + require.Equal(t, FinalizeCommitted, res.Outcome, name) + at = r.clock.Now().Sub(r.start) + 500*time.Millisecond + } + r.until(at + time.Minute) + + commits := r.commits() + for _, name := range []string{"save-1", "save-2", "save-3"} { + var holding *savedCommit + for i := range commits { + if slices.Contains(commits[i].configMaps, "edit-of-"+name) { + holding = &commits[i] + } + } + require.NotNil(t, holding, "%s's write is on the remote: %+v", name, commits) + assert.Equal(t, "feat: "+name, holding.message, "%s's message is on its own write", name) + res, _ := r.outcome(name) + assert.Equal(t, holding.hash, res.Commit, "%s reports the commit holding its write", name) + } + assert.Equal(t, []string{"edit-of-save-1", "unclaimed"}, commits[0].configMaps, + "the unclaimed write shares the first save's commit, as CurrentOrNext documents") + assert.Len(t, commits, 3, "one commit per save: %+v", commits) +} diff --git a/internal/git/commit_window_clock_test.go b/internal/git/commit_window_clock_test.go index 5b365dc6..f9b44657 100644 --- a/internal/git/commit_window_clock_test.go +++ b/internal/git/commit_window_clock_test.go @@ -35,7 +35,7 @@ func TestWindowClock_TheIdleTimeoutIsMeasuredOnTheWorkersClock(t *testing.T) { clock := useTestClock(worker) loop := newBranchWorkerEventLoop(worker, 10*time.Second) defer loop.stopTimers() - loop.lastPushAt = time.Now() // hold the push, so the local commit stays inspectable + holdPushes(loop) // hold the push, so the local commit stays inspectable writeTo(loop, "first") require.NotNil(t, loop.openWindow) @@ -64,7 +64,7 @@ func TestWindowClock_MaxDurationIsMeasuredOnTheWorkersClock(t *testing.T) { loop := newBranchWorkerEventLoop(worker, 10*time.Second) defer loop.stopTimers() loop.defaultWindow.maxDuration = 25 * time.Second - loop.lastPushAt = time.Now() + holdPushes(loop) for i, name := range []string{"a", "b", "c"} { writeTo(loop, name) diff --git a/internal/git/commit_window_metrics_test.go b/internal/git/commit_window_metrics_test.go index e58ce24c..17eb389d 100644 --- a/internal/git/commit_window_metrics_test.go +++ b/internal/git/commit_window_metrics_test.go @@ -71,7 +71,7 @@ func TestCommitWindowMetrics_EachCloseReasonIsCountedUnderItsOwnName(t *testing. reader, err := telemetry.InitTestExporter() require.NoError(t, err) _, loop := newMetricsLoop(t, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) // max_duration: the target's maximum runs out. writeTo(loop, "a") @@ -101,7 +101,7 @@ func TestCommitWindowMetrics_AnAttachKeepsTheOpeningTimeAndCountsTheRequestsTime reader, err := telemetry.InitTestExporter() require.NoError(t, err) _, loop := newMetricsLoop(t, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) writeTo(loop, "first") opened := time.Now().Add(-10 * time.Second) // as if the write had arrived ten seconds ago @@ -124,7 +124,7 @@ func TestCommitWindowMetrics_AttachNextCountsTheWindowItCloses(t *testing.T) { reader, err := telemetry.InitTestExporter() require.NoError(t, err) _, loop := newMetricsLoop(t, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) writeTo(loop, "earlier") next := attachReq("alice", time.Hour) @@ -171,7 +171,7 @@ func TestCommitWindowMetrics_APushReplayClosesNoWindow(t *testing.T) { worker, serverRepo, remoteURL := setupCommitPushSplitWorker(t) createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() writeTo(loop, "held") @@ -196,7 +196,7 @@ func TestCommitWindowMetrics_ACommitEmptyRecordIsNotAWindow(t *testing.T) { reader, err := telemetry.InitTestExporter() require.NoError(t, err) _, loop := newMetricsLoop(t, time.Hour) - loop.lastPushAt = time.Now() // hold the push, so the record can be seen + holdPushes(loop) // hold the push, so the record can be seen serviceAttach(loop, commitEmptyReq("alice", "save: nothing changed")) forceDue(loop) diff --git a/internal/git/commit_window_timers_test.go b/internal/git/commit_window_timers_test.go index 38e27f5e..c6bbdd4e 100644 --- a/internal/git/commit_window_timers_test.go +++ b/internal/git/commit_window_timers_test.go @@ -40,7 +40,7 @@ func TestWindowTimers_ContinuousActivityStopsAtTheTargetsMaxDuration(t *testing. loop := newBranchWorkerEventLoop(worker, time.Hour) // the idle timer never fires here defer loop.stopTimers() loop.defaultWindow.maxDuration = time.Hour - loop.lastPushAt = time.Now() // hold the push, so the local commit stays inspectable + holdPushes(loop) // hold the push, so the local commit stays inspectable writeTo(loop, "first") require.NotNil(t, loop.openWindow) @@ -63,7 +63,7 @@ func TestWindowTimers_ATargetMaxDurationOfZeroCommitsEveryWrite(t *testing.T) { loop := newBranchWorkerEventLoop(worker, time.Hour) defer loop.stopTimers() loop.defaultWindow.maxDuration = 0 - loop.lastPushAt = time.Now() + holdPushes(loop) writeTo(loop, "first") writeTo(loop, "second") @@ -196,7 +196,7 @@ func TestAttachSelection_NextClosesTheOpenWindowUnderItsOwnMessage(t *testing.T) loop := newBranchWorkerEventLoop(worker, time.Hour) defer loop.stopTimers() - loop.lastPushAt = time.Now() + holdPushes(loop) writeTo(loop, "earlier") current := namedAttachReq("current", time.Hour) current.Message = "the earlier save" @@ -232,7 +232,7 @@ func TestAttachSelection_TheWorkerReportsEachPhase(t *testing.T) { phase := func() CommitRequestPhase { return worker.LookupCommitRequestPhase("default", crName, "uid-"+crName) } assert.Empty(t, phase(), "nothing is reported before the worker registers the request") - loop.lastPushAt = time.Now() // keep the push behind its cooldown, so WaitingForPush is observable + holdPushes(loop) // keep the push behind its cooldown, so WaitingForPush is observable serviceAttach(loop, attachReq("alice", time.Hour)) assert.Equal(t, PhaseWaitingForWindow, phase()) @@ -257,7 +257,7 @@ func TestWindowTimers_AWaitingRequestWithMaxDurationZeroCommitsExactlyTheNextWri worker, _, _ := setupCommitPushSplitWorker(t) createPlainGitTarget(t, worker, "team-a", "team-a") loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() // hold the pushes, so the retained writes can be inspected + holdPushes(loop) // hold the pushes, so the retained writes can be inspected defer loop.stopTimers() req := attachReq("alice", time.Hour) diff --git a/internal/git/decided_write_test.go b/internal/git/decided_write_test.go index 806c74ac..8431a3af 100644 --- a/internal/git/decided_write_test.go +++ b/internal/git/decided_write_test.go @@ -53,7 +53,7 @@ func TestDecidedWrite_SurvivesAFailedRebuildAndLandsThroughTheRetry(t *testing.T f.publish("prime") loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() // the first write waits out the push cooldown + holdPushes(loop) // the first write waits out the push cooldown defer loop.stopTimers() // One write is committed and retained, and then the checkout has to be rebuilt before anything @@ -163,7 +163,7 @@ func TestDecidedWrite_AdmissionClosesAtTheBudgetDuringAnOutage(t *testing.T) { f.publish("prime") f.worker.branchBufferMaxBytes = 1 // any retained write fills it loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() write := func(name string) { loop.handleQueueItem(WorkItem{Request: &WriteRequest{ @@ -289,7 +289,7 @@ func TestDecidedWrite_SameObjectEditsReplayAsDecidedOntoARemoteHoldingTheLast(t f.publish("prime") loop := newBranchWorkerEventLoop(f.worker, 0) t.Cleanup(loop.stopTimers) - loop.lastPushAt = time.Now() // the cooldown holds both writes back + holdPushes(loop) // the cooldown holds both writes back edit := func(value string) []byte { event := configMapTargetEvent("settings", "alice", ledgerTargetName) diff --git a/internal/git/dirty_worktree_recovery_test.go b/internal/git/dirty_worktree_recovery_test.go index 4bd07f75..4820fc58 100644 --- a/internal/git/dirty_worktree_recovery_test.go +++ b/internal/git/dirty_worktree_recovery_test.go @@ -141,7 +141,7 @@ func TestEveryLoopCommitPathRecoversADirtyWorktree(t *testing.T) { repoPath := worker.repoPath() loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() // Retained work, so commitPendingWrites cannot reset for us. diff --git a/internal/git/parent_recovery_test.go b/internal/git/parent_recovery_test.go index 0abc2e83..4067c24c 100644 --- a/internal/git/parent_recovery_test.go +++ b/internal/git/parent_recovery_test.go @@ -90,7 +90,7 @@ func TestParentRecovery_RetainedWritesLandAfterAFailedFirstAttempt(t *testing.T) f.loop.runParentProbe() // what the probe timer does } - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) liveWrite(f.loop, "cm1") require.Len(t, f.loop.pendingWrites, 1) gitIn(t, f.repoDir, "update-ref", "-d", "refs/heads/release") // the parent goes away @@ -205,7 +205,7 @@ func TestParentRecovery_ProbeBudgetIsPerWorker(t *testing.T) { if retained { f.worker.SetParentBranch("release") require.NoError(t, f.worker.ensureRepositoryInitialized(f.worker.ctx)) - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) liveWrite(f.loop, "first") require.Len(t, f.loop.pendingWrites, 1) gitIn(t, f.repoDir, "update-ref", "-d", "refs/heads/release") diff --git a/internal/git/push_cooldown_save_test.go b/internal/git/push_cooldown_save_test.go new file mode 100644 index 00000000..c9cbb154 --- /dev/null +++ b/internal/git/push_cooldown_save_test.go @@ -0,0 +1,53 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestPushCooldown_TwoSavesInARowArePushedAtOnce pins gitops-api ask #41: a push a CommitRequest waits +// on skips the cooldown, so a save that follows a push by a second still resolves at once. +func TestPushCooldown_TwoSavesInARowArePushedAtOnce(t *testing.T) { + worker, _, _ := setupCommitPushSplitWorker(t) + createPlainGitTarget(t, worker, "team-a", "team-a") + loop := newBranchWorkerEventLoop(worker, time.Hour) + defer loop.stopTimers() + loop.lastPushAt = time.Now() // a push has just been made: the cooldown runs + + for _, name := range []string{"save-1", "save-2"} { + writeTo(loop, "edit-of-"+name) + req := namedAttachReq(name, 0) // commits the open window at the attach + req.Message = "feat: " + name + serviceAttach(loop, req) + + res, ok := worker.LookupCommitRequestOutcome("default", name, "uid-"+name) + require.True(t, ok, "%s is pushed and resolved inside the cooldown", name) + require.NoError(t, res.Err) + assert.Equal(t, FinalizeCommitted, res.Outcome) + assert.Empty(t, loop.pendingWrites, "%s's write is published", name) + assert.Nil(t, loop.pushTimer, "no push waits for the cooldown") + } +} + +// TestPushCooldown_PlainWritesStillWaitOutTheCooldown pins what the cooldown protects: commits that no +// save waits on, however fast they come, share one push once it ends. +func TestPushCooldown_PlainWritesStillWaitOutTheCooldown(t *testing.T) { + worker, _, _ := setupCommitPushSplitWorker(t) + createPlainGitTarget(t, worker, "team-a", "team-a") + loop := newBranchWorkerEventLoop(worker, time.Hour) + defer loop.stopTimers() + loop.defaultWindow.maxDuration = 0 // every write is a commit of its own + loop.lastPushAt = time.Now() + + writeTo(loop, "first") + writeTo(loop, "second") + loop.endWake(0) + + assert.Len(t, loop.pendingWrites, 2, "both commits wait for the cooldown") + assert.NotNil(t, loop.pushTimer, "one push is scheduled for when it ends") +} diff --git a/internal/git/refusal_observation_test.go b/internal/git/refusal_observation_test.go index bfcb9e0b..208e84ba 100644 --- a/internal/git/refusal_observation_test.go +++ b/internal/git/refusal_observation_test.go @@ -67,7 +67,7 @@ func refuseResync(t *testing.T, f *ledgerFixture, l *branchWorkerEventLoop, imag } f.worker.lastReapply["default/"+ledgerTargetName] = time.Now().Add(-2 * refusalReapplyInterval) f.worker.refusalMu.Unlock() - l.lastPushAt = time.Now() + holdPushes(l) event := overridesDeploymentEvent(image, 3) scope := deploymentResyncScope() @@ -543,7 +543,7 @@ func TestRefusalReapply_RecoveryCancelsARequestQueuedForTheSameCollection(t *tes f := overlayRefusalFixture(t, "recovery-cancels-queued") l := newBranchWorkerEventLoop(f.worker, time.Hour) t.Cleanup(l.stopTimers) - l.lastPushAt = time.Now() + holdPushes(l) scope := ResyncScopeFor( schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}, "production") diff --git a/internal/git/replay_failure_test.go b/internal/git/replay_failure_test.go index b11d2d26..86bd04c8 100644 --- a/internal/git/replay_failure_test.go +++ b/internal/git/replay_failure_test.go @@ -73,7 +73,7 @@ func TestReplayFailure_DoesNotLetThePushSettleWorkThatWasReset(t *testing.T) { f.publish("prime") loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() // One retained write, committed locally and waiting out the push cooldown. @@ -122,7 +122,7 @@ func TestReplayFailure_DoesNotResolveACommitRequestThatNeverLanded(t *testing.T) f.publish("prime") loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ @@ -183,7 +183,7 @@ func TestReplayFailure_HoldsAWholeBatchNotJustItsFirstWrite(t *testing.T) { f.publish("prime") loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() // Two retained writes, each its own local commit. @@ -221,7 +221,7 @@ func TestReplayFailure_RecoveryRunsBeforeEveryCommitPathToo(t *testing.T) { f.publish("prime") loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ @@ -261,7 +261,7 @@ func TestReplayFailure_OnTheContentionPathIsAlsoHeld(t *testing.T) { f.publish("prime") loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ @@ -313,7 +313,7 @@ func TestReplayFailure_AResetThatDiesHalfwayIsAlsoHeld(t *testing.T) { f.publish("prime") loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) defer loop.stopTimers() loop.handleQueueItem(WorkItem{Request: &WriteRequest{ diff --git a/internal/git/replay_refusal_test.go b/internal/git/replay_refusal_test.go index 7a175e80..85acb01b 100644 --- a/internal/git/replay_refusal_test.go +++ b/internal/git/replay_refusal_test.go @@ -41,7 +41,7 @@ func newReplayRefusalFixture(t *testing.T, slug string) (*ledgerFixture, *branch f.publish("prime") loop := newBranchWorkerEventLoop(f.worker, time.Hour) - loop.lastPushAt = time.Now() + holdPushes(loop) t.Cleanup(loop.stopTimers) return f, loop } diff --git a/internal/git/retry_test.go b/internal/git/retry_test.go index 18ed1438..b722a5ea 100644 --- a/internal/git/retry_test.go +++ b/internal/git/retry_test.go @@ -151,7 +151,7 @@ func TestPublicationRetry_AMissingParentIsProbedOnTheOneSchedule(t *testing.T) { f.worker.SetParentBranch("release") require.NoError(t, f.worker.ensureRepositoryInitialized(f.worker.ctx)) - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) liveWrite(f.loop, "cm1") gitIn(t, f.repoDir, "update-ref", "-d", "refs/heads/release") f.loop.pushPending() @@ -173,7 +173,7 @@ func TestPublicationRetry_AFailureAfterTheParentReturnsWaitsForRecovery(t *testi clock := useTestClock(f.worker) f.worker.SetParentBranch("release") require.NoError(t, f.worker.ensureRepositoryInitialized(f.worker.ctx)) - f.loop.lastPushAt = time.Now() + holdPushes(f.loop) liveWrite(f.loop, "cm1") gitIn(t, f.repoDir, "update-ref", "-d", "refs/heads/release") f.loop.pushPending() diff --git a/internal/git/save_apply.go b/internal/git/save_apply.go index 97434096..61ce8942 100644 --- a/internal/git/save_apply.go +++ b/internal/git/save_apply.go @@ -79,6 +79,7 @@ func (w *BranchWorker) recordSaveApply(id commitRequestID, apply SaveApply) { askedAt = entry.askedAt } w.crApplies[id] = saveApplyEntry{apply: apply, askedAt: askedAt, updatedAt: now} + w.reportCommitRequestChange(id) } // LookupSaveApply reports where the apply a CommitRequest asked for stands, and false while the diff --git a/internal/git/worker_manager.go b/internal/git/worker_manager.go index 16d49f00..4acf3f9e 100644 --- a/internal/git/worker_manager.go +++ b/internal/git/worker_manager.go @@ -174,6 +174,10 @@ type WorkerManager struct { // whose publication report changed. See GitBranchEvents. gitBranchEvents chan event.GenericEvent + // commitRequestChanges carries the CommitRequests whose standing on a worker changed. See + // CommitRequestChanges. + commitRequestChanges chan CommitRequestChange + // claims is which GitBranch holds each repository branch (branch_owner.go). The GitBranch // reconciler writes it; EnsureWorker and the sweep read it. Reach it through Claims, which makes // one for a manager not built by NewWorkerManager. @@ -218,19 +222,20 @@ func NewWorkerManager( sensitiveResources types.SensitiveResourcePolicy, ) *WorkerManager { return &WorkerManager{ - Client: client, - Log: log, - limits: limits.withDefaults(), - sensitiveResources: sensitiveResources, - workers: make(map[BranchKey]*BranchWorker), - commitRequests: &commitRequestOwners{}, - remotes: make(map[BranchKey]RemoteObservation), - replacements: make(map[BranchKey]struct{}), - renderFidelityGate: NewRenderFidelityGate(), - publicationEvents: make(chan event.GenericEvent, publicationEventBuffer), - gitBranchEvents: make(chan event.GenericEvent, publicationEventBuffer), - claims: NewBranchClaims(), - applierAdmitted: authz.ApplierAdmitted, + Client: client, + Log: log, + limits: limits.withDefaults(), + sensitiveResources: sensitiveResources, + workers: make(map[BranchKey]*BranchWorker), + commitRequests: &commitRequestOwners{}, + remotes: make(map[BranchKey]RemoteObservation), + replacements: make(map[BranchKey]struct{}), + renderFidelityGate: NewRenderFidelityGate(), + publicationEvents: make(chan event.GenericEvent, publicationEventBuffer), + gitBranchEvents: make(chan event.GenericEvent, publicationEventBuffer), + commitRequestChanges: make(chan CommitRequestChange, commitRequestChangeBuffer), + claims: NewBranchClaims(), + applierAdmitted: authz.ApplierAdmitted, } } @@ -529,6 +534,7 @@ func (m *WorkerManager) startWorkerLocked(key BranchKey, req WorkerRequest, gitB managerCtx := m.ctx // read under m.mu; the worker's loop calls the reporter later gitBranch := req.GitBranch worker.publicationReporter = func() { go m.notifyPublication(managerCtx, gitBranch) } + worker.commitRequestReporter = m.notifyCommitRequestChange worker.scanAcceptance = m.scanAcceptance worker.applyRequester = m.applyRequester worker.claimStatus = m.claimStatus diff --git a/internal/git/write_boundary_precondition_test.go b/internal/git/write_boundary_precondition_test.go index 99a5e51f..ed4a2e38 100644 --- a/internal/git/write_boundary_precondition_test.go +++ b/internal/git/write_boundary_precondition_test.go @@ -239,7 +239,7 @@ func TestEventLoop_LiveFanInRefusal_FailsGitTargetAndCommitsNothing(t *testing.T } loop := newBranchWorkerEventLoop(worker, time.Hour) - loop.lastPushAt = time.Now() // keep the (irrelevant) push out of this test + holdPushes(loop) // keep the (irrelevant) push out of this test defer loop.stopTimers() event := overridesDeploymentEvent("ghcr.io/example/podinfo:9.9.9", 3) diff --git a/internal/watch/resource_policy_test.go b/internal/watch/resource_policy_test.go new file mode 100644 index 00000000..9d97bbc8 --- /dev/null +++ b/internal/watch/resource_policy_test.go @@ -0,0 +1,34 @@ +// SPDX-License-Identifier: Apache-2.0 + +package watch + +import ( + "testing" + + "github.com/stretchr/testify/assert" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// TestScanEntries_ACommitRequestIsNeverMirrored pins that a CommitRequest is not state to mirror: a +// WatchRule that names its type, or covers it with "*", finds it denied by policy, so a request kept +// in a folder beside the objects it saves never reaches Git. +func TestScanEntries_ACommitRequestIsNeverMirrored(t *testing.T) { + entries := scanEntriesForResourceList(&metav1.APIResourceList{ + GroupVersion: "configbutler.ai/v1alpha3", + APIResources: []metav1.APIResource{ + {Name: "commitrequests", Kind: "CommitRequest", Namespaced: true, Verbs: []string{"get", "list", "watch"}}, + {Name: "commitrequests/status", Kind: "CommitRequest", Namespaced: true, Verbs: []string{"get"}}, + {Name: "watchrules", Kind: "WatchRule", Namespaced: true, Verbs: []string{"get", "list", "watch"}}, + }, + }, true) + + allowed := map[string]bool{} + for _, entry := range entries { + allowed[entry.GVR.Resource] = entry.Allowed + if entry.GVR.Resource == "commitrequests" { + assert.Equal(t, defaultResourceExclusionReason, entry.PolicyReason) + } + } + assert.False(t, allowed["commitrequests"], "a CommitRequest is a request, never mirrored") + assert.True(t, allowed["watchrules"], "the other configbutler.ai kinds are not excluded by this rule") +} diff --git a/test/e2e/commit_request_e2e_test.go b/test/e2e/commit_request_e2e_test.go index b8945f06..d69a9b31 100644 --- a/test/e2e/commit_request_e2e_test.go +++ b/test/e2e/commit_request_e2e_test.go @@ -386,6 +386,112 @@ 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. + It("saves a folder applied with kubectl apply -f (objects first) in one commit with the request's message", + 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-configmaps" + ) + basePath := "e2e/commit-request-bundle" + + By("watching ConfigMaps too, which the folder holds") + _, err := kubectlRunWithStdin(testNs, fmt.Sprintf(`apiVersion: configbutler.ai/v1alpha3 +kind: WatchRule +metadata: + name: %s +spec: + gitTargetRef: + name: %s + rules: + - resources: ["configmaps"] +`, ruleName, gitTargetName), "apply", "-f", "-") + Expect(err).NotTo(HaveOccurred(), "failed to apply the ConfigMap WatchRule") + DeferCleanup(func() { cleanupWatchRule(ruleName, testNs) }) + verifyResourceStatus("watchrule", ruleName, testNs, "True", "Succeeded", "") + waitForStreamsRunning(gitTargetName, testNs) + + var baseSHA string + Eventually(func(g Gomega) { + baseSHA = remoteBranchHead(g, repo.CheckoutDir) + g.Expect(baseSHA).NotTo(BeEmpty(), "main exists from the previous spec's save") + }, 30*time.Second, 2*time.Second).Should(Succeed()) + + folder := renderFixtureFolder(fixtureRoot, map[string]string{"__E2E_GITTARGET__": gitTargetName}) + DeferCleanup(func() { _ = os.RemoveAll(folder) }) + + By("applying the folder: the ConfigMaps first, then the CommitRequest") + _, err = kubectlRunInNamespace(testNs, "apply", "-f", folder) + Expect(err).NotTo(HaveOccurred(), "failed to apply the folder") + + By("waiting for the request to commit the folder") + var reportedSHA string + Eventually(func(g Gomega) { + reportedSHA = expectCommitRequestCommitted(g, testNs, requestName, + recentCommitDiagnostics(repo.CheckoutDir, basePath)) + }, 2*time.Minute, 2*time.Second).Should(Succeed()) + + By("verifying one commit holds both ConfigMaps, under the request's message and author") + pullLatestRepoState(Default, repo.CheckoutDir) + commits, err := gitRun(repo.CheckoutDir, "rev-list", baseSHA+"..main") + Expect(err).NotTo(HaveOccurred()) + Expect(strings.Fields(commits)).To(Equal([]string{reportedSHA}), + "the folder lands in exactly one commit, the one the request reports\n%s", + recentCommitDiagnostics(repo.CheckoutDir, basePath)) + subject, err := gitRun(repo.CheckoutDir, "log", "-1", "--pretty=%B", reportedSHA) + Expect(err).NotTo(HaveOccurred()) + Expect(strings.TrimSpace(subject)).To(Equal(message)) + paths, err := gitRun(repo.CheckoutDir, "diff-tree", "-r", "--name-only", "--no-commit-id", reportedSHA) + Expect(err).NotTo(HaveOccurred()) + Expect(strings.Fields(paths)).To(ConsistOf( + fmt.Sprintf("%s/%s/configmaps/app-config.yaml", basePath, testNs), + fmt.Sprintf("%s/%s/configmaps/app-flags.yaml", basePath, testNs), + ), "the commit changes the two ConfigMaps and nothing else: the CommitRequest is never mirrored") + if !configuredAuthorModeEnabled() { + author := commitRequestField(Default, testNs, requestName, "{.status.author}") + commitAuthor, err := gitRun(repo.CheckoutDir, "log", "-1", "--pretty=%an", reportedSHA) + Expect(err).NotTo(HaveOccurred()) + Expect(author).NotTo(BeEmpty()) + 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") + } + _, _ = kubectlRunInNamespace(testNs, "delete", "commitrequest", requestName, "--ignore-not-found=true") + }) }) // commitRequestManifest renders a single CommitRequest document with an explicit message and a diff --git a/test/e2e/fixtures/commit-request-folder/10-app-config.yaml b/test/e2e/fixtures/commit-request-folder/10-app-config.yaml new file mode 100644 index 00000000..00a58f90 --- /dev/null +++ b/test/e2e/fixtures/commit-request-folder/10-app-config.yaml @@ -0,0 +1,6 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: app-config +data: + replicas: "3" diff --git a/test/e2e/fixtures/commit-request-folder/20-app-flags.yaml b/test/e2e/fixtures/commit-request-folder/20-app-flags.yaml new file mode 100644 index 00000000..b6431af4 --- /dev/null +++ b/test/e2e/fixtures/commit-request-folder/20-app-flags.yaml @@ -0,0 +1,6 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: app-flags +data: + checkout: enabled diff --git a/test/e2e/fixtures/commit-request-folder/30-save.yaml b/test/e2e/fixtures/commit-request-folder/30-save.yaml new file mode 100644 index 00000000..62c05e62 --- /dev/null +++ b/test/e2e/fixtures/commit-request-folder/30-save.yaml @@ -0,0 +1,11 @@ +# The save, applied with the objects it saves. kubectl applies a folder's files in name order, so +# this request lands after the ConfigMaps. Give it a new name for every apply: the spec is +# immutable, and a finished request saves nothing more. +apiVersion: configbutler.ai/v1alpha3 +kind: CommitRequest +metadata: + name: save-app-config +spec: + gitTargetRef: + name: __E2E_GITTARGET__ + message: "feat: add the app config from a folder" diff --git a/test/e2e/inplace_edit_e2e_test.go b/test/e2e/inplace_edit_e2e_test.go index 50c28aa9..62895415 100644 --- a/test/e2e/inplace_edit_e2e_test.go +++ b/test/e2e/inplace_edit_e2e_test.go @@ -348,6 +348,13 @@ func seedCommentIntoRepoFile(repo *RepoArtifacts, namespace, relPath, comment st func renderInPlaceFixtureFolder(fixtureRoot, namespace string) string { GinkgoHelper() + return renderFixtureFolder(fixtureRoot, map[string]string{"__E2E_NAMESPACE__": namespace}) +} + +// renderFixtureFolder copies a fixture folder into a new temporary directory, replacing each +// placeholder with its value in every file, and returns the directory. +func renderFixtureFolder(fixtureRoot string, placeholders map[string]string) string { + GinkgoHelper() rendered, err := os.MkdirTemp("", "gitops-reverser-e2e-manifest-folder-*") Expect(err).NotTo(HaveOccurred(), "failed to create rendered fixture directory") @@ -373,7 +380,9 @@ func renderInPlaceFixtureFolder(fixtureRoot, namespace string) string { if readErr != nil { return readErr } - content = []byte(strings.ReplaceAll(string(content), "__E2E_NAMESPACE__", namespace)) + for placeholder, value := range placeholders { + content = []byte(strings.ReplaceAll(string(content), placeholder, value)) + } if err := os.MkdirAll(filepath.Dir(dst), 0o750); err != nil { return err }