From 6a2137e579d475fce8bb6863b21996559c62cfc1 Mon Sep 17 00:00:00 2001 From: Michael Basov <124149774+banlor@users.noreply.github.com> Date: Sat, 19 Sep 2026 11:36:30 +0500 Subject: [PATCH] Reconcile GitRepository includes on artifact content changes Index local include references and enqueue dependants when an included artifact revision or digest changes. Ignore metadata-only updates and skip suspended, deleting and cyclic dependants. Add envtest coverage for nested includes without changing the parent Git revision. Assisted-by: Hermes Agent (gpt-6-astra / openai-codex) Signed-off-by: Michael Basov <124149774+banlor@users.noreply.github.com> --- docs/spec/v1/gitrepositories.md | 7 + .../controller/gitrepository_controller.go | 9 ++ internal/controller/gitrepository_include.go | 103 ++++++++++++ .../gitrepository_include_propagation_test.go | 147 ++++++++++++++++++ .../controller/gitrepository_include_test.go | 94 +++++++++++ .../source_content_predicate_test.go | 63 ++++++++ internal/controller/source_predicate.go | 3 +- 7 files changed, 425 insertions(+), 1 deletion(-) create mode 100644 internal/controller/gitrepository_include.go create mode 100644 internal/controller/gitrepository_include_propagation_test.go create mode 100644 internal/controller/gitrepository_include_test.go create mode 100644 internal/controller/source_content_predicate_test.go diff --git a/docs/spec/v1/gitrepositories.md b/docs/spec/v1/gitrepositories.md index 4c9d244af..d4965c4f4 100644 --- a/docs/spec/v1/gitrepositories.md +++ b/docs/spec/v1/gitrepositories.md @@ -852,6 +852,13 @@ multiple benefits over regular submodules: - Multiple `GitRepository` objects could include the same repository, which decreases the amount of cloning done compared to using submodules. +Changes to an included Artifact's revision or digest trigger reconciliation +outside the interval window. The including Artifact retains its own Git revision; +its digest reflects the combined contents. This also propagates changes through +nested includes. Include references are local to the GitRepository's namespace. +Avoid circular includes: to prevent immediate rebuild loops, changes are not +propagated along cyclic include edges. Those repositories remain interval-driven. + ```yaml --- apiVersion: source.toolkit.fluxcd.io/v1 diff --git a/internal/controller/gitrepository_controller.go b/internal/controller/gitrepository_controller.go index a3c27c9b6..df96442b3 100644 --- a/internal/controller/gitrepository_controller.go +++ b/internal/controller/gitrepository_controller.go @@ -45,6 +45,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" @@ -196,10 +197,18 @@ func (r *GitRepositoryReconciler) SetupWithManager(mgr ctrl.Manager, opts GitRep r.features = features.FeatureGates() } + if err := mgr.GetCache().IndexField(context.Background(), &sourcev1.GitRepository{}, + indexKeyGitRepositoryInclude, indexGitRepositoryIncludes); err != nil { + return fmt.Errorf("failed indexing GitRepository includes: %w", err) + } + return ctrl.NewControllerManagedBy(mgr). For(&sourcev1.GitRepository{}, builder.WithPredicates( predicate.Or(predicate.GenerationChangedPredicate{}, predicates.ReconcileRequestedPredicate{}), )). + Watches(&sourcev1.GitRepository{}, + handler.EnqueueRequestsFromMapFunc(r.requestsForIncludeChange), + builder.WithPredicates(SourceRevisionChangePredicate{})). WithOptions(controller.Options{ RateLimiter: opts.RateLimiter, }). diff --git a/internal/controller/gitrepository_include.go b/internal/controller/gitrepository_include.go new file mode 100644 index 000000000..cc0c9ab8b --- /dev/null +++ b/internal/controller/gitrepository_include.go @@ -0,0 +1,103 @@ +/* +Copyright 2026 The Flux authors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +import ( + "context" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/util/sets" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + sourcev1 "github.com/fluxcd/source-controller/api/v1" +) + +const indexKeyGitRepositoryInclude = ".metadata.gitRepositoryInclude" + +// indexGitRepositoryIncludes indexes local references, independent of artifact availability. +func indexGitRepositoryIncludes(o client.Object) []string { + repo, ok := o.(*sourcev1.GitRepository) + if !ok { + return nil + } + refs := sets.New[string]() + for _, incl := range repo.Spec.Include { + refs.Insert(incl.GitRepositoryRef.Name) + } + return sets.List(refs) +} + +// requestsForIncludeChange enqueues repositories including the changed artifact. +func (r *GitRepositoryReconciler) requestsForIncludeChange(ctx context.Context, o client.Object) []reconcile.Request { + repo, ok := o.(*sourcev1.GitRepository) + if !ok || repo.GetArtifact() == nil { + return nil + } + var list sourcev1.GitRepositoryList + if err := r.List(ctx, &list, client.InNamespace(repo.Namespace), client.MatchingFields{ + indexKeyGitRepositoryInclude: repo.Name, + }); err != nil { + ctrl.LoggerFrom(ctx).Error(err, "failed to list GitRepositories for include change") + return nil + } + if len(list.Items) == 0 { + return nil + } + + // Do not turn an existing include cycle into an event-driven rebuild loop. + // Each repository is visited at most once, through the manager's scoped cache. + ancestors, err := r.includeDependencies(ctx, repo) + if err != nil { + ctrl.LoggerFrom(ctx).Error(err, "failed to check GitRepository include dependencies") + return nil + } + var requests []reconcile.Request + for i := range list.Items { + dependent := &list.Items[i] + if dependent.Spec.Suspend || !dependent.DeletionTimestamp.IsZero() || ancestors.Has(dependent.Name) { + continue + } + requests = append(requests, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(dependent)}) + } + return requests +} + +// includeDependencies returns transitive includes and the repository itself. +// Cyclic edges remain interval-driven, as they were before include watches. +func (r *GitRepositoryReconciler) includeDependencies(ctx context.Context, repo *sourcev1.GitRepository) (sets.Set[string], error) { + seen := sets.New(repo.Name) + pending := append([]sourcev1.GitRepositoryInclude(nil), repo.Spec.Include...) + for len(pending) > 0 { + name := pending[0].GitRepositoryRef.Name + pending = pending[1:] + if seen.Has(name) { + continue + } + seen.Insert(name) + var dep sourcev1.GitRepository + if err := r.Get(ctx, client.ObjectKey{Namespace: repo.Namespace, Name: name}, &dep); err != nil { + if apierrors.IsNotFound(err) { + continue + } + return nil, err + } + pending = append(pending, dep.Spec.Include...) + } + return seen, nil +} diff --git a/internal/controller/gitrepository_include_propagation_test.go b/internal/controller/gitrepository_include_propagation_test.go new file mode 100644 index 000000000..abde04926 --- /dev/null +++ b/internal/controller/gitrepository_include_propagation_test.go @@ -0,0 +1,147 @@ +/* +Copyright 2026 The Flux authors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +import ( + "os" + "path/filepath" + "testing" + "time" + + gogit "github.com/go-git/go-git/v5" + . "github.com/onsi/gomega" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/fluxcd/pkg/apis/meta" + "github.com/fluxcd/pkg/git" + "github.com/fluxcd/pkg/gittestserver" + "github.com/fluxcd/pkg/runtime/conditions" + sourcev1 "github.com/fluxcd/source-controller/api/v1" +) + +// TestGitRepositoryIncludePropagation exercises the manager watch, checkout, +// composition and artifact publication through two include edges. +func TestGitRepositoryIncludePropagation(t *testing.T) { + g := NewWithT(t) + server, err := gittestserver.NewTempGitServer() + g.Expect(err).NotTo(HaveOccurred()) + t.Cleanup(func() { g.Expect(os.RemoveAll(server.Root())).To(Succeed()) }) + server.AutoCreate() + g.Expect(server.StartHTTP()).To(Succeed()) + t.Cleanup(server.StopHTTP) + + leafDir, mainDir := t.TempDir(), t.TempDir() + g.Expect(os.WriteFile(filepath.Join(leafDir, "value.txt"), []byte("before"), 0o644)).To(Succeed()) + g.Expect(os.WriteFile(filepath.Join(mainDir, "root.txt"), []byte("unchanged"), 0o644)).To(Succeed()) + leafGit, err := initGitRepo(server, leafDir, git.DefaultBranch, "/leaf.git") + g.Expect(err).NotTo(HaveOccurred()) + _, err = initGitRepo(server, mainDir, git.DefaultBranch, "/main.git") + g.Expect(err).NotTo(HaveOccurred()) + + newSource := func(path string, include *sourcev1.GitRepository) *sourcev1.GitRepository { + repo := &sourcev1.GitRepository{ + ObjectMeta: metav1.ObjectMeta{GenerateName: "include-chain-", Namespace: "default"}, + Spec: sourcev1.GitRepositorySpec{ + URL: server.HTTPAddress() + path, + Interval: metav1.Duration{Duration: time.Hour}, + }, + } + if include != nil { + repo.Spec.Include = []sourcev1.GitRepositoryInclude{{GitRepositoryRef: meta.LocalObjectReference{Name: include.Name}, ToPath: "included"}} + } + g.Expect(k8sClient.Create(ctx, repo)).To(Succeed()) + t.Cleanup(func() { g.Expect(client.IgnoreNotFound(k8sClient.Delete(ctx, repo))).To(Succeed()) }) + g.Eventually(func() bool { + if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(repo), repo); err != nil { + return false + } + return conditions.IsReady(repo) && repo.GetArtifact() != nil + }, 30*time.Second, 100*time.Millisecond).Should(BeTrue()) + return repo + } + leaf := newSource("/leaf.git", nil) + middle := newSource("/main.git", leaf) + outer := newSource("/main.git", middle) + middleBefore, outerBefore := middle.Status.Artifact.DeepCopy(), outer.Status.Artifact.DeepCopy() + + // Only the leaf's Git revision changes. The other objects have hour-long + // intervals, so their updates must arrive via the include watches. + g.Expect(os.WriteFile(filepath.Join(leafDir, "value.txt"), []byte("after"), 0o644)).To(Succeed()) + g.Expect(commitFromFixture(leafGit, leafDir)).To(Succeed()) + g.Expect(leafGit.Push(&gogit.PushOptions{})).To(Succeed()) + g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(leaf), leaf)).To(Succeed()) + leaf.Annotations = map[string]string{meta.ReconcileRequestAnnotation: "leaf-update"} + g.Expect(k8sClient.Update(ctx, leaf)).To(Succeed()) + for _, obj := range []*sourcev1.GitRepository{middle, outer} { + before := middleBefore + if obj == outer { + before = outerBefore + } + g.Eventually(func() bool { + if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(obj), obj); err != nil { + return false + } + return conditions.IsReady(obj) && obj.GetArtifact() != nil && obj.GetArtifact().Digest != before.Digest + }, 30*time.Second, 100*time.Millisecond).Should(BeTrue()) + g.Expect(obj.Status.Artifact.Revision).To(Equal(before.Revision)) + } + g.Expect(middle.Status.IncludedArtifacts[0].Digest).NotTo(BeEmpty()) + g.Expect(outer.Status.IncludedArtifacts[0].Digest).To(Equal(middle.Status.Artifact.Digest)) + extracted := filepath.Join(t.TempDir(), "artifact") + g.Expect(testStorage.CopyToPath(outer.Status.Artifact, ".", extracted)).To(Succeed()) + content, err := os.ReadFile(filepath.Join(extracted, "included", "included", "value.txt")) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(string(content)).To(Equal("after")) + + // A forced no-op reconciliation keeps the artifact stable. + before := outer.Status.Artifact.DeepCopy() + outer.Annotations = map[string]string{meta.ReconcileRequestAnnotation: "no-op"} + g.Expect(k8sClient.Update(ctx, outer)).To(Succeed()) + g.Eventually(func() string { + g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(outer), outer)).To(Succeed()) + return outer.Status.LastHandledReconcileAt + }, 30*time.Second, 100*time.Millisecond).Should(Equal("no-op")) + g.Expect(outer.Status.Artifact).To(Equal(before)) + + // Timestamp-only source events do not requeue either dependent. + middleVersion, outerVersion := middle.ResourceVersion, outer.ResourceVersion + g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(leaf), leaf)).To(Succeed()) + leaf.Status.Artifact.LastUpdateTime = metav1.Now() + g.Expect(k8sClient.Status().Update(ctx, leaf)).To(Succeed()) + g.Consistently(func() []string { + g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(middle), middle)).To(Succeed()) + g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(outer), outer)).To(Succeed()) + return []string{middle.ResourceVersion, outer.ResourceVersion} + }, time.Second, 100*time.Millisecond).Should(Equal([]string{middleVersion, outerVersion})) +} + +func TestGitRepositoryIncludeArchiveStability(t *testing.T) { + g := NewWithT(t) + dir := t.TempDir() + file := filepath.Join(dir, "value") + g.Expect(os.WriteFile(file, []byte("unchanged"), 0o644)).To(Succeed()) + obj := &sourcev1.GitRepository{ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "archive-stability"}} + artifact := testStorage.NewArtifactFor(sourcev1.GitRepositoryKind, obj, "same", "test.tar.gz") + g.Expect(testStorage.MkdirAll(artifact)).To(Succeed()) + t.Cleanup(func() { _, err := testStorage.RemoveAll(artifact); g.Expect(err).NotTo(HaveOccurred()) }) + g.Expect(testStorage.Archive(&artifact, dir, nil)).To(Succeed()) + digest := artifact.Digest + g.Expect(os.Chtimes(file, time.Unix(1000, 0), time.Unix(1000, 0))).To(Succeed()) + g.Expect(testStorage.Archive(&artifact, dir, nil)).To(Succeed()) + g.Expect(artifact.Digest).To(Equal(digest)) +} diff --git a/internal/controller/gitrepository_include_test.go b/internal/controller/gitrepository_include_test.go new file mode 100644 index 000000000..d6bf6c1c2 --- /dev/null +++ b/internal/controller/gitrepository_include_test.go @@ -0,0 +1,94 @@ +/* +Copyright 2026 The Flux authors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +import ( + "context" + "testing" + + . "github.com/onsi/gomega" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + "github.com/fluxcd/pkg/apis/meta" + sourcev1 "github.com/fluxcd/source-controller/api/v1" +) + +func TestGitRepositoryIncludeMapping(t *testing.T) { + g := NewWithT(t) + ctx := context.Background() + scheme := runtime.NewScheme() + g.Expect(sourcev1.AddToScheme(scheme)).To(Succeed()) + newRepo := func(ns, name string, includes ...string) *sourcev1.GitRepository { + r := &sourcev1.GitRepository{ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: name}} + for _, ref := range includes { + r.Spec.Include = append(r.Spec.Include, sourcev1.GitRepositoryInclude{GitRepositoryRef: meta.LocalObjectReference{Name: ref}}) + } + return r + } + leaf := newRepo("one", "leaf") + leaf.Status.Artifact = &meta.Artifact{Revision: "same", Digest: "sha256:new"} + parent := newRepo("one", "parent", "leaf", "leaf") + otherNS := newRepo("two", "parent", "leaf") + suspended := newRepo("one", "suspended", "leaf") + suspended.Spec.Suspend = true + c := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(leaf, parent, otherNS, suspended). + WithIndex(&sourcev1.GitRepository{}, indexKeyGitRepositoryInclude, indexGitRepositoryIncludes).Build() + r := &GitRepositoryReconciler{Client: c} + request := reconcile.Request{NamespacedName: client.ObjectKeyFromObject(parent)} + g.Expect(indexGitRepositoryIncludes(parent)).To(Equal([]string{"leaf"})) + g.Expect(r.requestsForIncludeChange(ctx, leaf)).To(Equal([]reconcile.Request{request})) + + // Updating the specification removes old references and adds new ones. + g.Expect(c.Get(ctx, client.ObjectKeyFromObject(parent), parent)).To(Succeed()) + parent.Spec.Include = newRepo("one", "parent", "replacement").Spec.Include + g.Expect(c.Update(ctx, parent)).To(Succeed()) + g.Expect(r.requestsForIncludeChange(ctx, leaf)).To(BeEmpty()) + replacement := newRepo("one", "replacement") + replacement.Status.Artifact = leaf.Status.Artifact.DeepCopy() + g.Expect(r.requestsForIncludeChange(ctx, replacement)).To(Equal([]reconcile.Request{request})) + replacement.Status.Artifact = nil + g.Expect(r.requestsForIncludeChange(ctx, replacement)).To(BeEmpty()) +} + +func TestGitRepositoryIncludeCycles(t *testing.T) { + for _, size := range []int{1, 2, 3} { + t.Run(string(rune('0'+size)), func(t *testing.T) { + g := NewWithT(t) + scheme := runtime.NewScheme() + g.Expect(sourcev1.AddToScheme(scheme)).To(Succeed()) + var objs []client.Object + for i := 0; i < size; i++ { + objs = append(objs, &sourcev1.GitRepository{ + ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: string(rune('a' + i))}, + Spec: sourcev1.GitRepositorySpec{Include: []sourcev1.GitRepositoryInclude{{GitRepositoryRef: meta.LocalObjectReference{Name: string(rune('a' + (i+1)%size))}}}}, + Status: sourcev1.GitRepositoryStatus{Artifact: &meta.Artifact{Revision: "same", Digest: "sha256:new"}}, + }) + } + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...). + WithIndex(&sourcev1.GitRepository{}, indexKeyGitRepositoryInclude, indexGitRepositoryIncludes).Build() + r := &GitRepositoryReconciler{Client: c} + for _, obj := range objs { + g.Expect(r.requestsForIncludeChange(context.Background(), obj)).To(BeEmpty()) + } + }) + } +} diff --git a/internal/controller/source_content_predicate_test.go b/internal/controller/source_content_predicate_test.go new file mode 100644 index 000000000..c9caa851d --- /dev/null +++ b/internal/controller/source_content_predicate_test.go @@ -0,0 +1,63 @@ +/* +Copyright 2026 The Flux authors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +import ( + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/event" + + "github.com/fluxcd/pkg/apis/meta" + sourcev1 "github.com/fluxcd/source-controller/api/v1" +) + +func TestSourceRevisionChangePredicate_Content(t *testing.T) { + base := &sourcev1.GitRepository{Status: sourcev1.GitRepositoryStatus{ + Artifact: &meta.Artifact{Revision: "main@sha1:abc", Digest: "sha256:aaa"}, + }} + for _, tt := range []struct { + name string + change func(*sourcev1.GitRepository) + want bool + }{ + {"unchanged", func(*sourcev1.GitRepository) {}, false}, + {"digest only", func(r *sourcev1.GitRepository) { r.Status.Artifact.Digest = "sha256:bbb" }, true}, + {"revision only", func(r *sourcev1.GitRepository) { r.Status.Artifact.Revision = "main@sha1:def" }, true}, + {"timestamp only", func(r *sourcev1.GitRepository) { r.Status.Artifact.LastUpdateTime = metav1.Now() }, false}, + {"url only", func(r *sourcev1.GitRepository) { r.Status.Artifact.URL = "http://localhost/new" }, false}, + {"conditions only", func(r *sourcev1.GitRepository) { + r.Status.Conditions = []metav1.Condition{{Type: meta.ReadyCondition, Status: metav1.ConditionTrue}} + }, false}, + {"artifact removed", func(r *sourcev1.GitRepository) { r.Status.Artifact = nil }, false}, + } { + t.Run(tt.name, func(t *testing.T) { + updated := base.DeepCopy() + tt.change(updated) + if got := (SourceRevisionChangePredicate{}).Update(event.UpdateEvent{ObjectOld: base, ObjectNew: updated}); got != tt.want { + t.Fatalf("Update() = %v, want %v", got, tt.want) + } + }) + } + p := SourceRevisionChangePredicate{} + if !p.Update(event.UpdateEvent{ObjectOld: &sourcev1.GitRepository{}, ObjectNew: base}) { + t.Fatal("first artifact must enqueue dependents") + } + if p.Update(event.UpdateEvent{ObjectNew: base}) || p.Update(event.UpdateEvent{ObjectOld: base}) { + t.Fatal("incomplete events must be ignored") + } +} diff --git a/internal/controller/source_predicate.go b/internal/controller/source_predicate.go index 968f2def9..36c1e6615 100644 --- a/internal/controller/source_predicate.go +++ b/internal/controller/source_predicate.go @@ -47,7 +47,8 @@ func (SourceRevisionChangePredicate) Update(e event.UpdateEvent) bool { } if oldSource.GetArtifact() != nil && newSource.GetArtifact() != nil && - oldSource.GetArtifact().Revision != newSource.GetArtifact().Revision { + (oldSource.GetArtifact().Revision != newSource.GetArtifact().Revision || + oldSource.GetArtifact().Digest != newSource.GetArtifact().Digest) { return true }