From fed61d2815c622862ca35482561200091e2a277f Mon Sep 17 00:00:00 2001 From: Dario Tranchitella Date: Mon, 17 Aug 2026 10:51:09 +0200 Subject: [PATCH] feat: replicating resources upon namespace creation (#2080) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 * 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 --------- Signed-off-by: Dario Tranchitella 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> --- api/v1beta2/tenantresource_types.go | 6 + internal/controllers/resources/collect.go | 125 ++++- .../controllers/resources/collect_test.go | 51 ++ internal/controllers/resources/global.go | 118 +---- .../controllers/resources/impersonation.go | 152 ++++++ internal/controllers/resources/manager.go | 8 + .../resources/namespace_trigger.go | 476 ++++++++++++++++++ internal/controllers/resources/namespaced.go | 60 +-- .../resources/scoped_replication.go | 108 ++++ pkg/api/meta/processed.go | 75 +++ pkg/api/meta/processed_test.go | 89 ++++ pkg/api/processor/processor.go | 13 + pkg/api/processor/processor_func.go | 68 +++ pkg/api/processor/processor_func_test.go | 145 ++++++ pkg/runtime/predicates/created_after.go | 49 ++ pkg/runtime/predicates/created_after_test.go | 110 ++++ 16 files changed, 1482 insertions(+), 171 deletions(-) create mode 100644 internal/controllers/resources/impersonation.go create mode 100644 internal/controllers/resources/namespace_trigger.go create mode 100644 internal/controllers/resources/scoped_replication.go create mode 100644 pkg/runtime/predicates/created_after.go create mode 100644 pkg/runtime/predicates/created_after_test.go diff --git a/api/v1beta2/tenantresource_types.go b/api/v1beta2/tenantresource_types.go index 3b9b6119..71f49573 100644 --- a/api/v1beta2/tenantresource_types.go +++ b/api/v1beta2/tenantresource_types.go @@ -60,6 +60,12 @@ type TenantResourceCommonSpec struct { Resources []ResourceSpec `json:"resources"` } +// IsCordoned states whether the replication is paused, in which case no apply nor deletion +// must be performed. Being an optional field, an unset value is not cordoned. +func (s *TenantResourceCommonSpec) IsCordoned() bool { + return s.Cordoned != nil && *s.Cordoned +} + type TenantResourceCommonSpecSettings struct { // Enabling this allows TenanResources to interact with objects which were not created by a TenantResource. In this case on prune no deletion of the entire object is made. // +kubebuilder:default=false diff --git a/internal/controllers/resources/collect.go b/internal/controllers/resources/collect.go index 55198e9d..78abd709 100644 --- a/internal/controllers/resources/collect.go +++ b/internal/controllers/resources/collect.go @@ -217,7 +217,7 @@ func (co *Collector) AddToAccumulation( return err } - if err := co.validateClusterScopedObjectAllowed(opts, obj); err != nil { + if err := co.validateClusterScopedObjectAllowed(opts, obj, ns); err != nil { return err } @@ -276,6 +276,58 @@ func (co *Collector) AddToAccumulation( return nil } +// CollectForNamespace collects the items of a single ResourceSpec targeting the given Namespace, +// replicating into it the already loaded source objects. +// +// The given options are used as a template: the iterator is always derived from the target +// Namespace, thus callers are not required to prepare it. +func (co *Collector) CollectForNamespace( + ctx context.Context, + c client.Client, + opts CollectorOptions, + tnt capsulev1beta2.Tenant, + resourceIndex string, + spec capsulev1beta2.ResourceSpec, + sources map[gvk.ResourceKey]*unstructured.Unstructured, + target *corev1.Namespace, +) error { + log := log.FromContext(ctx) + + opts.Iterator = NewCollectorIteratorOptions(&tnt, target, spec) + + for _, obj := range sources { + if obj.GetNamespace() == target.GetName() { + continue + } + + // Rejected upfront, before imposing the target Namespace on a copy which could never + // be applied as such. + if err := co.validateClusterScopedObjectAllowed(opts, obj, target); err != nil { + return err + } + + replica := obj.DeepCopy() + if err := sanitize.SanitizeObject(replica, c.Scheme(), co.objectSanitizeOptions); err != nil { + return err + } + + replica.SetNamespace(target.GetName()) + + log.V(4).Info( + "adding replication for namespaced item", + "name", replica.GetName(), + "namespace", replica.GetNamespace(), + "kind", replica.GetKind(), + ) + + if err := co.AddToAccumulation(&tnt, target, opts, spec, replica, "replica", false); err != nil { + return err + } + } + + return co.Collect(ctx, c, opts, &tnt, resourceIndex, spec, target) +} + func (co *Collector) CollectNamespacedItems( ctx context.Context, c client.Client, @@ -383,11 +435,21 @@ func GatherAdditionalMetadata( return labels, annotations } +// Ensures the given object can take part to the accumulation: +// - a cluster-scoped object is accumulated only when the specification allows it; +// - a cluster-scoped object is never accumulated for a target Namespace, since stamping +// the target on it would track the very same object once per Namespace. Beside being +// applied over and over, pruning any of those entries would delete the object the +// remaining Namespaces are still referring to. +// +// Cluster-scoped objects remain replicable through the None and Tenant scopes, which impose +// no target Namespace at all. func (co *Collector) validateClusterScopedObjectAllowed( opts CollectorOptions, obj *unstructured.Unstructured, + ns *corev1.Namespace, ) error { - if opts.AllowClusterScopedObjects { + if opts.AllowClusterScopedObjects && ns == nil { return nil } @@ -396,11 +458,18 @@ func (co *Collector) validateClusterScopedObjectAllowed( return err } - if !isNamespaced { + if isNamespaced { + return nil + } + + if !opts.AllowClusterScopedObjects { return fmt.Errorf("cluster-scoped kind %s/%s is not allowed", obj.GetAPIVersion(), obj.GetKind()) } - return nil + return fmt.Errorf( + "cluster-scoped kind %s/%s cannot be replicated into the Namespace %s", + obj.GetAPIVersion(), obj.GetKind(), ns.GetName(), + ) } // Handles a single generator item. @@ -449,35 +518,47 @@ func (co *Collector) handleRawItem( return obj, nil } +// Builds the selector matching the Namespaces of the given Tenant which are targeted by +// the resource specification, allowing to evaluate a single Namespace without listing +// them all. +func (co *Collector) namespaceSelector( + tnt capsulev1beta2.Tenant, + resource capsulev1beta2.ResourceSpec, +) (labels.Selector, error) { + selector := labels.NewSelector() + + if resource.NamespaceSelector != nil { + var err error + + selector, err = metav1.LabelSelectorAsSelector(resource.NamespaceSelector) + if err != nil { + return nil, fmt.Errorf("cannot create Namespace selector for Namespace filtering and resource replication: %w", err) + } + } + + // Resources can be replicated only on Namespaces belonging to the same Global: + // preventing a boundary cross by enforcing the selection. + tntRequirement, err := labels.NewRequirement(meta.TenantLabel, selection.Equals, []string{tnt.GetName()}) + if err != nil { + return nil, fmt.Errorf("unable to create requirement for Namespace filtering and resource replication: %w", err) + } + + return selector.Add(*tntRequirement), nil +} + func (co *Collector) selectedTenantNamespaces( ctx context.Context, log logr.Logger, tnt capsulev1beta2.Tenant, resource capsulev1beta2.ResourceSpec, ) (ns []*corev1.Namespace, err error) { - // Creating Namespace selector - var selector labels.Selector - - if resource.NamespaceSelector != nil { - selector, err = metav1.LabelSelectorAsSelector(resource.NamespaceSelector) - if err != nil { - log.Error(err, "cannot create Namespace selector for Namespace filtering and resource replication") - - return nil, err - } - } else { - selector = labels.NewSelector() - } - // Resources can be replicated only on Namespaces belonging to the same Global: - // preventing a boundary cross by enforcing the selection. - tntRequirement, err := labels.NewRequirement(meta.TenantLabel, selection.Equals, []string{tnt.GetName()}) + selector, err := co.namespaceSelector(tnt, resource) if err != nil { - log.Error(err, "unable to create requirement for Namespace filtering and resource replication") + log.Error(err, "cannot create selector for Namespace filtering and resource replication") return nil, err } - selector = selector.Add(*tntRequirement) // Selecting the targeted Namespace according to the TenantResource specification. namespaces := corev1.NamespaceList{} if err = co.gatherClient.List(ctx, &namespaces, client.MatchingLabelsSelector{Selector: selector}); err != nil { diff --git a/internal/controllers/resources/collect_test.go b/internal/controllers/resources/collect_test.go index 2fbc1b07..9c96563f 100644 --- a/internal/controllers/resources/collect_test.go +++ b/internal/controllers/resources/collect_test.go @@ -7,7 +7,9 @@ import ( "strings" "testing" + corev1 "k8s.io/api/core/v1" k8smeta "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/runtime/schema" @@ -78,6 +80,55 @@ func TestCollectorAddToAccumulationClusterScopedObjects(t *testing.T) { t.Fatalf("expected object to be accumulated, got %d items", len(acc)) } }) + + t.Run("rejects cluster scoped object targeting a namespace", func(t *testing.T) { + t.Parallel() + + acc := processor.Accumulator{} + obj := newUnstructured("v1", "Namespace", "", "example") + + opts := CollectorOptions{ + Accumulator: acc, + AllowClusterScopedObjects: true, + } + + target := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "tenant-a"}} + + err := collector.AddToAccumulation(nil, target, opts, capsuleResourceSpec(), obj, "test", true) + if err == nil { + t.Fatal("expected error, got nil") + } + + if !strings.Contains(err.Error(), "cannot be replicated into the Namespace tenant-a") { + t.Fatalf("expected a namespaced replication error, got %v", err) + } + + if len(acc) != 0 { + t.Fatalf("expected object not to be accumulated, got %d items", len(acc)) + } + }) + + t.Run("keeps allowing namespaced object targeting a namespace", func(t *testing.T) { + t.Parallel() + + acc := processor.Accumulator{} + obj := newUnstructured("v1", "ConfigMap", "source", "example") + + opts := CollectorOptions{ + Accumulator: acc, + AllowClusterScopedObjects: true, + } + + target := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "tenant-a"}} + + if err := collector.AddToAccumulation(nil, target, opts, capsuleResourceSpec(), obj, "test", true); err != nil { + t.Fatalf("expected no error, got %v", err) + } + + if len(acc) != 1 { + t.Fatalf("expected object to be accumulated, got %d items", len(acc)) + } + }) } func newUnstructured(apiVersion, kind, namespace, name string) *unstructured.Unstructured { diff --git a/internal/controllers/resources/global.go b/internal/controllers/resources/global.go index d0869222..83998913 100644 --- a/internal/controllers/resources/global.go +++ b/internal/controllers/resources/global.go @@ -38,7 +38,6 @@ import ( "github.com/projectcapsule/capsule/pkg/runtime/configuration" tenantresourceindexer "github.com/projectcapsule/capsule/pkg/runtime/indexers/tenantresource" "github.com/projectcapsule/capsule/pkg/runtime/predicates" - "github.com/projectcapsule/capsule/pkg/runtime/sanitize" ) type globalResourceController struct { @@ -50,6 +49,7 @@ type globalResourceController struct { collector Collector configuration configuration.Configuration metrics *metrics.GlobalTenantResourceRecorder + clients impersonatedClientLoader[*capsulev1beta2.GlobalTenantResource] impersonation *cache.ImpersonationCache } @@ -69,6 +69,12 @@ func (r *globalResourceController) SetupWithManager(mgr ctrl.Manager, ctrlConfig mgr.GetAPIReader(), mgr.GetRESTMapper(), ) + r.clients = impersonatedClientLoader[*capsulev1beta2.GlobalTenantResource]{ + client: r.client, + configuration: r.configuration, + impersonation: r.impersonation, + resolve: globalServiceAccount, + } return ctrl.NewControllerManagedBy(mgr). For( @@ -174,7 +180,7 @@ func (r *globalResourceController) Reconcile(ctx context.Context, request reconc // On Deletion these checks are skipped. //nolint:nestif if tntResource.DeletionTimestamp.IsZero() { - if tntResource.Spec.Cordoned != nil && *tntResource.Spec.Cordoned { + if tntResource.Spec.IsCordoned() { log.V(5).Info("global tenant resource cordoned") return reconcile.Result{}, nil @@ -392,7 +398,6 @@ func (r *globalResourceController) reconcile( }) } -//nolint:gocognit func (r *globalResourceController) gatherResources( ctx context.Context, c client.Client, @@ -478,39 +483,14 @@ func (r *globalResourceController) gatherResources( opts.AllowCrossNamespaceSelection = false for _, innerNs := range namespaces { - opts.Iterator = NewCollectorIteratorOptions(&tnt, innerNs, resource) - - for _, obj := range objs { - if obj.GetNamespace() == innerNs.GetName() { - continue - } - - target := obj.DeepCopy() - if err := sanitize.SanitizeObject(target, c.Scheme(), r.collector.objectSanitizeOptions); err != nil { - return err - } - - target.SetNamespace(innerNs.GetName()) - - log.V(4).Info( - "adding replication for namespaced item", - "name", target.GetName(), - "namespace", target.GetNamespace(), - "kind", target.GetKind(), - ) - - if err := r.collector.AddToAccumulation(&tnt, innerNs, opts, resource, target, "replica", false); err != nil { - return err - } - } - - if err := r.collector.Collect( + if err := r.collector.CollectForNamespace( ctx, c, opts, - &tnt, + tnt, strconv.Itoa(resourceIndex), resource, + objs, innerNs, ); err != nil { return err @@ -523,88 +503,22 @@ func (r *globalResourceController) gatherResources( return nil } -//nolint:dupl func (r *globalResourceController) loadClient( ctx context.Context, log logr.Logger, tntResource *capsulev1beta2.GlobalTenantResource, ) (client.Client, error) { - sa := r.impersonatedServiceAccount(ctx, log, tntResource) - if sa == nil { - sa, ns := configuration.ControllerServiceAccount() + c, sa, err := r.clients.Load(ctx, log, tntResource) - tntResource.Status.ServiceAccount = &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ - Name: meta.RFC1123Name(sa), - Namespace: meta.RFC1123SubdomainName(ns), - } + // The resolved identity is posted to the status even along a failure, as it states + // which ServiceAccount the replication was attempted with. + tntResource.Status.ServiceAccount = sa - return r.client, nil - } - - tntResource.Status.ServiceAccount = &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ - Name: sa.Name, - Namespace: sa.Namespace, - } - - re, err := r.configuration.ServiceAccountClient(ctx) if err != nil { - log.Error(err, "failed to load impersonated rest client") - return nil, err } - log.V(5).Info("using impersonation client", "serviceaccount", sa.Name, "namespace", sa.Namespace) - - return r.impersonation.LoadOrCreate(ctx, log, re, r.client.Scheme(), *sa) -} - -func (r *globalResourceController) impersonatedServiceAccount( - ctx context.Context, - log logr.Logger, - tntResource *capsulev1beta2.GlobalTenantResource, -) *meta.NamespacedRFC1123ObjectReferenceWithNamespace { - if sa := tntResource.Spec.ServiceAccount; sa != nil { - name := sa.Name.String() - ns := sa.Namespace.String() - - if name == "" || ns == "" { - log.V(4).Info("serviceAccount reference is set but incomplete; ignoring", - "name", name, "namespace", ns, - ) - - return nil - } - - return &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ - Name: sa.Name, - Namespace: sa.Namespace, - } - } - - cfg := r.configuration.ServiceAccountClientProperties() - - name := cfg.GlobalDefaultServiceAccount.String() - ns := cfg.GlobalDefaultServiceAccountNamespace.String() - - nameSet := name != "" - nsSet := ns != "" - - if nameSet != nsSet { - log.V(2).Info("invalid config: global default service account requires both name and namespace", - "name", name, "namespace", ns, - ) - - return nil - } - - if !nameSet && !nsSet { - return nil - } - - return &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ - Name: cfg.GlobalDefaultServiceAccount, - Namespace: cfg.GlobalDefaultServiceAccountNamespace, - } + return c, nil } func (r *globalResourceController) updateReconcilingStatus(ctx context.Context, instance *capsulev1beta2.GlobalTenantResource) error { diff --git a/internal/controllers/resources/impersonation.go b/internal/controllers/resources/impersonation.go new file mode 100644 index 00000000..2857a794 --- /dev/null +++ b/internal/controllers/resources/impersonation.go @@ -0,0 +1,152 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + + "github.com/go-logr/logr" + "sigs.k8s.io/controller-runtime/pkg/client" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" + "github.com/projectcapsule/capsule/internal/cache" + "github.com/projectcapsule/capsule/pkg/api/meta" + "github.com/projectcapsule/capsule/pkg/runtime/configuration" +) + +// Resolves the ServiceAccount the given replication resource must be replicated with, +// either declared by the resource itself or defaulted by the Capsule configuration. +// A nil result states no impersonation is required. +// +// The resolution is the sole difference between the replication resources, since the +// ServiceAccount reference they declare is scoped differently. +type serviceAccountResolver[T client.Object] func( + cfg configuration.Configuration, + log logr.Logger, + obj T, +) *meta.NamespacedRFC1123ObjectReferenceWithNamespace + +// Loads the client a replication resource must be replicated with, dealing with the +// optional ServiceAccount impersonation. +// +// The type parameter keeps the resolver bound to the resource it can actually resolve, +// as the ServiceAccount reference lives on the resource specific part of the spec. +type impersonatedClientLoader[T client.Object] struct { + client client.Client + configuration configuration.Configuration + impersonation *cache.ImpersonationCache + resolve serviceAccountResolver[T] +} + +// Load returns the client along with the ServiceAccount identity it acts as. +// The identity is always resolved, even along an error and even when no impersonation is +// required, in which case it is the one of the controller itself: callers willing to +// report it on the status can do so unconditionally. +func (l impersonatedClientLoader[T]) Load( + ctx context.Context, + log logr.Logger, + obj T, +) (client.Client, *meta.NamespacedRFC1123ObjectReferenceWithNamespace, error) { + sa := l.resolve(l.configuration, log, obj) + if sa == nil { + // No impersonation required: the controller replicates with its own identity. + name, namespace := configuration.ControllerServiceAccount() + + return l.client, &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ + Name: meta.RFC1123Name(name), + Namespace: meta.RFC1123SubdomainName(namespace), + }, nil + } + + re, err := l.configuration.ServiceAccountClient(ctx) + if err != nil { + log.Error(err, "failed to load impersonated rest client") + + return nil, sa, err + } + + log.V(5).Info("using impersonation client", "serviceaccount", sa.Name, "namespace", sa.Namespace) + + c, err := l.impersonation.LoadOrCreate(ctx, log, re, l.client.Scheme(), *sa) + + return c, sa, err +} + +// Resolves the ServiceAccount of a GlobalTenantResource: being cluster scoped, it must +// declare the Namespace of the ServiceAccount along with its name. +func globalServiceAccount( + cfg configuration.Configuration, + log logr.Logger, + tntResource *capsulev1beta2.GlobalTenantResource, +) *meta.NamespacedRFC1123ObjectReferenceWithNamespace { + if sa := tntResource.Spec.ServiceAccount; sa != nil { + name := sa.Name.String() + ns := sa.Namespace.String() + + if name == "" || ns == "" { + log.V(4).Info("serviceAccount reference is set but incomplete; ignoring", + "name", name, "namespace", ns, + ) + + return nil + } + + return &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ + Name: sa.Name, + Namespace: sa.Namespace, + } + } + + props := cfg.ServiceAccountClientProperties() + + name := props.GlobalDefaultServiceAccount.String() + ns := props.GlobalDefaultServiceAccountNamespace.String() + + nameSet := name != "" + nsSet := ns != "" + + if nameSet != nsSet { + log.V(2).Info("invalid config: global default service account requires both name and namespace", + "name", name, "namespace", ns, + ) + + return nil + } + + if !nameSet && !nsSet { + return nil + } + + return &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ + Name: props.GlobalDefaultServiceAccount, + Namespace: props.GlobalDefaultServiceAccountNamespace, + } +} + +// Resolves the ServiceAccount of a TenantResource: being namespaced, the ServiceAccount +// is always expected to live in the very same Namespace of the resource, preventing a +// Tenant owner from impersonating an identity outside of its boundaries. +func namespacedServiceAccount( + cfg configuration.Configuration, + _ logr.Logger, + tntResource *capsulev1beta2.TenantResource, +) *meta.NamespacedRFC1123ObjectReferenceWithNamespace { + if tntResource.Spec.ServiceAccount != nil { + return &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ + Name: tntResource.Spec.ServiceAccount.Name, + Namespace: meta.RFC1123SubdomainName(tntResource.Namespace), + } + } + + props := cfg.ServiceAccountClientProperties() + + if props.TenantDefaultServiceAccount == "" { + return nil + } + + return &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ + Name: props.TenantDefaultServiceAccount, + Namespace: meta.RFC1123SubdomainName(tntResource.Namespace), + } +} diff --git a/internal/controllers/resources/manager.go b/internal/controllers/resources/manager.go index 05973867..435eeffd 100644 --- a/internal/controllers/resources/manager.go +++ b/internal/controllers/resources/manager.go @@ -22,6 +22,14 @@ func Add( opts utils.ControllerOptions, cache *cache.ImpersonationCache, ) (err error) { + if err = (&NamespaceTrigger{ + log: log.WithName("Global"), + configuration: configuration, + impersonation: cache, + }).SetupWithManager(mgr, opts); err != nil { + return fmt.Errorf("unable to create watcher controller: %w", err) + } + if err = (&globalResourceController{ log: log.WithName("Global"), configuration: configuration, diff --git a/internal/controllers/resources/namespace_trigger.go b/internal/controllers/resources/namespace_trigger.go new file mode 100644 index 00000000..3020a553 --- /dev/null +++ b/internal/controllers/resources/namespace_trigger.go @@ -0,0 +1,476 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + "errors" + "fmt" + "strconv" + "time" + + "github.com/go-logr/logr" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/sets" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" + "sigs.k8s.io/controller-runtime/pkg/client" + ctrllog "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" + "github.com/projectcapsule/capsule/internal/cache" + "github.com/projectcapsule/capsule/internal/controllers/utils" + "github.com/projectcapsule/capsule/pkg/api" + "github.com/projectcapsule/capsule/pkg/api/meta" + "github.com/projectcapsule/capsule/pkg/api/processor" + "github.com/projectcapsule/capsule/pkg/runtime/configuration" + "github.com/projectcapsule/capsule/pkg/runtime/predicates" + tpl "github.com/projectcapsule/capsule/pkg/template" +) + +// NamespaceTrigger is watching for Namespace events, and trigger the +// GlobalTenantResource or TenantResource target the watched object: +// this is used to propagate resources in new freshly created Namespace +// without waiting for the resyncPeriod of the Resources. +type NamespaceTrigger struct { + client client.Client + reader client.Reader + log logr.Logger + configuration configuration.Configuration + impersonation *cache.ImpersonationCache + processor processor.Processor + collector Collector + + globalClients impersonatedClientLoader[*capsulev1beta2.GlobalTenantResource] + namespacedClients impersonatedClientLoader[*capsulev1beta2.TenantResource] + globalStatus scopedStatusPatcher[*capsulev1beta2.GlobalTenantResource] + namespacedStatus scopedStatusPatcher[*capsulev1beta2.TenantResource] +} + +func (r *NamespaceTrigger) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) { + log := ctrllog.FromContext(ctx) + + var namespace corev1.Namespace + if err := r.client.Get(ctx, request.NamespacedName, &namespace); err != nil { + if apierrors.IsNotFound(err) { + log.V(5).Info("Request object not found, could have been deleted after reconcile request") + + return reconcile.Result{}, nil + } + + return reconcile.Result{}, err + } + + tntName, found := namespace.GetLabels()[meta.TenantLabel] + if !found { + log.V(5).Info("cannot retrieve Tenant ownership from Namespace's Tenant label") + + return reconcile.Result{}, nil + } + + var tnt capsulev1beta2.Tenant + if err := r.client.Get(ctx, types.NamespacedName{Name: tntName}, &tnt); err != nil { + return reconcile.Result{}, err + } + + log = log.WithValues("tenant", tnt.GetName()) + + syncErr := errors.Join( + r.replicateNamespacedResources(ctx, log, tnt, &namespace), + r.replicateGlobalResources(ctx, log, tnt, &namespace), + ) + + return reconcile.Result{}, syncErr +} + +func (r *NamespaceTrigger) SetupWithManager(mgr ctrl.Manager, ctrlConfig utils.ControllerOptions) error { + r.client = mgr.GetClient() + r.reader = mgr.GetAPIReader() + + r.processor = processor.Processor{ + Configuration: r.configuration, + GatherClient: mgr.GetAPIReader(), + AllowCrossNamespaceSelection: true, + + Mapper: mgr.GetRESTMapper(), + } + r.collector = NewCollector( + mgr.GetAPIReader(), + mgr.GetRESTMapper(), + ) + r.globalClients = impersonatedClientLoader[*capsulev1beta2.GlobalTenantResource]{ + client: r.client, + configuration: r.configuration, + impersonation: r.impersonation, + resolve: globalServiceAccount, + } + r.namespacedClients = impersonatedClientLoader[*capsulev1beta2.TenantResource]{ + client: r.client, + configuration: r.configuration, + impersonation: r.impersonation, + resolve: namespacedServiceAccount, + } + r.globalStatus = scopedStatusPatcher[*capsulev1beta2.GlobalTenantResource]{ + client: r.client, + reader: r.reader, + factory: func() *capsulev1beta2.GlobalTenantResource { return &capsulev1beta2.GlobalTenantResource{} }, + status: func(o *capsulev1beta2.GlobalTenantResource) *capsulev1beta2.TenantResourceCommonStatus { + return &o.Status.TenantResourceCommonStatus + }, + } + r.namespacedStatus = scopedStatusPatcher[*capsulev1beta2.TenantResource]{ + client: r.client, + reader: r.reader, + factory: func() *capsulev1beta2.TenantResource { return &capsulev1beta2.TenantResource{} }, + status: func(o *capsulev1beta2.TenantResource) *capsulev1beta2.TenantResourceCommonStatus { + return &o.Status.TenantResourceCommonStatus + }, + } + + return ctrl.NewControllerManagedBy(mgr). + Named("NamespaceWatcher"). + For( + &corev1.Namespace{}, + // Only the Namespaces of a Tenant which are created from now on: the already + // existing ones are covered by the full reconciliation of the resources. + builder.WithPredicates( + predicates.LabelPresentPredicate{Label: meta.TenantLabel}, + predicates.NewCreatedAfterPredicate(time.Now()), + ), + ). + WithOptions(ctrlConfig.Runtime.ToControllerOptions()). + Complete(r) +} + +// Replicates into the given Namespace every TenantResource of the Tenant. +func (r *NamespaceTrigger) replicateNamespacedResources( + ctx context.Context, + log logr.Logger, + tnt capsulev1beta2.Tenant, + namespace *corev1.Namespace, +) (syncErr error) { + var tr capsulev1beta2.TenantResourceList + + nsSet := sets.New[string](tnt.Status.Namespaces...) + nsSet.Insert(namespace.GetName()) + + for ns := range nsSet { + var list capsulev1beta2.TenantResourceList + if err := r.client.List(ctx, &list, client.InNamespace(ns)); err != nil { + log.Error(err, "cannot retrieve TenantResourceList", "namespace", ns) + + return err + } + + tr.Items = append(tr.Items, list.Items...) + } + + for _, tntResource := range tr.Items { + ilog := log.WithValues("tenantresource", tntResource.GetName(), "source", tntResource.GetNamespace()) + + if skip, reason := skipScopedReplication(&tntResource, &tntResource.Spec.TenantResourceCommonSpec); skip { + ilog.V(5).Info("TenantResource is not replicating, ignoring", "reason", reason) + + continue + } + + if err := r.replicateNamespaced(ctx, ilog, &tntResource, tnt, namespace); err != nil { + ilog.Error(err, "cannot replicate the TenantResource on the Namespace") + + syncErr = errors.Join(syncErr, err) + } + } + + return syncErr +} + +// Replicates into the given Namespace every GlobalTenantResource selecting the Tenant. +func (r *NamespaceTrigger) replicateGlobalResources( + ctx context.Context, + log logr.Logger, + tnt capsulev1beta2.Tenant, + namespace *corev1.Namespace, +) (syncErr error) { + var gtr capsulev1beta2.GlobalTenantResourceList + if err := r.client.List(ctx, >r); err != nil { + log.Error(err, "cannot retrieve GlobalTenantResourceList") + + return err + } + + for _, tntResource := range gtr.Items { + ilog := log.WithValues("globaltenantresource", tntResource.GetName()) + + selector, err := metav1.LabelSelectorAsSelector(&tntResource.Spec.TenantSelector) + if err != nil { + ilog.Error(err, "cannot create MatchingLabelsSelector for Global filtering") + + continue + } + + if !selector.Matches(labels.Set(tnt.GetLabels())) { + ilog.V(5).Info("Tenant is not selected by the GlobalTenantResource, ignoring") + + continue + } + + if skip, reason := skipGlobalNamespaceReplication(tntResource); skip { + ilog.V(5).Info("GlobalTenantResource is not replicating on a Namespace basis, ignoring", "reason", reason) + + continue + } + + if err := r.replicateGlobal(ctx, ilog, &tntResource, tnt, namespace); err != nil { + ilog.Error(err, "cannot replicate the GlobalTenantResource on the Namespace") + + syncErr = errors.Join(syncErr, err) + } + } + + return syncErr +} + +// States whether the given GlobalTenantResource is out of the Namespace scoped +// replication duties, along with the reason. +func skipGlobalNamespaceReplication(tntResource capsulev1beta2.GlobalTenantResource) (bool, string) { + // Any other scope is not replicating on a per Namespace basis: the accumulated items + // would not be addressable by a Namespace scope at all. + if tntResource.Spec.Scope != api.ResourceScopeNamespace { + return true, "scope is " + string(tntResource.Spec.Scope) + } + + return skipScopedReplication(&tntResource, &tntResource.Spec.TenantResourceCommonSpec) +} + +// Replicates a single GlobalTenantResource into the given Namespace of the given Tenant, +// reporting the outcome on the status without touching the items of the other Namespaces. +func (r *NamespaceTrigger) replicateGlobal( + ctx context.Context, + log logr.Logger, + tntResource *capsulev1beta2.GlobalTenantResource, + tnt capsulev1beta2.Tenant, + namespace *corev1.Namespace, +) error { + // The resolved ServiceAccount is intentionally discarded: posting it to the status is + // a duty of the GlobalTenantResource controller. + c, _, err := r.globalClients.Load(ctx, log, tntResource) + if err != nil { + return fmt.Errorf("failed to load serviceaccount client: %w", err) + } + + scope := processor.Scope{ + Tenant: tnt.GetName(), + Namespace: namespace.GetName(), + } + + acc := processor.Accumulator{} + + // Bailing out on a partial accumulation, as the full reconciliation does: reporting it + // would drop the status entries of the items which could not be gathered, making the + // full reconciliation lose track of them. + if err := r.gatherGlobalResources(ctx, c, log, tntResource, tnt, namespace, acc); err != nil { + return fmt.Errorf("failed to gather resources: %w", err) + } + + owner := meta.GetLooseOwnerReference(tntResource) + + items, reconcileErr := r.processor.ReconcileNamespace( + ctx, + log, + c, + tntResource.Status.ProcessedItems, + acc, + scopedProcessorOptions(tntResource, &tntResource.Spec.TenantResourceCommonSpec, &owner), + scope, + ) + + // The items are reported even along an error, since they carry the outcome of the + // single objects which have been processed. + statusErr := r.globalStatus.Patch(ctx, tntResource, scope, items) + + return errors.Join(reconcileErr, statusErr) +} + +// Replicates a single TenantResource into the given Namespace of the given Tenant, +// reporting the outcome on the status without touching the items of the other Namespaces. +func (r *NamespaceTrigger) replicateNamespaced( + ctx context.Context, + log logr.Logger, + tntResource *capsulev1beta2.TenantResource, + tnt capsulev1beta2.Tenant, + namespace *corev1.Namespace, +) error { + c, _, err := r.namespacedClients.Load(ctx, log, tntResource) + if err != nil { + return fmt.Errorf("failed to load serviceaccount client: %w", err) + } + + scope := processor.Scope{ + Tenant: tnt.GetName(), + Namespace: namespace.GetName(), + } + + acc := processor.Accumulator{} + + if err := r.gatherNamespacedResources(ctx, c, log, tntResource, tnt, namespace, acc); err != nil { + return fmt.Errorf("failed to gather resources: %w", err) + } + + // A TenantResource is namespaced, hence it cannot own objects living in another + // Namespace: no owner reference is set, mirroring the full reconciliation. + items, reconcileErr := r.processor.ReconcileNamespace( + ctx, + log, + c, + tntResource.Status.ProcessedItems, + acc, + scopedProcessorOptions(tntResource, &tntResource.Spec.TenantResourceCommonSpec, nil), + scope, + ) + + statusErr := r.namespacedStatus.Patch(ctx, tntResource, scope, items) + + return errors.Join(reconcileErr, statusErr) +} + +// Accumulates the items of a GlobalTenantResource the given Namespace must be holding. +func (r *NamespaceTrigger) gatherGlobalResources( + ctx context.Context, + c client.Client, + log logr.Logger, + tntResource *capsulev1beta2.GlobalTenantResource, + tnt capsulev1beta2.Tenant, + namespace *corev1.Namespace, + acc processor.Accumulator, +) error { + opts := CollectorOptions{ + Accumulator: acc, + AllowClusterScopedObjects: true, + } + + for resourceIndex, resource := range tntResource.Spec.Resources { + targeted, err := r.targetsNamespace(log, tnt, resource, namespace, resourceIndex) + if err != nil { + return err + } + + if !targeted { + continue + } + + // Sources are loaded cluster-wide, as they can live outside of the target Namespace. + opts.AllowCrossNamespaceSelection = true + + sources, err := r.collector.CollectNamespacedItems(ctx, c, opts, resource, nil, tnt) + if err != nil { + return err + } + + opts.AllowCrossNamespaceSelection = false + + if err := r.collector.CollectForNamespace( + ctx, + c, + opts, + tnt, + strconv.Itoa(resourceIndex), + resource, + sources, + namespace, + ); err != nil { + return err + } + } + + return nil +} + +// States whether the given resource specification is replicating on the given Namespace. +func (r *NamespaceTrigger) targetsNamespace( + log logr.Logger, + tnt capsulev1beta2.Tenant, + resource capsulev1beta2.ResourceSpec, + namespace *corev1.Namespace, + resourceIndex int, +) (bool, error) { + selector, err := r.collector.namespaceSelector(tnt, resource) + if err != nil { + return false, err + } + + if !selector.Matches(labels.Set(namespace.GetLabels())) { + log.V(5).Info("Namespace is not targeted by the resource, ignoring", "resource", resourceIndex) + + return false, nil + } + + return true, nil +} + +// Accumulates the items of a TenantResource the given Namespace must be holding. +func (r *NamespaceTrigger) gatherNamespacedResources( + ctx context.Context, + c client.Client, + log logr.Logger, + tntResource *capsulev1beta2.TenantResource, + tnt capsulev1beta2.Tenant, + namespace *corev1.Namespace, + acc processor.Accumulator, +) error { + // The Namespace has just been created, thus it may not have landed on the Tenant status + // yet: it is a legit replication target nonetheless, and the validator must know about it + // to not reject the items referring to it. + allowed := sets.New[string](tnt.Status.Namespaces...) + allowed.Insert(namespace.GetName()) + + // The very same boundaries of the full reconciliation: a Tenant owner must not be able to + // select cluster scoped objects, nor objects living outside of its own Namespaces. + opts := CollectorOptions{ + Accumulator: acc, + AllowCrossNamespaceSelection: false, + AllowClusterScopedObjects: false, + ValidatorNamespaces: tpl.NewNamespaceValidator(false, allowed), + } + + // The sources of a TenantResource always live in the Namespace it is deployed in. + source := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: tntResource.GetNamespace()}, + } + + for resourceIndex, resource := range tntResource.Spec.Resources { + targeted, err := r.targetsNamespace(log, tnt, resource, namespace, resourceIndex) + if err != nil { + return err + } + + if !targeted { + continue + } + + sources, err := r.collector.CollectNamespacedItems(ctx, c, opts, resource, source, tnt) + if err != nil { + return err + } + + if err := r.collector.CollectForNamespace( + ctx, + c, + opts, + tnt, + strconv.Itoa(resourceIndex), + resource, + sources, + namespace, + ); err != nil { + return err + } + } + + return nil +} diff --git a/internal/controllers/resources/namespaced.go b/internal/controllers/resources/namespaced.go index c428f80e..fa854726 100644 --- a/internal/controllers/resources/namespaced.go +++ b/internal/controllers/resources/namespaced.go @@ -51,6 +51,7 @@ type namespacedResourceController struct { collector Collector configuration configuration.Configuration metrics *metrics.TenantResourceRecorder + clients impersonatedClientLoader[*capsulev1beta2.TenantResource] impersonation *cache.ImpersonationCache } @@ -69,6 +70,12 @@ func (r *namespacedResourceController) SetupWithManager(mgr ctrl.Manager, ctrlCo mgr.GetAPIReader(), mgr.GetRESTMapper(), ) + r.clients = impersonatedClientLoader[*capsulev1beta2.TenantResource]{ + client: r.client, + configuration: r.configuration, + impersonation: r.impersonation, + resolve: namespacedServiceAccount, + } return ctrl.NewControllerManagedBy(mgr). For( @@ -181,7 +188,7 @@ func (r *namespacedResourceController) Reconcile(ctx context.Context, request re // On Deletion these checks are skipped. //nolint:nestif if tntResource.DeletionTimestamp.IsZero() { - if tntResource.Spec.Cordoned != nil && *tntResource.Spec.Cordoned { + if tntResource.Spec.IsCordoned() { log.V(5).Info("tenant resource cordoned") return reconcile.Result{}, nil @@ -531,63 +538,22 @@ func (r *namespacedResourceController) gatherResources( return nil } -//nolint:dupl func (r *namespacedResourceController) loadClient( ctx context.Context, log logr.Logger, tntResource *capsulev1beta2.TenantResource, ) (client.Client, error) { - sa := r.impersonatedServiceAccount(ctx, log, tntResource) - if sa == nil { - sa, ns := configuration.ControllerServiceAccount() + c, sa, err := r.clients.Load(ctx, log, tntResource) - tntResource.Status.ServiceAccount = &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ - Name: meta.RFC1123Name(sa), - Namespace: meta.RFC1123SubdomainName(ns), - } + // The resolved identity is posted to the status even along a failure, as it states + // which ServiceAccount the replication was attempted with. + tntResource.Status.ServiceAccount = sa - return r.client, nil - } - - tntResource.Status.ServiceAccount = &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ - Name: sa.Name, - Namespace: sa.Namespace, - } - - re, err := r.configuration.ServiceAccountClient(ctx) if err != nil { - log.Error(err, "failed to load impersonated rest client") - return nil, err } - log.V(5).Info("using impersonation client", "serviceaccount", sa.Name, "namespace", sa.Namespace) - - return r.impersonation.LoadOrCreate(ctx, log, re, r.client.Scheme(), *sa) -} - -func (r *namespacedResourceController) impersonatedServiceAccount( - ctx context.Context, - log logr.Logger, - tntResource *capsulev1beta2.TenantResource, -) *meta.NamespacedRFC1123ObjectReferenceWithNamespace { - if tntResource.Spec.ServiceAccount != nil { - return &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ - Name: tntResource.Spec.ServiceAccount.Name, - Namespace: meta.RFC1123SubdomainName(tntResource.Namespace), - } - } - - cfg := r.configuration.ServiceAccountClientProperties() - - if cfg.TenantDefaultServiceAccount == "" { - return nil - } - - return &meta.NamespacedRFC1123ObjectReferenceWithNamespace{ - Name: cfg.TenantDefaultServiceAccount, - Namespace: meta.RFC1123SubdomainName(tntResource.Namespace), - } + return c, nil } func (r *namespacedResourceController) updateReconcilingStatus(ctx context.Context, instance *capsulev1beta2.TenantResource) error { diff --git a/internal/controllers/resources/scoped_replication.go b/internal/controllers/resources/scoped_replication.go new file mode 100644 index 00000000..0d84b08b --- /dev/null +++ b/internal/controllers/resources/scoped_replication.go @@ -0,0 +1,108 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + "reflect" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/util/retry" + "sigs.k8s.io/controller-runtime/pkg/client" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" + "github.com/projectcapsule/capsule/pkg/api/meta" + "github.com/projectcapsule/capsule/pkg/api/processor" +) + +// Grafts the processed items of a single Namespace scope on the status of a replication +// resource, leaving the items of every other Namespace untouched. +// +// The type parameter keeps the accessors bound to the resource they belong to, whereas +// the patching itself is shared: the status holding the processed items is the very same +// for the namespaced and the global replication resources. +type scopedStatusPatcher[T client.Object] struct { + client client.Client + reader client.Reader + // Builds an empty instance to read the latest state into. + factory func() T + // Accesses the status shared by every replication resource. + status func(T) *capsulev1beta2.TenantResourceCommonStatus +} + +// Patch merges the given items in the status of the resource, +// reporting them for the given scope only. +// The in memory instance is kept aligned with what has been written. +func (p scopedStatusPatcher[T]) Patch( + ctx context.Context, + instance T, + scope processor.Scope, + items meta.ProcessedItems, +) error { + return retry.RetryOnConflict(retry.DefaultBackoff, func() error { + latest := p.factory() + // The uncached reader is mandatory here: a stale list of processed items would + // resurrect the ones the full reconciliation just pruned. + if err := p.reader.Get(ctx, client.ObjectKeyFromObject(instance), latest); err != nil { + return client.IgnoreNotFound(err) + } + + status := p.status(latest) + original := status.DeepCopy() + + status.ProcessedItems.ReplaceScope(scope.Tenant, scope.Namespace, items) + status.UpdateStats() + + // Only the shared status is compared, since it is the only one being mutated: the + // resource specific part cannot have drifted. + if reflect.DeepEqual(*original, *status) { + return nil + } + + if err := p.client.Status().Update(ctx, latest); err != nil { + return err + } + + // Keep the in-memory object aligned with what we just wrote. + *p.status(instance) = *status + + return nil + }) +} + +// Builds the processor options out of the settings shared by every replication resource. +// +// The field owner must be derived exactly as the full reconciliation does, otherwise the +// two would fight over the server-side apply ownership of the very same objects. +func scopedProcessorOptions( + obj client.Object, + spec *capsulev1beta2.TenantResourceCommonSpec, + owner *metav1.OwnerReference, +) processor.ProcessorOptions { + return processor.ProcessorOptions{ + FieldOwnerPrefix: getFieldOwner(obj.GetName(), obj.GetNamespace()), + Prune: *spec.PruningOnDelete, + Adopt: *spec.Settings.Adopt, + Force: *spec.Settings.Force, + Owner: owner, + } +} + +// States whether the given replication resource is out of the Namespace scoped replication +// duties, along with the reason. +// Replicating the same mechanism of reconciling GlobalTenantResource or TenantResource objects. +func skipScopedReplication( + obj client.Object, + spec *capsulev1beta2.TenantResourceCommonSpec, +) (bool, string) { + if !obj.GetDeletionTimestamp().IsZero() { + return true, "is being deleted" + } + + if spec.IsCordoned() { + return true, "is cordoned" + } + + return false, "" +} diff --git a/pkg/api/meta/processed.go b/pkg/api/meta/processed.go index 47081c1a..3402bd1a 100644 --- a/pkg/api/meta/processed.go +++ b/pkg/api/meta/processed.go @@ -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 +} diff --git a/pkg/api/meta/processed_test.go b/pkg/api/meta/processed_test.go index 6dc49efc..ad55ea1c 100644 --- a/pkg/api/meta/processed_test.go +++ b/pkg/api/meta/processed_test.go @@ -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) + } +} diff --git a/pkg/api/processor/processor.go b/pkg/api/processor/processor.go index e2eebaaf..23f2c9ae 100644 --- a/pkg/api/processor/processor.go +++ b/pkg/api/processor/processor.go @@ -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 +} diff --git a/pkg/api/processor/processor_func.go b/pkg/api/processor/processor_func.go index 2dd4c8c9..3b2c9d84 100644 --- a/pkg/api/processor/processor_func.go +++ b/pkg/api/processor/processor_func.go @@ -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, diff --git a/pkg/api/processor/processor_func_test.go b/pkg/api/processor/processor_func_test.go index f43ee853..86b8e01a 100644 --- a/pkg/api/processor/processor_func_test.go +++ b/pkg/api/processor/processor_func_test.go @@ -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, + } +} diff --git a/pkg/runtime/predicates/created_after.go b/pkg/runtime/predicates/created_after.go new file mode 100644 index 00000000..b4275e69 --- /dev/null +++ b/pkg/runtime/predicates/created_after.go @@ -0,0 +1,49 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package predicates + +import ( + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/event" +) + +// CreatedAfterPredicate passes the creation of the objects which came to life starting from +// the given time, discarding every other event. +// +// Filtering the event type alone is not enough to react to the new objects only: the +// informer replays its initial list as creation events at every start, thus the already +// existing objects would be processed again on each restart. Comparing the creation +// timestamp discards that replay. +// +// The creation timestamp is tracked by the API Server with a second granularity: Since +// is expected to be truncated accordingly, as NewCreatedAfterPredicate does, otherwise +// the objects created within the very same second would be discarded. +type CreatedAfterPredicate struct { + Since metav1.Time +} + +// NewCreatedAfterPredicate builds a predicate passing the objects created starting from the given time. +func NewCreatedAfterPredicate(since time.Time) CreatedAfterPredicate { + return CreatedAfterPredicate{ + Since: metav1.NewTime(since.Truncate(time.Second)), + } +} + +func (p CreatedAfterPredicate) Create(e event.CreateEvent) bool { + if e.Object == nil { + return false + } + + created := e.Object.GetCreationTimestamp() + + return !created.Before(&p.Since) +} + +func (p CreatedAfterPredicate) Delete(event.DeleteEvent) bool { return false } + +func (p CreatedAfterPredicate) Update(event.UpdateEvent) bool { return false } + +func (p CreatedAfterPredicate) Generic(event.GenericEvent) bool { return false } diff --git a/pkg/runtime/predicates/created_after_test.go b/pkg/runtime/predicates/created_after_test.go new file mode 100644 index 00000000..70d2bc77 --- /dev/null +++ b/pkg/runtime/predicates/created_after_test.go @@ -0,0 +1,110 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package predicates_test + +import ( + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/event" + + "github.com/projectcapsule/capsule/pkg/runtime/predicates" +) + +func TestCreatedAfterPredicate_Create(t *testing.T) { + t.Parallel() + + since := time.Date(2026, time.August, 11, 10, 0, 0, 0, time.UTC) + p := predicates.NewCreatedAfterPredicate(since) + + namespace := func(created time.Time) *corev1.Namespace { + return &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", + CreationTimestamp: metav1.NewTime(created), + }, + } + } + + for _, tc := range []struct { + name string + obj *corev1.Namespace + want bool + }{ + { + name: "created afterwards", + obj: namespace(since.Add(time.Second)), + want: true, + }, + { + name: "created within the very same second", + obj: namespace(since), + want: true, + }, + { + name: "already existing", + obj: namespace(since.Add(-time.Hour)), + want: false, + }, + { + name: "no creation timestamp", + obj: &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "example"}}, + want: false, + }, + } { + if got := p.Create(event.CreateEvent{Object: tc.obj}); got != tc.want { + t.Fatalf("%s: expected %v, got %v", tc.name, tc.want, got) + } + } + + if p.Create(event.CreateEvent{}) { + t.Fatal("expected a nil object to be discarded") + } +} + +func TestCreatedAfterPredicate_TruncatesToSecond(t *testing.T) { + t.Parallel() + + // The API Server tracks the creation timestamp with a second granularity: a Namespace + // created right after the start up must not be discarded because of the sub-second + // remainder of the lower bound. + since := time.Date(2026, time.August, 11, 10, 0, 0, 500_000_000, time.UTC) + p := predicates.NewCreatedAfterPredicate(since) + + created := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + CreationTimestamp: metav1.NewTime(since.Truncate(time.Second)), + }, + } + + if !p.Create(event.CreateEvent{Object: created}) { + t.Fatal("expected the object created within the same second to be passed") + } +} + +func TestCreatedAfterPredicate_IgnoresAnyOtherEvent(t *testing.T) { + t.Parallel() + + p := predicates.NewCreatedAfterPredicate(time.Date(2026, time.August, 11, 10, 0, 0, 0, time.UTC)) + + recent := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + CreationTimestamp: metav1.NewTime(time.Date(2026, time.August, 11, 11, 0, 0, 0, time.UTC)), + }, + } + + if p.Update(event.UpdateEvent{ObjectOld: recent, ObjectNew: recent}) { + t.Fatal("expected update events to be discarded") + } + + if p.Delete(event.DeleteEvent{Object: recent}) { + t.Fatal("expected delete events to be discarded") + } + + if p.Generic(event.GenericEvent{Object: recent}) { + t.Fatal("expected generic events to be discarded") + } +}