feat: replicating resources upon namespace creation (#2080)

* feat: replicating resources upon namespace creation This feature enhances the (Global)TenantResource replication by triggering a replication of resources upon a Namespace creation: this speeds up the replication of resources without waiting for the resyncPeriod that could be delayed for several reasons. Signed-off-by: Dario Tranchitella <dario@tranchitella.eu> * fix: addressing fix from github copilot Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Signed-off-by: Dario Tranchitella <dario@tranchitella.eu> --------- Signed-off-by: Dario Tranchitella <dario@tranchitella.eu> Co-authored-by: Oliver Bähler <26610571+oliverbaehler@users.noreply.github.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
2026-08-25 16:07:24 +00:00 · 2026-08-17 10:51:09 +02:00
parent 305cabd4f6
commit fed61d2815
16 changed files with 1482 additions and 171 deletions
@@ -37,6 +37,77 @@ func (p *ProcessedItems) RemoveItem(item ObjectReferenceStatus) {
 	*p = filtered
 }

+// InScope returns a copy of the items replicated into the given Tenant Namespace.
+// The result shares no backing array with the receiver, thus it can be freely
+// mutated through UpdateItem or RemoveItem.
+func (p ProcessedItems) InScope(tenant, namespace string) ProcessedItems {
+	scoped := make(ProcessedItems, 0, len(p))
+
+	for _, stat := range p {
+		if !isInScope(stat, tenant, namespace) {
+			continue
+		}
+
+		scoped = append(scoped, stat)
+	}
+
+	return scoped
+}
+
+// ReplaceScope swaps the items belonging to the given Tenant Namespace with the provided
+// ones, leaving the items of any other scope untouched: this allows a Namespace scoped
+// reconciliation to report its outcome without dropping what has been processed for
+// the remaining Namespaces.
+//
+// Items which are already tracked keep their position to avoid pointless status
+// churn, whereas the ones missing from the given list are dropped, since the scope
+// is no longer processing them. Provided items out of the given scope are ignored.
+func (p *ProcessedItems) ReplaceScope(tenant, namespace string, items ProcessedItems) {
+	replacements := make(map[gvk.ResourceID]ObjectReferenceStatusCondition, len(items))
+
+	for _, item := range items {
+		if !isInScope(item, tenant, namespace) {
+			continue
+		}
+
+		replacements[item.ResourceID] = item.ObjectReferenceStatusCondition
+	}
+
+	filtered := make(ProcessedItems, 0, len(*p)+len(replacements))
+
+	for _, stat := range *p {
+		if !isInScope(stat, tenant, namespace) {
+			filtered = append(filtered, stat)
+
+			continue
+		}
+
+		condition, ok := replacements[stat.ResourceID]
+		if !ok {
+			continue
+		}
+
+		stat.ObjectReferenceStatusCondition = condition
+
+		filtered = append(filtered, stat)
+
+		delete(replacements, stat.ResourceID)
+	}
+
+	// Appending in the provided order the items which were not yet tracked.
+	for _, item := range items {
+		if _, ok := replacements[item.ResourceID]; !ok {
+			continue
+		}
+
+		filtered = append(filtered, item)
+
+		delete(replacements, item.ResourceID)
+	}
+
+	*p = filtered
+}
+
 // Removes a condition by type.
 // Returns actual item pointer, not a copy.
 func (p *ProcessedItems) GetItem(ref gvk.ResourceID) *ObjectReferenceStatus {
@@ -72,3 +143,7 @@ func (p ProcessedItems) SortDeterministic() {
 func (p *ProcessedItems) isEqual(a, b ObjectReferenceStatus) bool {
 	return a.ResourceID == b.ResourceID
 }
+
+func isInScope(item ObjectReferenceStatus, tenant, namespace string) bool {
+	return item.Tenant == tenant && item.Namespace == namespace
+}
@@ -186,3 +186,92 @@ func TestProcessedItems_SortDeterministic(t *testing.T) {
 		}
 	}
 }
+
+func TestProcessedItems_InScope(t *testing.T) {
+	now := metav1.NewTime(time.Now())
+
+	inA := mkItem("tenant-a", "ns-a", "name-a", "Secret", metav1.ConditionTrue, "Ready", "", true, now)
+	inB := mkItem("tenant-a", "ns-a", "name-b", "ConfigMap", metav1.ConditionTrue, "Ready", "", true, now)
+	otherNs := mkItem("tenant-a", "ns-b", "name-a", "Secret", metav1.ConditionTrue, "Ready", "", true, now)
+	otherTnt := mkItem("tenant-b", "ns-a", "name-a", "Secret", metav1.ConditionTrue, "Ready", "", true, now)
+	clusterWide := mkItem("tenant-a", "", "name-a", "ClusterRole", metav1.ConditionTrue, "Ready", "", true, now)
+
+	p := meta.ProcessedItems{inA, otherNs, inB, otherTnt, clusterWide}
+
+	scoped := p.InScope("tenant-a", "ns-a")
+
+	if len(scoped) != 2 {
+		t.Fatalf("expected 2 scoped items, got %d", len(scoped))
+	}
+
+	if scoped[0].ResourceID != inA.ResourceID || scoped[1].ResourceID != inB.ResourceID {
+		t.Fatalf("expected the scoped items in their original order, got %+v", scoped)
+	}
+
+	// The receiver must not be affected by any mutation of the returned copy.
+	updated := inA
+	updated.Message = "mutated"
+	scoped.UpdateItem(updated)
+
+	if p[0].Message != "" {
+		t.Fatalf("expected the source item to be untouched, got message %q", p[0].Message)
+	}
+}
+
+func TestProcessedItems_ReplaceScope(t *testing.T) {
+	now := metav1.NewTime(time.Now())
+
+	tracked := mkItem("tenant-a", "ns-a", "name-a", "Secret", metav1.ConditionFalse, "Ready", "failed", true, now)
+	gone := mkItem("tenant-a", "ns-a", "name-gone", "Secret", metav1.ConditionTrue, "Ready", "", true, now)
+	otherNs := mkItem("tenant-a", "ns-b", "name-a", "Secret", metav1.ConditionTrue, "Ready", "keep", true, now)
+	otherTnt := mkItem("tenant-b", "ns-a", "name-a", "Secret", metav1.ConditionTrue, "Ready", "keep", true, now)
+
+	p := meta.ProcessedItems{tracked, otherNs, gone, otherTnt}
+
+	reapplied := mkItem("tenant-a", "ns-a", "name-a", "Secret", metav1.ConditionTrue, "Ready", "applied", true, now)
+	fresh := mkItem("tenant-a", "ns-a", "name-new", "ConfigMap", metav1.ConditionTrue, "Ready", "", false, now)
+	// Out of the replaced scope: it must not leak into the result.
+	foreign := mkItem("tenant-a", "ns-c", "name-a", "Secret", metav1.ConditionTrue, "Ready", "", true, now)
+
+	p.ReplaceScope("tenant-a", "ns-a", meta.ProcessedItems{reapplied, fresh, foreign})
+
+	want := []gvk.ResourceID{
+		reapplied.ResourceID, // updated in place, position preserved
+		otherNs.ResourceID,   // untouched
+		otherTnt.ResourceID,  // untouched, shifted by the dropped item
+		fresh.ResourceID,     // appended
+	}
+
+	if len(p) != len(want) {
+		t.Fatalf("expected %d items, got %d: %+v", len(want), len(p), p)
+	}
+
+	for idx := range want {
+		if p[idx].ResourceID != want[idx] {
+			t.Fatalf("at index %d: expected %v, got %v", idx, want[idx], p[idx].ResourceID)
+		}
+	}
+
+	if p[0].Status != metav1.ConditionTrue || p[0].Message != "applied" {
+		t.Fatalf("expected the tracked item condition to be replaced, got %+v", p[0].ObjectReferenceStatusCondition)
+	}
+
+	if p[1].Message != "keep" || p[2].Message != "keep" {
+		t.Fatal("expected the items of the other scopes to keep their condition")
+	}
+}
+
+func TestProcessedItems_ReplaceScope_EmptyDropsScopeOnly(t *testing.T) {
+	now := metav1.NewTime(time.Now())
+
+	inScope := mkItem("tenant-a", "ns-a", "name-a", "Secret", metav1.ConditionTrue, "Ready", "", true, now)
+	otherNs := mkItem("tenant-a", "ns-b", "name-a", "Secret", metav1.ConditionTrue, "Ready", "", true, now)
+
+	p := meta.ProcessedItems{inScope, otherNs}
+
+	p.ReplaceScope("tenant-a", "ns-a", nil)
+
+	if len(p) != 1 || p[0].ResourceID != otherNs.ResourceID {
+		t.Fatalf("expected only the out of scope item to survive, got %+v", p)
+	}
+}
@@ -9,6 +9,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"

 	"github.com/projectcapsule/capsule/pkg/runtime/configuration"
+	"github.com/projectcapsule/capsule/pkg/runtime/gvk"
 )

 type Processor struct {
@@ -25,3 +26,15 @@ type ProcessorOptions struct {
 	Force            bool
 	Owner            *metav1.OwnerReference
 }
+
+// Scope narrows a reconciliation down to the items replicated into a
+// single Namespace of a single Tenant.
+type Scope struct {
+	Tenant    string
+	Namespace string
+}
+
+// Matches states whether the given resource identity belongs to the scope.
+func (s Scope) Matches(id gvk.ResourceID) bool {
+	return id.Tenant == s.Tenant && id.Namespace == s.Namespace
+}
@@ -50,6 +50,74 @@ func (p *Processor) Reconcile(
 	return nil
 }

+// ReconcileNamespace applies the accumulated items targeting a single Namespace of a
+// single Tenant, returning the processed items of that scope only.
+//
+// Contrarily to Reconcile, the given Accumulator is not authoritative for the whole
+// Tenant: nothing is pruned, nor disowned, since the items missing from it may just
+// belong to a Namespace this run knows nothing about. Pruning remains a duty of the
+// full reconciliation.
+//
+// The returned items are meant to be grafted on the persisted status through
+// meta.ProcessedItems.ReplaceScope, which leaves the other Namespaces untouched.
+// They are returned along an error too, since they carry the outcome of the single
+// items which have been processed.
+func (p *Processor) ReconcileNamespace(
+	ctx context.Context,
+	log logr.Logger,
+	c client.Client,
+	current meta.ProcessedItems,
+	acc Accumulator,
+	opts ProcessorOptions,
+	scope Scope,
+) (meta.ProcessedItems, error) {
+	if scope.Namespace == "" {
+		return nil, fmt.Errorf("cannot process a Namespace scope without a Namespace")
+	}
+
+	log = log.WithValues("tenant", scope.Tenant, "namespace", scope.Namespace)
+
+	processed := current.InScope(scope.Tenant, scope.Namespace)
+
+	scoped, skipped := scopedAccumulator(acc, scope)
+	if skipped > 0 {
+		log.V(5).Info("ignored accumulated items out of the processed scope", "ignored", skipped)
+	}
+
+	log.V(5).Info("starting scoped processing", "present", len(processed), "items", len(scoped))
+
+	if itemErrors := p.applyAccumulatedItems(ctx, log, c, &processed, scoped, opts); itemErrors > 0 {
+		return processed, fmt.Errorf("applying of %d resources failed", itemErrors)
+	}
+
+	log.V(4).Info("scoped processing completed")
+
+	return processed, nil
+}
+
+// Retains the accumulated items belonging to the given scope only, along with the
+// amount of the ignored ones.
+func scopedAccumulator(acc Accumulator, scope Scope) (Accumulator, int) {
+	scoped := make(Accumulator, len(acc))
+	ignored := 0
+
+	for key, item := range acc {
+		if item == nil {
+			continue
+		}
+
+		if !scope.Matches(item.Resource) {
+			ignored++
+
+			continue
+		}
+
+		scoped[key] = item
+	}
+
+	return scoped, ignored
+}
+
 func (p *Processor) pruneProcessedItems(
 	ctx context.Context,
 	log logr.Logger,
@@ -4,9 +4,11 @@
 package processor

 import (
+	"context"
 	"errors"
 	"testing"

+	"github.com/go-logr/logr"
 	k8smeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -133,3 +135,146 @@ func TestFailAndRecord(t *testing.T) {
 		t.Fatalf("expected message %q, got %q", "prefix: boom", got.Message)
 	}
 }
+
+func TestProcessorScopeMatches(t *testing.T) {
+	t.Parallel()
+
+	scope := Scope{Tenant: "tenant-a", Namespace: "ns-a"}
+
+	for _, tc := range []struct {
+		name string
+		id   gvk.ResourceID
+		want bool
+	}{
+		{
+			name: "same tenant and namespace",
+			id:   resourceID("tenant-a", "ns-a", "settings"),
+			want: true,
+		},
+		{
+			name: "other namespace",
+			id:   resourceID("tenant-a", "ns-b", "settings"),
+			want: false,
+		},
+		{
+			name: "other tenant",
+			id:   resourceID("tenant-b", "ns-a", "settings"),
+			want: false,
+		},
+		{
+			name: "no namespace",
+			id:   resourceID("tenant-a", "", "settings"),
+			want: false,
+		},
+	} {
+		if got := scope.Matches(tc.id); got != tc.want {
+			t.Fatalf("%s: expected %v, got %v", tc.name, tc.want, got)
+		}
+	}
+}
+
+func TestScopedAccumulator(t *testing.T) {
+	t.Parallel()
+
+	inScope := resourceID("tenant-a", "ns-a", "settings")
+	otherNs := resourceID("tenant-a", "ns-b", "settings")
+	otherTnt := resourceID("tenant-b", "ns-a", "settings")
+
+	acc := Accumulator{
+		inScope.GetKey(""):  {Resource: inScope},
+		otherNs.GetKey(""):  {Resource: otherNs},
+		otherTnt.GetKey(""): {Resource: otherTnt},
+		"empty":             nil,
+	}
+
+	scoped, ignored := scopedAccumulator(acc, Scope{Tenant: "tenant-a", Namespace: "ns-a"})
+
+	if len(scoped) != 1 {
+		t.Fatalf("expected a single scoped item, got %d", len(scoped))
+	}
+
+	if scoped[inScope.GetKey("")] == nil {
+		t.Fatal("expected the in scope item to be retained")
+	}
+
+	if ignored != 2 {
+		t.Fatalf("expected 2 ignored items, got %d", ignored)
+	}
+
+	// The nil entry is dropped without being accounted as ignored.
+	if _, ok := scoped["empty"]; ok {
+		t.Fatal("expected the empty entry to be dropped")
+	}
+}
+
+func TestReconcileNamespaceRequiresNamespace(t *testing.T) {
+	t.Parallel()
+
+	items, err := (&Processor{}).ReconcileNamespace(
+		context.Background(),
+		logr.Discard(),
+		nil,
+		nil,
+		Accumulator{},
+		ProcessorOptions{},
+		Scope{Tenant: "tenant-a"},
+	)
+	if err == nil {
+		t.Fatal("expected an error for a scope without a Namespace")
+	}
+
+	if items != nil {
+		t.Fatalf("expected no processed item, got %+v", items)
+	}
+}
+
+func TestReconcileNamespaceSeedsScopeOnly(t *testing.T) {
+	t.Parallel()
+
+	inScope := resourceID("tenant-a", "ns-a", "settings")
+	otherNs := resourceID("tenant-a", "ns-b", "settings")
+
+	current := meta.ProcessedItems{
+		{ResourceID: otherNs, ObjectReferenceStatusCondition: meta.ObjectReferenceStatusCondition{Created: true}},
+		{ResourceID: inScope, ObjectReferenceStatusCondition: meta.ObjectReferenceStatusCondition{Created: true}},
+	}
+
+	// An empty Accumulator reaches out to no client at all: the outcome is only made
+	// of what was already tracked for the reconciled scope.
+	items, err := (&Processor{}).ReconcileNamespace(
+		context.Background(),
+		logr.Discard(),
+		nil,
+		current,
+		Accumulator{},
+		ProcessorOptions{},
+		Scope{Tenant: "tenant-a", Namespace: "ns-a"},
+	)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	if len(items) != 1 || items[0].ResourceID != inScope {
+		t.Fatalf("expected only the in scope item, got %+v", items)
+	}
+
+	if !items[0].Created {
+		t.Fatal("expected the tracked creation flag to be carried over")
+	}
+
+	if len(current) != 2 {
+		t.Fatalf("expected the given items to be left untouched, got %+v", current)
+	}
+}
+
+func resourceID(tenant, namespace, name string) gvk.ResourceID {
+	return gvk.ResourceID{
+		TenantResourceIDWithOrigin: gvk.TenantResourceIDWithOrigin{
+			TenantResourceID: gvk.TenantResourceID{Tenant: tenant},
+		},
+		Version:   "v1",
+		Kind:      "ConfigMap",
+		Name:      name,
+		Namespace: namespace,
+	}
+}