diff --git a/api/v1beta2/tenantresource_triggers_test.go b/api/v1beta2/tenantresource_triggers_test.go new file mode 100644 index 000000000..1b9aaeba2 --- /dev/null +++ b/api/v1beta2/tenantresource_triggers_test.go @@ -0,0 +1,90 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package v1beta2_test + +import ( + "testing" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" + capruntime "github.com/projectcapsule/capsule/pkg/api/runtime" +) + +func trigger(apiGroups []string, kinds []string, ops ...capsulev1beta2.TriggerOperation) capsulev1beta2.TriggerSpec { + return capsulev1beta2.TriggerSpec{ + VersionKinds: capruntime.VersionKinds{APIGroups: apiGroups, Kinds: kinds}, + Operations: ops, + } +} + +func TestTriggerSpec_MatchesOperation(t *testing.T) { + tests := []struct { + name string + spec capsulev1beta2.TriggerSpec + op capsulev1beta2.TriggerOperation + want bool + }{ + { + name: "empty operations matches every op", + spec: trigger(nil, []string{"Secret"}), + op: capsulev1beta2.TriggerOperationDelete, + want: true, + }, + { + name: "listed operation matches", + spec: trigger(nil, []string{"Secret"}, capsulev1beta2.TriggerOperationCreate, capsulev1beta2.TriggerOperationUpdate), + op: capsulev1beta2.TriggerOperationUpdate, + want: true, + }, + { + name: "unlisted operation does not match", + spec: trigger(nil, []string{"Secret"}, capsulev1beta2.TriggerOperationCreate), + op: capsulev1beta2.TriggerOperationDelete, + want: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := tc.spec.MatchesOperation(tc.op); got != tc.want { + t.Fatalf("MatchesOperation(%q) = %v, want %v", tc.op, got, tc.want) + } + }) + } +} + +func TestTriggerVersionKinds(t *testing.T) { + spec := capsulev1beta2.TenantResourceCommonSpec{ + Triggers: []capsulev1beta2.TriggerSpec{ + // Empty apiGroups means core v1. + trigger(nil, []string{"Secret"}), + // Duplicate selector must be de-duplicated. + trigger([]string{"v1"}, []string{"Secret"}, capsulev1beta2.TriggerOperationUpdate), + // Concrete group/version. + trigger([]string{"apps/v1"}, []string{"Deployment"}), + // Bare group expands to a version-less selector. + trigger([]string{"batch"}, []string{"Job", "CronJob"}), + // Empty kind must be dropped. + trigger([]string{"v1"}, []string{""}), + }, + } + + got := spec.TriggerVersionKinds() + + want := map[capruntime.VersionKind]struct{}{ + {APIVersion: "", Kind: "Secret"}: {}, + {APIVersion: "apps/v1", Kind: "Deployment"}: {}, + {APIVersion: "batch/*", Kind: "Job"}: {}, + {APIVersion: "batch/*", Kind: "CronJob"}: {}, + } + + if len(got) != len(want) { + t.Fatalf("expected %d selectors, got %d: %v", len(want), len(got), got) + } + + for _, vk := range got { + if _, ok := want[vk]; !ok { + t.Fatalf("unexpected selector %+v", vk) + } + } +} diff --git a/api/v1beta2/tenantresource_types.go b/api/v1beta2/tenantresource_types.go index 3b9b61195..2cdad79b5 100644 --- a/api/v1beta2/tenantresource_types.go +++ b/api/v1beta2/tenantresource_types.go @@ -4,11 +4,17 @@ package v1beta2 import ( + "slices" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" "github.com/projectcapsule/capsule/pkg/api" "github.com/projectcapsule/capsule/pkg/api/meta" + capruntime "github.com/projectcapsule/capsule/pkg/api/runtime" + "github.com/projectcapsule/capsule/pkg/runtime/selectors" tpl "github.com/projectcapsule/capsule/pkg/template" ) @@ -58,6 +64,110 @@ type TenantResourceCommonSpec struct { Cordoned *bool `json:"cordoned,omitempty"` // Defines the rules to select targeting Namespace, along with the objects that must be replicated. Resources []ResourceSpec `json:"resources"` + // Triggers re-render this resource (near-)immediately when matching cluster + // objects change, instead of only waiting for resyncPeriod. This lets you keep + // a high resyncPeriod while still reacting quickly to changes of the objects + // the rendering depends on (e.g. Secrets or ServiceAccounts referenced through + // a resource's context). Each trigger installs a metadata-only watch per + // referenced kind; a watch is torn down automatically once no resource + // references its kind anymore. + // +optional + // +kubebuilder:validation:MaxItems=100 + Triggers []TriggerSpec `json:"triggers,omitempty"` +} + +// TriggerOperation is the object lifecycle event a trigger reacts to. +// +kubebuilder:validation:Enum=CREATE;UPDATE;DELETE +type TriggerOperation string + +const ( + // TriggerOperationCreate reacts to creations of matching objects. + TriggerOperationCreate TriggerOperation = "CREATE" + // TriggerOperationUpdate reacts to updates of matching objects. + TriggerOperationUpdate TriggerOperation = "UPDATE" + // TriggerOperationDelete reacts to deletions of matching objects. + TriggerOperationDelete TriggerOperation = "DELETE" +) + +// TriggerSpec declares the cluster object kinds whose changes cause the owning +// TenantResource / GlobalTenantResource to be re-rendered. +// +// Wildcards are rejected: every kind selected by a trigger is armed as a +// dedicated watch, so the selection must be a bounded, concrete set. +// +// +kubebuilder:validation:XValidation:rule="!self.kinds.exists(k, k.contains('*'))",message="wildcard kinds are not supported in triggers" +// +kubebuilder:validation:XValidation:rule="!has(self.apiGroups) || !self.apiGroups.exists(g, g.contains('*'))",message="wildcard apiGroups are not supported in triggers" +type TriggerSpec struct { + capruntime.VersionKinds `json:",inline"` + + // Operations that cause a re-render. When empty, all operations + // (CREATE, UPDATE and DELETE) are considered. + // +optional + Operations []TriggerOperation `json:"operations,omitempty"` + // Selector narrows the trigger to objects whose labels match. When omitted, + // every object of the referenced kind matches. + // +optional + Selector *metav1.LabelSelector `json:"selector,omitempty"` + // NamespaceSelector narrows the trigger to objects living in namespaces whose + // labels match. It is only honored for the cluster-scoped GlobalTenantResource; + // for the namespaced TenantResource it is ignored, as the trigger is always + // scoped to the namespaces of the owning Tenant. + // +optional + NamespaceSelector *metav1.LabelSelector `json:"namespaceSelector,omitempty"` +} + +// MatchesOperation reports whether the trigger reacts to the given operation. +// An empty operation list matches every operation. +func (t TriggerSpec) MatchesOperation(op TriggerOperation) bool { + if len(t.Operations) == 0 { + return true + } + + return slices.Contains(t.Operations, op) +} + +// Matches reports whether the trigger reacts to a change of the given kind and +// operation whose object carries the given labels. Namespace scoping +// (NamespaceSelector, tenant scoping) is consumer policy and not evaluated here. +func (t TriggerSpec) Matches(gvk schema.GroupVersionKind, op TriggerOperation, lbls map[string]string) bool { + if !t.MatchesGroupVersionKind(gvk) || !t.MatchesOperation(op) { + return false + } + + if t.Selector == nil { + return true + } + + ok, err := selectors.MatchesSelector(labels.Set(lbls), *t.Selector) + + return err == nil && ok +} + +// TriggerVersionKinds returns the de-duplicated set of kind selectors +// referenced by the resource's triggers. Selectors without a concrete version +// (e.g. apiGroups: ["apps"]) are resolved to a watchable GroupVersionKind by +// the trigger watch manager via the REST mapper. +func (s *TenantResourceCommonSpec) TriggerVersionKinds() []capruntime.VersionKind { + seen := make(map[capruntime.VersionKind]struct{}, len(s.Triggers)) + out := make([]capruntime.VersionKind, 0, len(s.Triggers)) + + for _, t := range s.Triggers { + for _, vk := range t.VersionKinds.VersionKinds() { + if vk.Kind == "" { + continue + } + + if _, ok := seen[vk]; ok { + continue + } + + seen[vk] = struct{}{} + + out = append(out, vk) + } + } + + return out } type TenantResourceCommonSpecSettings struct { diff --git a/api/v1beta2/zz_generated.deepcopy.go b/api/v1beta2/zz_generated.deepcopy.go index a5d24bc43..a9253fe42 100644 --- a/api/v1beta2/zz_generated.deepcopy.go +++ b/api/v1beta2/zz_generated.deepcopy.go @@ -1989,6 +1989,13 @@ func (in *TenantResourceCommonSpec) DeepCopyInto(out *TenantResourceCommonSpec) (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.Triggers != nil { + in, out := &in.Triggers, &out.Triggers + *out = make([]TriggerSpec, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TenantResourceCommonSpec. @@ -2391,3 +2398,34 @@ func (in *TenantStatusRuleStatusItem) DeepCopy() *TenantStatusRuleStatusItem { in.DeepCopyInto(out) return out } + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *TriggerSpec) DeepCopyInto(out *TriggerSpec) { + *out = *in + in.VersionKinds.DeepCopyInto(&out.VersionKinds) + if in.Operations != nil { + in, out := &in.Operations, &out.Operations + *out = make([]TriggerOperation, len(*in)) + copy(*out, *in) + } + if in.Selector != nil { + in, out := &in.Selector, &out.Selector + *out = new(metav1.LabelSelector) + (*in).DeepCopyInto(*out) + } + if in.NamespaceSelector != nil { + in, out := &in.NamespaceSelector, &out.NamespaceSelector + *out = new(metav1.LabelSelector) + (*in).DeepCopyInto(*out) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TriggerSpec. +func (in *TriggerSpec) DeepCopy() *TriggerSpec { + if in == nil { + return nil + } + out := new(TriggerSpec) + in.DeepCopyInto(out) + return out +} diff --git a/charts/capsule/crds/capsule.clastix.io_globaltenantresources.yaml b/charts/capsule/crds/capsule.clastix.io_globaltenantresources.yaml index 4388e30b4..e7b29233f 100644 --- a/charts/capsule/crds/capsule.clastix.io_globaltenantresources.yaml +++ b/charts/capsule/crds/capsule.clastix.io_globaltenantresources.yaml @@ -479,6 +479,175 @@ spec: type: object type: object x-kubernetes-map-type: atomic + triggers: + description: |- + Triggers re-render this resource (near-)immediately when matching cluster + objects change, instead of only waiting for resyncPeriod. This lets you keep + a high resyncPeriod while still reacting quickly to changes of the objects + the rendering depends on (e.g. Secrets or ServiceAccounts referenced through + a resource's context). Each trigger installs a metadata-only watch per + referenced kind; a watch is torn down automatically once no resource + references its kind anymore. + items: + description: |- + TriggerSpec declares the cluster object kinds whose changes cause the owning + TenantResource / GlobalTenantResource to be re-rendered. + + Wildcards are rejected: every kind selected by a trigger is armed as a + dedicated watch, so the selection must be a bounded, concrete set. + properties: + apiGroups: + description: |- + API groups or API group/version selectors of the referents. + + Empty or omitted APIGroups means the core Kubernetes API version "v1". + Use "*" to match all API groups and versions. + + Examples: + - [] or [""] means core "v1". + - ["v1"] means core "v1". + - ["apps"] means any version in the "apps" API group. + - ["apps/v1"] means only "apps/v1". + - ["apps", "batch/v1"] means any "apps" version and "batch/v1". + - ["*"] means all API groups and versions. + items: + maxLength: 317 + type: string + maxItems: 100 + type: array + kinds: + description: |- + Kinds of the referents. + + Use "*" to match all kinds. + items: + maxLength: 63 + minLength: 1 + type: string + maxItems: 100 + minItems: 1 + type: array + namespaceSelector: + description: |- + NamespaceSelector narrows the trigger to objects living in namespaces whose + labels match. It is only honored for the cluster-scoped GlobalTenantResource; + for the namespaced TenantResource it is ignored, as the trigger is always + scoped to the namespaces of the owning Tenant. + properties: + matchExpressions: + description: matchExpressions is a list of label selector + requirements. The requirements are ANDed. + items: + description: |- + A label selector requirement is a selector that contains values, a key, and an operator that + relates the key and values. + properties: + key: + description: key is the label key that the selector + applies to. + type: string + operator: + description: |- + operator represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: |- + values is an array of string values. If the operator is In or NotIn, + the values array must be non-empty. If the operator is Exists or DoesNotExist, + the values array must be empty. This array is replaced during a strategic + merge patch. + items: + type: string + type: array + x-kubernetes-list-type: atomic + required: + - key + - operator + type: object + type: array + x-kubernetes-list-type: atomic + matchLabels: + additionalProperties: + type: string + description: |- + matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels + map is equivalent to an element of matchExpressions, whose key field is "key", the + operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + x-kubernetes-map-type: atomic + operations: + description: |- + Operations that cause a re-render. When empty, all operations + (CREATE, UPDATE and DELETE) are considered. + items: + description: TriggerOperation is the object lifecycle event + a trigger reacts to. + enum: + - CREATE + - UPDATE + - DELETE + type: string + type: array + selector: + description: |- + Selector narrows the trigger to objects whose labels match. When omitted, + every object of the referenced kind matches. + properties: + matchExpressions: + description: matchExpressions is a list of label selector + requirements. The requirements are ANDed. + items: + description: |- + A label selector requirement is a selector that contains values, a key, and an operator that + relates the key and values. + properties: + key: + description: key is the label key that the selector + applies to. + type: string + operator: + description: |- + operator represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: |- + values is an array of string values. If the operator is In or NotIn, + the values array must be non-empty. If the operator is Exists or DoesNotExist, + the values array must be empty. This array is replaced during a strategic + merge patch. + items: + type: string + type: array + x-kubernetes-list-type: atomic + required: + - key + - operator + type: object + type: array + x-kubernetes-list-type: atomic + matchLabels: + additionalProperties: + type: string + description: |- + matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels + map is equivalent to an element of matchExpressions, whose key field is "key", the + operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + x-kubernetes-map-type: atomic + required: + - kinds + type: object + x-kubernetes-validations: + - message: wildcard kinds are not supported in triggers + rule: '!self.kinds.exists(k, k.contains(''*''))' + - message: wildcard apiGroups are not supported in triggers + rule: '!has(self.apiGroups) || !self.apiGroups.exists(g, g.contains(''*''))' + maxItems: 100 + type: array required: - resources - resyncPeriod diff --git a/charts/capsule/crds/capsule.clastix.io_rulestatuses.yaml b/charts/capsule/crds/capsule.clastix.io_rulestatuses.yaml index b8cfcaa02..7f669f94b 100644 --- a/charts/capsule/crds/capsule.clastix.io_rulestatuses.yaml +++ b/charts/capsule/crds/capsule.clastix.io_rulestatuses.yaml @@ -135,7 +135,9 @@ spec: - ["apps", "batch/v1"] means any "apps" version and "batch/v1". - ["*"] means all API groups and versions. items: + maxLength: 317 type: string + maxItems: 100 type: array kinds: description: |- @@ -143,8 +145,10 @@ spec: Use "*" to match all kinds. items: + maxLength: 63 minLength: 1 type: string + maxItems: 100 minItems: 1 type: array labels: @@ -540,7 +544,9 @@ spec: - ["apps", "batch/v1"] means any "apps" version and "batch/v1". - ["*"] means all API groups and versions. items: + maxLength: 317 type: string + maxItems: 100 type: array kinds: description: |- @@ -548,8 +554,10 @@ spec: Use "*" to match all kinds. items: + maxLength: 63 minLength: 1 type: string + maxItems: 100 minItems: 1 type: array labels: @@ -883,7 +891,9 @@ spec: - ["apps", "batch/v1"] means any "apps" version and "batch/v1". - ["*"] means all API groups and versions. items: + maxLength: 317 type: string + maxItems: 100 type: array kinds: description: |- @@ -891,8 +901,10 @@ spec: Use "*" to match all kinds. items: + maxLength: 63 minLength: 1 type: string + maxItems: 100 minItems: 1 type: array labels: diff --git a/charts/capsule/crds/capsule.clastix.io_tenantresources.yaml b/charts/capsule/crds/capsule.clastix.io_tenantresources.yaml index 840a09206..f8fdf6247 100644 --- a/charts/capsule/crds/capsule.clastix.io_tenantresources.yaml +++ b/charts/capsule/crds/capsule.clastix.io_tenantresources.yaml @@ -415,6 +415,175 @@ spec: You may create collisions with this. type: boolean type: object + triggers: + description: |- + Triggers re-render this resource (near-)immediately when matching cluster + objects change, instead of only waiting for resyncPeriod. This lets you keep + a high resyncPeriod while still reacting quickly to changes of the objects + the rendering depends on (e.g. Secrets or ServiceAccounts referenced through + a resource's context). Each trigger installs a metadata-only watch per + referenced kind; a watch is torn down automatically once no resource + references its kind anymore. + items: + description: |- + TriggerSpec declares the cluster object kinds whose changes cause the owning + TenantResource / GlobalTenantResource to be re-rendered. + + Wildcards are rejected: every kind selected by a trigger is armed as a + dedicated watch, so the selection must be a bounded, concrete set. + properties: + apiGroups: + description: |- + API groups or API group/version selectors of the referents. + + Empty or omitted APIGroups means the core Kubernetes API version "v1". + Use "*" to match all API groups and versions. + + Examples: + - [] or [""] means core "v1". + - ["v1"] means core "v1". + - ["apps"] means any version in the "apps" API group. + - ["apps/v1"] means only "apps/v1". + - ["apps", "batch/v1"] means any "apps" version and "batch/v1". + - ["*"] means all API groups and versions. + items: + maxLength: 317 + type: string + maxItems: 100 + type: array + kinds: + description: |- + Kinds of the referents. + + Use "*" to match all kinds. + items: + maxLength: 63 + minLength: 1 + type: string + maxItems: 100 + minItems: 1 + type: array + namespaceSelector: + description: |- + NamespaceSelector narrows the trigger to objects living in namespaces whose + labels match. It is only honored for the cluster-scoped GlobalTenantResource; + for the namespaced TenantResource it is ignored, as the trigger is always + scoped to the namespaces of the owning Tenant. + properties: + matchExpressions: + description: matchExpressions is a list of label selector + requirements. The requirements are ANDed. + items: + description: |- + A label selector requirement is a selector that contains values, a key, and an operator that + relates the key and values. + properties: + key: + description: key is the label key that the selector + applies to. + type: string + operator: + description: |- + operator represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: |- + values is an array of string values. If the operator is In or NotIn, + the values array must be non-empty. If the operator is Exists or DoesNotExist, + the values array must be empty. This array is replaced during a strategic + merge patch. + items: + type: string + type: array + x-kubernetes-list-type: atomic + required: + - key + - operator + type: object + type: array + x-kubernetes-list-type: atomic + matchLabels: + additionalProperties: + type: string + description: |- + matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels + map is equivalent to an element of matchExpressions, whose key field is "key", the + operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + x-kubernetes-map-type: atomic + operations: + description: |- + Operations that cause a re-render. When empty, all operations + (CREATE, UPDATE and DELETE) are considered. + items: + description: TriggerOperation is the object lifecycle event + a trigger reacts to. + enum: + - CREATE + - UPDATE + - DELETE + type: string + type: array + selector: + description: |- + Selector narrows the trigger to objects whose labels match. When omitted, + every object of the referenced kind matches. + properties: + matchExpressions: + description: matchExpressions is a list of label selector + requirements. The requirements are ANDed. + items: + description: |- + A label selector requirement is a selector that contains values, a key, and an operator that + relates the key and values. + properties: + key: + description: key is the label key that the selector + applies to. + type: string + operator: + description: |- + operator represents a key's relationship to a set of values. + Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: |- + values is an array of string values. If the operator is In or NotIn, + the values array must be non-empty. If the operator is Exists or DoesNotExist, + the values array must be empty. This array is replaced during a strategic + merge patch. + items: + type: string + type: array + x-kubernetes-list-type: atomic + required: + - key + - operator + type: object + type: array + x-kubernetes-list-type: atomic + matchLabels: + additionalProperties: + type: string + description: |- + matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels + map is equivalent to an element of matchExpressions, whose key field is "key", the + operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + x-kubernetes-map-type: atomic + required: + - kinds + type: object + x-kubernetes-validations: + - message: wildcard kinds are not supported in triggers + rule: '!self.kinds.exists(k, k.contains(''*''))' + - message: wildcard apiGroups are not supported in triggers + rule: '!has(self.apiGroups) || !self.apiGroups.exists(g, g.contains(''*''))' + maxItems: 100 + type: array required: - resources - resyncPeriod diff --git a/charts/capsule/crds/capsule.clastix.io_tenants.yaml b/charts/capsule/crds/capsule.clastix.io_tenants.yaml index 75c696e34..3b6f4ac45 100644 --- a/charts/capsule/crds/capsule.clastix.io_tenants.yaml +++ b/charts/capsule/crds/capsule.clastix.io_tenants.yaml @@ -2583,7 +2583,9 @@ spec: - ["apps", "batch/v1"] means any "apps" version and "batch/v1". - ["*"] means all API groups and versions. items: + maxLength: 317 type: string + maxItems: 100 type: array kinds: description: |- @@ -2591,8 +2593,10 @@ spec: Use "*" to match all kinds. items: + maxLength: 63 minLength: 1 type: string + maxItems: 100 minItems: 1 type: array labels: diff --git a/e2e/replications_triggers_test.go b/e2e/replications_triggers_test.go new file mode 100644 index 000000000..40e1282f6 --- /dev/null +++ b/e2e/replications_triggers_test.go @@ -0,0 +1,164 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package e2e + +import ( + "context" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" + "github.com/projectcapsule/capsule/pkg/api" + apimeta "github.com/projectcapsule/capsule/pkg/api/meta" + "github.com/projectcapsule/capsule/pkg/api/rbac" + capruntime "github.com/projectcapsule/capsule/pkg/api/runtime" + "github.com/projectcapsule/capsule/pkg/template" +) + +var _ = Describe("GlobalTenantResource triggers", Label("replications", "triggers", "globaltenantresource"), func() { + ctx := context.Background() + + It("re-renders before resyncPeriod when a watched Secret changes", func() { + owner := rbac.UserSpec{Name: "e2e-trg-owner", Kind: rbac.OwnerKind("User")} + + tenant := &capsulev1beta2.Tenant{ + ObjectMeta: metav1.ObjectMeta{ + Name: "e2e-trg-tenant", + Labels: map[string]string{"trigger": "e2e"}, + }, + Spec: capsulev1beta2.TenantSpec{ + Owners: rbac.OwnerListSpec{{ + CoreOwnerSpec: rbac.CoreOwnerSpec{UserSpec: owner}, + }}, + }, + } + + EventuallyCreation(func() error { + tenant.ResourceVersion = "" + + return k8sClient.Create(ctx, tenant) + }).Should(Succeed()) + DeferCleanup(func() { EventuallyDeletion(tenant) }) + TenantReady(tenant, metav1.ConditionTrue, defaultTimeoutInterval) + + tenantNs := "e2e-trg-ns" + namespace := NewNamespace(tenantNs, map[string]string{apimeta.TenantLabel: tenant.GetName()}) + NamespaceCreation(namespace, owner, defaultTimeoutInterval).Should(Succeed()) + DeferCleanup(func() { ForceDeleteNamespace(ctx, tenantNs) }) + + sourceNs := "e2e-trg-source" + EventuallyCreation(func() error { + return k8sClient.Create(ctx, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: sourceNs}}) + }).Should(Succeed()) + DeferCleanup(func() { ForceDeleteNamespace(ctx, sourceNs) }) + + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: "trg-secret", + Namespace: sourceNs, + Labels: map[string]string{ + "pullsecret.company.com": "true", + "revision": "1", + }, + }, + Type: corev1.SecretTypeOpaque, + StringData: map[string]string{"key": "value"}, + } + EventuallyCreation(func() error { return k8sClient.Create(ctx, secret) }).Should(Succeed()) + + gtr := &capsulev1beta2.GlobalTenantResource{ + ObjectMeta: metav1.ObjectMeta{ + Name: "e2e-trg-gtr", + Labels: map[string]string{"e2e.capsule.dev/test-suite": "true"}, + }, + Spec: capsulev1beta2.GlobalTenantResourceSpec{ + Scope: api.ResourceScopeNamespace, + TenantSelector: metav1.LabelSelector{MatchLabels: map[string]string{"trigger": "e2e"}}, + TenantResourceCommonSpec: capsulev1beta2.TenantResourceCommonSpec{ + // Deliberately high: a passing test proves the re-render is + // driven by the trigger, not by the resync. + ResyncPeriod: metav1.Duration{Duration: 10 * time.Minute}, + PruningOnDelete: ptr.To(true), + Triggers: []capsulev1beta2.TriggerSpec{{ + VersionKinds: capruntime.VersionKinds{Kinds: []string{"Secret"}}, + Operations: []capsulev1beta2.TriggerOperation{capsulev1beta2.TriggerOperationUpdate}, + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"pullsecret.company.com": "true"}, + }, + }}, + Resources: []capsulev1beta2.ResourceSpec{{ + Context: &template.TemplateContext{ + Resources: []*template.TemplateResourceReference{{ + Index: "secrets", + ResourceReference: template.ResourceReference{ + VersionKind: capruntime.VersionKind{APIVersion: "v1", Kind: "Secret"}, + Namespace: sourceNs, + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"pullsecret.company.com": "true"}, + }, + }, + }}, + }, + Generators: []capsulev1beta2.TemplateItemSpec{{ + MissingKey: "error", + Template: `--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: trg-cm +data: + revision: '{{ (index $.secrets 0).metadata.labels.revision }}' +`, + }}, + }}, + }, + }, + } + + EventuallyCreation(func() error { return k8sClient.Create(ctx, gtr) }).Should(Succeed()) + DeferCleanup(func() { _ = k8sClient.Delete(ctx, gtr) }) + + // Initial render carries the secret's current revision label. + expectConfigMapData(tenantNs, "trg-cm", map[string]string{"revision": "1"}) + + // The resource advertises its armed watches through a status condition. + Eventually(func(g Gomega) { + current := &capsulev1beta2.GlobalTenantResource{} + g.Expect(k8sClient.Get(ctx, client.ObjectKey{Name: gtr.Name}, current)).To(Succeed()) + + cond := current.Status.Conditions.GetConditionByType(apimeta.TriggersCondition) + g.Expect(cond).NotTo(BeNil()) + g.Expect(cond.Status).To(Equal(metav1.ConditionTrue)) + }, defaultTimeoutInterval, defaultPollInterval).Should(Succeed()) + + // Mutate the watched Secret. + Eventually(func(g Gomega) { + current := &corev1.Secret{} + g.Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "trg-secret", Namespace: sourceNs}, current)).To(Succeed()) + + if current.Labels == nil { + current.Labels = map[string]string{} + } + + current.Labels["revision"] = "2" + + g.Expect(k8sClient.Update(ctx, current)).To(Succeed()) + }, defaultTimeoutInterval, defaultPollInterval).Should(Succeed()) + + // The rendered ConfigMap must reflect the change well before the 10m resync. + Eventually(func(g Gomega) { + cm := &corev1.ConfigMap{} + g.Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "trg-cm", Namespace: tenantNs}, cm)).To(Succeed()) + g.Expect(cm.Data).To(HaveKeyWithValue("revision", "2")) + }, defaultTimeoutInterval, defaultPollInterval).Should(Succeed()) + }) +}) diff --git a/internal/controllers/resources/global.go b/internal/controllers/resources/global.go index af6898570..907ccca26 100644 --- a/internal/controllers/resources/global.go +++ b/internal/controllers/resources/global.go @@ -25,6 +25,7 @@ import ( ctrllog "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" + "sigs.k8s.io/controller-runtime/pkg/source" capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" "github.com/projectcapsule/capsule/internal/cache" @@ -37,6 +38,7 @@ import ( "github.com/projectcapsule/capsule/pkg/runtime/configuration" "github.com/projectcapsule/capsule/pkg/runtime/predicates" "github.com/projectcapsule/capsule/pkg/runtime/sanitize" + "github.com/projectcapsule/capsule/pkg/runtime/watch" ) type globalResourceController struct { @@ -50,6 +52,9 @@ type globalResourceController struct { metrics *metrics.GlobalTenantResourceRecorder impersonation *cache.ImpersonationCache + + triggers *watch.Manager + triggerSrc source.Source } func (r *globalResourceController) SetupWithManager(mgr ctrl.Manager, ctrlConfig utils.ControllerOptions) error { @@ -94,6 +99,7 @@ func (r *globalResourceController) SetupWithManager(mgr ctrl.Manager, ctrlConfig &capsulev1beta2.Tenant{}, handler.EnqueueRequestsFromMapFunc(r.enqueueRequestFromTenant), ). + WatchesRawSource(r.triggerSrc). WithOptions(ctrlConfig.Runtime.ToControllerOptions()). Complete(r) } @@ -110,6 +116,11 @@ func (r *globalResourceController) Reconcile(ctx context.Context, request reconc r.metrics.DeleteMetrics(request.Name) + // Release any trigger watches this resource was keeping alive. + if serr := r.triggers.Sync(ctx, globalTriggerOwnerKey(request.Name), nil); serr != nil { + log.V(2).Error(serr, "failed to release trigger watches") + } + return reconcile.Result{}, nil } @@ -128,10 +139,14 @@ func (r *globalResourceController) Reconcile(ctx context.Context, request reconc var statusErr error + var triggerCond meta.Condition + //nolint:dupl defer func() { meta.RemoveReconcileTriggerAnnotation(tntResource) + tntResource.Status.Conditions.UpdateConditionByType(triggerCond) + reconcileErr := err if statusErr != nil { reconcileErr = statusErr @@ -168,6 +183,17 @@ func (r *globalResourceController) Reconcile(ctx context.Context, request reconc err = nil }() + // Arm (or, while terminating, release) the dynamic trigger watches for the + // kinds this resource depends on, independent of cordon/dependency state. + // No tenant scoping: a GlobalTenantResource may watch objects in non-tenant + // namespaces (e.g. a replication source), which never carry the tenant + // label, and cluster-scoped kinds are never stamped at all. Narrowing to + // label-carrying objects would silently blind those triggers. + triggerCond = syncTriggers( + ctx, r.triggers, globalTriggerOwnerKey(tntResource.Name), + tntResource, tntResource.Spec.TenantResourceCommonSpec, false, log, + ) + // On Deletion these checks are skipped. //nolint:nestif if tntResource.DeletionTimestamp.IsZero() { diff --git a/internal/controllers/resources/manager.go b/internal/controllers/resources/manager.go index 059738675..eaf3df193 100644 --- a/internal/controllers/resources/manager.go +++ b/internal/controllers/resources/manager.go @@ -13,6 +13,7 @@ import ( "github.com/projectcapsule/capsule/internal/controllers/utils" "github.com/projectcapsule/capsule/internal/metrics" "github.com/projectcapsule/capsule/pkg/runtime/configuration" + "github.com/projectcapsule/capsule/pkg/runtime/watch" ) func Add( @@ -22,12 +23,51 @@ func Add( opts utils.ControllerOptions, cache *cache.ImpersonationCache, ) (err error) { + // Shared, reference-counted watch manager driving the dynamic trigger + // informers for both controllers. Each watched kind runs in its own + // metadata-only cache (never the shared manager cache): the factory strips + // managedFields and the last-applied annotation, which the shared cache + // cannot (SSA ownership reads during prune/adoption need managedFields). + // A kind gets at most two informers: one narrowed server-side to objects + // carrying the tenant label (shared by every namespaced TenantResource) + // and one unfiltered (GlobalTenantResources); trigger selectors are + // matched in the sinks. Added to the manager, the caches run leader-gated, + // alongside the reconcilers that arm watches. + triggers := watch.NewManager(watch.MetadataCacheFactory(mgr), mgr.GetClient(), mgr.GetRESTMapper(), log.WithName("Triggers")) + if err = mgr.Add(triggers); err != nil { + return fmt.Errorf("unable to register trigger watch manager: %w", err) + } + + // Each sink enqueues matching owner keys directly into its controller's + // workqueue via a triggerSource. The workqueue dedups by owner key, so bursts + // to a single owner collapse to one reconcile. This does not pace a + // self-trigger loop from a non-deterministic template (output changes every + // apply -> new watch event -> re-render): that is a misconfiguration bounded + // only by resyncPeriod. Deterministic templates converge on their own, because + // a no-op SSA apply produces no resourceVersion bump and thus no event. + globalSource := &triggerSource{} + namespacedSource := &triggerSource{} + + triggers.RegisterSink(&globalTriggerSink{ + reader: mgr.GetClient(), + enqueue: globalSource.enqueue, + log: log.WithName("Triggers").WithName("Global"), + }) + triggers.RegisterSink(&namespacedTriggerSink{ + reader: mgr.GetClient(), + enqueue: namespacedSource.enqueue, + log: log.WithName("Triggers").WithName("Namespaced"), + }) + if err = (&globalResourceController{ log: log.WithName("Global"), configuration: configuration, metrics: metrics.MustMakeGlobalTenantResourceRecorder(), impersonation: cache, + + triggers: triggers, + triggerSrc: globalSource, }).SetupWithManager(mgr, opts); err != nil { return fmt.Errorf("unable to create global controller: %w", err) } @@ -38,6 +78,9 @@ func Add( metrics: metrics.MustMakeTenantResourceRecorder(), impersonation: cache, + + triggers: triggers, + triggerSrc: namespacedSource, }).SetupWithManager(mgr, opts); err != nil { return fmt.Errorf("unable to create namespaced controller: %w", err) } diff --git a/internal/controllers/resources/namespaced.go b/internal/controllers/resources/namespaced.go index 1b2c01f6e..ef423a37b 100644 --- a/internal/controllers/resources/namespaced.go +++ b/internal/controllers/resources/namespaced.go @@ -27,6 +27,7 @@ import ( ctrllog "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" + "sigs.k8s.io/controller-runtime/pkg/source" capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" "github.com/projectcapsule/capsule/internal/cache" @@ -38,6 +39,7 @@ import ( "github.com/projectcapsule/capsule/pkg/runtime/configuration" "github.com/projectcapsule/capsule/pkg/runtime/predicates" "github.com/projectcapsule/capsule/pkg/runtime/sanitize" + "github.com/projectcapsule/capsule/pkg/runtime/watch" tpl "github.com/projectcapsule/capsule/pkg/template" "github.com/projectcapsule/capsule/pkg/tenant" ) @@ -53,6 +55,9 @@ type namespacedResourceController struct { metrics *metrics.TenantResourceRecorder impersonation *cache.ImpersonationCache + + triggers *watch.Manager + triggerSrc source.Source } func (r *namespacedResourceController) SetupWithManager(mgr ctrl.Manager, ctrlConfig cutils.ControllerOptions) error { @@ -130,6 +135,7 @@ func (r *namespacedResourceController) SetupWithManager(mgr ctrl.Manager, ctrlCo }, ), ). + WatchesRawSource(r.triggerSrc). WithOptions(ctrlConfig.Runtime.ToControllerOptions()). Complete(r) } @@ -146,6 +152,11 @@ func (r *namespacedResourceController) Reconcile(ctx context.Context, request re r.metrics.DeleteMetrics(request.Name, request.Namespace) + // Release any trigger watches this resource was keeping alive. + if serr := r.triggers.Sync(ctx, namespacedTriggerOwnerKey(request.Namespace, request.Name), nil); serr != nil { + log.V(2).Error(serr, "failed to release trigger watches") + } + return reconcile.Result{}, nil } @@ -164,10 +175,14 @@ func (r *namespacedResourceController) Reconcile(ctx context.Context, request re var statusErr error + var triggerCond meta.Condition + //nolint:dupl defer func() { meta.RemoveReconcileTriggerAnnotation(tntResource) + tntResource.Status.Conditions.UpdateConditionByType(triggerCond) + reconcileErr := err if statusErr != nil { reconcileErr = statusErr @@ -204,6 +219,18 @@ func (r *namespacedResourceController) Reconcile(ctx context.Context, request re err = nil }() + // Arm (or, while terminating, release) the dynamic trigger watches for the + // kinds this resource depends on, independent of cordon/dependency state. + // Cluster-scoped kinds are rejected: a namespaced TenantResource can only + // observe objects inside Tenant namespaces. The watches are shared across + // tenants and narrowed server-side to objects carrying the tenant label the + // metadata webhook stamps on every namespaced object in a Tenant namespace; + // the sink routes each event to the right owner via the indexed lookup. + triggerCond = syncTriggers( + ctx, r.triggers, namespacedTriggerOwnerKey(tntResource.Namespace, tntResource.Name), + tntResource, tntResource.Spec.TenantResourceCommonSpec, true, log, + ) + // On Deletion these checks are skipped. //nolint:nestif if tntResource.DeletionTimestamp.IsZero() { diff --git a/internal/controllers/resources/triggers.go b/internal/controllers/resources/triggers.go new file mode 100644 index 000000000..bec22ee65 --- /dev/null +++ b/internal/controllers/resources/triggers.go @@ -0,0 +1,380 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + "errors" + "fmt" + "sync" + + "github.com/go-logr/logr" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/client-go/util/workqueue" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" + capsulemeta "github.com/projectcapsule/capsule/pkg/api/meta" + capruntime "github.com/projectcapsule/capsule/pkg/api/runtime" + "github.com/projectcapsule/capsule/pkg/runtime/indexers/tenantresource" + "github.com/projectcapsule/capsule/pkg/runtime/watch" + "github.com/projectcapsule/capsule/pkg/tenant" + "github.com/projectcapsule/capsule/pkg/utils" +) + +// syncTriggers arms (or, while terminating, releases) the trigger watches for +// a resource and returns the Triggers condition summarizing the outcome. +// namespacedOwner says the owner is a namespaced TenantResource, which implies +// two narrowings at once: cluster-scoped kinds are not watched and are +// surfaced through the condition instead (such a resource can only observe +// objects inside Tenant namespaces), and every watch is narrowed server-side +// to objects carrying Capsule's tenant label, with the sink routing each event +// to the right owner. +func syncTriggers( + ctx context.Context, + m *watch.Manager, + ownerKey string, + obj client.Object, + spec capsulev1beta2.TenantResourceCommonSpec, + namespacedOwner bool, + log logr.Logger, +) capsulemeta.Condition { + if !obj.GetDeletionTimestamp().IsZero() { + if err := m.Sync(ctx, ownerKey, nil); err != nil { + log.V(2).Error(err, "failed to release trigger watches") + } + + return newTriggersCondition(obj, 0, nil) + } + + specs, rejected, specErr := triggerWatchSpecs(m, spec.Triggers, namespacedOwner) + + armErr := errors.Join(specErr, m.Sync(ctx, ownerKey, specs)) + + // A namespaced TenantResource cannot observe cluster-scoped objects. Fold the + // rejected kinds into the condition error regardless of armErr: + // triggerWatchSpecs returns them even alongside an error, so the user sees + // the rejection on the first reconcile instead of only after fixing an + // unrelated armErr. + if len(rejected) > 0 { + armErr = errors.Join(armErr, fmt.Errorf( + "cluster-scoped triggers are not allowed on a namespaced TenantResource: %v", rejected)) + } + + return newTriggersCondition(obj, len(specs), armErr) +} + +// versionKindResolver resolves kind selectors to concrete GroupVersionKinds, +// split by scope. Satisfied by *watch.Manager. +type versionKindResolver interface { + ResolveVersionKinds([]capruntime.VersionKind) (namespaced, clusterScoped []schema.GroupVersionKind, err error) +} + +// triggerWatchSpecs resolves the triggers' kinds into one watch spec per +// distinct kind. Trigger label selectors are deliberately NOT pushed down to +// the apiserver: they are matched in the sinks (TriggerSpec.Matches), so every +// owner of a scope shares one informer per kind instead of one per selector — +// and operation semantics stay honest (a selector-filtered watch would report +// an object relabeled into the selection as CREATE and out of it as DELETE). +// Cluster-scoped kinds are dropped into rejected when namespacedOwner is set. +// +// When namespacedOwner is set, every watch is narrowed server-side to objects +// carrying Capsule's tenant label: one informer per kind across all tenants, +// without per-tenant informers or a growing tenant-value selector. Deliberate +// trade-off: objects missing the label (created while the webhook was down, +// its failurePolicy is Ignore, or predating it and never written since) do not +// reach a tenant-scoped watch and cannot fire triggers until a write stamps +// them. GlobalTenantResource watches stay unfiltered — they may legitimately +// observe unlabeled objects (non-tenant namespaces, cluster-scoped kinds) — so +// a kind watched by both scopes runs two informers. +func triggerWatchSpecs( + r versionKindResolver, + triggers []capsulev1beta2.TriggerSpec, + namespacedOwner bool, +) (specs []watch.Spec, rejected []string, _ error) { + var errs []error + + kinds := sets.New[schema.GroupVersionKind]() + rejectedSeen := sets.New[schema.GroupVersionKind]() + + for _, t := range triggers { + namespaced, clusterScoped, resolveErr := r.ResolveVersionKinds(t.VersionKinds.VersionKinds()) + if resolveErr != nil { + errs = append(errs, resolveErr) + } + + gvks := namespaced + + if namespacedOwner { + for _, g := range clusterScoped { + if rejectedSeen.Has(g) { + continue + } + + rejectedSeen.Insert(g) + + rejected = append(rejected, g.String()) + } + } else { + gvks = append(gvks, clusterScoped...) + } + + kinds.Insert(gvks...) + } + + var sel *metav1.LabelSelector + if namespacedOwner { + sel = tenantScopeSelector() + } + + specs = make([]watch.Spec, 0, kinds.Len()) + + for g := range kinds { + specs = append(specs, watch.Spec{GVK: g, Selector: sel}) + } + + return specs, rejected, errors.Join(errs...) +} + +// tenantScopeSelector requires Capsule's tenant label to exist on the watched +// object; it is the only selector ever pushed down for namespaced +// TenantResource watches. +func tenantScopeSelector() *metav1.LabelSelector { + return &metav1.LabelSelector{ + MatchExpressions: []metav1.LabelSelectorRequirement{{ + Key: capsulemeta.NewTenantLabel, + Operator: metav1.LabelSelectorOpExists, + }}, + } +} + +// globalTriggerSink enqueues GlobalTenantResources whose triggers match a +// change. Capsule-replicated objects are deliberately NOT filtered out, so +// triggers can react to deletion of or tampering with rendered objects; the +// resulting self-trigger echo converges once the rendered output stops changing +// (a no-op SSA apply produces no watch event) and the workqueue dedups bursts by +// owner key. +type globalTriggerSink struct { + reader client.Reader + enqueue func(types.NamespacedName) + log logr.Logger +} + +func (s *globalTriggerSink) Notify(ctx context.Context, gvk schema.GroupVersionKind, op watch.Operation, obj metav1.Object) { + list := &capsulev1beta2.GlobalTenantResourceList{} + if err := s.reader.List(ctx, list, client.MatchingFields{ + tenantresource.TriggersIndexerFieldName: tenantresource.TriggerGKKey(gvk.GroupKind()), + }); err != nil { + s.log.V(2).Error(err, "failed to list globaltenantresources for trigger", "gvk", gvk.String()) + + return + } + + // watch.Operation values match TriggerOperation by construction. + op2 := capsulev1beta2.TriggerOperation(op) + + for i := range list.Items { + gtr := &list.Items[i] + if !s.matches(ctx, gtr.Spec.Triggers, gvk, op2, obj) { + continue + } + + s.enqueue(types.NamespacedName{Name: gtr.Name}) + } +} + +func (s *globalTriggerSink) matches( + ctx context.Context, + triggers []capsulev1beta2.TriggerSpec, + gvk schema.GroupVersionKind, + op capsulev1beta2.TriggerOperation, + obj metav1.Object, +) bool { + for _, t := range triggers { + if !t.Matches(gvk, op, obj.GetLabels()) { + continue + } + + if !s.namespaceSelectorMatches(ctx, t.NamespaceSelector, obj.GetNamespace()) { + continue + } + + return true + } + + return false +} + +func (s *globalTriggerSink) namespaceSelectorMatches(ctx context.Context, sel *metav1.LabelSelector, namespace string) bool { + // A namespace selector is meaningless for cluster-scoped objects, and a nil + // selector matches everything without fetching the Namespace. + if sel == nil || namespace == "" { + return true + } + + ns := &corev1.Namespace{} + if err := s.reader.Get(ctx, types.NamespacedName{Name: namespace}, ns); err != nil { + return false + } + + ok, err := utils.IsNamespaceSelectedBySelector(ns, sel) + + return err == nil && ok +} + +// namespacedTriggerSink enqueues TenantResources whose triggers match a change, +// scoped to objects living in the namespaces of the resource's own Tenant. +// Like the global sink, capsule-replicated objects are not filtered out; the +// self-trigger echo converges once the rendered output stops changing and the +// workqueue dedups bursts by owner key. +type namespacedTriggerSink struct { + reader client.Reader + enqueue func(types.NamespacedName) + log logr.Logger +} + +func (s *namespacedTriggerSink) Notify(ctx context.Context, gvk schema.GroupVersionKind, op watch.Operation, obj metav1.Object) { + // Namespaced TenantResources only react to namespaced objects. + if obj.GetNamespace() == "" { + return + } + + objTenant := s.tenantOfNamespace(ctx, obj.GetNamespace()) + if objTenant == "" { + return + } + + list := &capsulev1beta2.TenantResourceList{} + if err := s.reader.List(ctx, list, client.MatchingFields{ + tenantresource.TriggersIndexerFieldName: tenantresource.TriggerGKKey(gvk.GroupKind()), + }); err != nil { + s.log.V(2).Error(err, "failed to list tenantresources for trigger", "gvk", gvk.String()) + + return + } + + // watch.Operation values match TriggerOperation by construction. + op2 := capsulev1beta2.TriggerOperation(op) + + // Many TenantResources share namespaces; resolve each namespace's tenant at + // most once per event, and only for resources whose triggers match at all. + tenants := map[string]string{obj.GetNamespace(): objTenant} + + for i := range list.Items { + tr := &list.Items[i] + + if !triggerMatches(tr.Spec.Triggers, gvk, op2, obj.GetLabels()) { + continue + } + + // Only fire for objects in a namespace of the resource's own Tenant. + trTenant, ok := tenants[tr.GetNamespace()] + if !ok { + trTenant = s.tenantOfNamespace(ctx, tr.GetNamespace()) + tenants[tr.GetNamespace()] = trTenant + } + + if trTenant != objTenant { + continue + } + + s.enqueue(types.NamespacedName{Name: tr.Name, Namespace: tr.Namespace}) + } +} + +func (s *namespacedTriggerSink) tenantOfNamespace(ctx context.Context, name string) string { + ns := &corev1.Namespace{} + if err := s.reader.Get(ctx, types.NamespacedName{Name: name}, ns); err != nil { + return "" + } + + return tenant.TenanLabelValue(ns) +} + +// triggerSource is a controller-runtime source that lets trigger sinks enqueue +// reconcile requests straight into the controller's workqueue. It replaces +// source.Channel: instead of wrapping each owner key in a fake typed object, +// pushing it through a buffered channel, and unwrapping it back into a request, +// the sink calls enqueue with the key. The workqueue itself provides the dedup +// by owner key and the backpressure the channel buffer used to approximate. +type triggerSource struct { + mu sync.Mutex + queue workqueue.TypedRateLimitingInterface[reconcile.Request] +} + +// Start records the controller's workqueue; it is called once when the manager +// starts. +func (s *triggerSource) Start(_ context.Context, q workqueue.TypedRateLimitingInterface[reconcile.Request]) error { + s.mu.Lock() + s.queue = q + s.mu.Unlock() + + return nil +} + +// enqueue schedules a reconcile of the owner identified by key. Calls before the +// controller has started are dropped and recovered by the resyncPeriod. +func (s *triggerSource) enqueue(key types.NamespacedName) { + s.mu.Lock() + q := s.queue + s.mu.Unlock() + + if q == nil { + return + } + + q.Add(reconcile.Request{NamespacedName: key}) +} + +// triggerMatches reports whether any trigger reacts to a change of the given kind +// and operation whose object carries the given labels. NamespaceSelector is not +// considered here; namespace scoping for namespaced TenantResources is enforced by +// the caller via the owning Tenant. +func triggerMatches( + triggers []capsulev1beta2.TriggerSpec, + gvk schema.GroupVersionKind, + op capsulev1beta2.TriggerOperation, + lbls map[string]string, +) bool { + for _, t := range triggers { + if t.Matches(gvk, op, lbls) { + return true + } + } + + return false +} + +// Owner keys are namespaced by consumer ("triggers/...") so future watch +// consumers (e.g. health checks) can share the manager without collisions. +func globalTriggerOwnerKey(name string) string { + return "triggers/global/" + name +} + +func namespacedTriggerOwnerKey(namespace, name string) string { + return "triggers/namespaced/" + namespace + "/" + name +} + +// newTriggersCondition builds the status condition summarizing the armed watches. +func newTriggersCondition(obj client.Object, count int, err error) capsulemeta.Condition { + cond := capsulemeta.NewTriggersCondition(obj) + + switch { + case err != nil: + cond.Status = metav1.ConditionFalse + cond.Reason = capsulemeta.FailedReason + cond.Message = err.Error() + case count == 0: + cond.Message = "no triggers configured" + default: + cond.Message = fmt.Sprintf("watching %d kind(s)", count) + } + + return cond +} diff --git a/internal/controllers/resources/triggers_bench_test.go b/internal/controllers/resources/triggers_bench_test.go new file mode 100644 index 000000000..c0fff9472 --- /dev/null +++ b/internal/controllers/resources/triggers_bench_test.go @@ -0,0 +1,78 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + "fmt" + "testing" + + "github.com/go-logr/logr" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" + capsulemeta "github.com/projectcapsule/capsule/pkg/api/meta" + capruntime "github.com/projectcapsule/capsule/pkg/api/runtime" + "github.com/projectcapsule/capsule/pkg/runtime/indexers/tenantresource" + "github.com/projectcapsule/capsule/pkg/runtime/watch" +) + +// BenchmarkNamespacedTriggerSink_Notify measures the per-event cost of the +// hottest sink path: one watched-object change fanned against N indexed +// TenantResources across 10 tenants. The fake client is slower than the +// production informer-backed reader (it deep-copies from its tracker), so +// this is an upper bound on real dispatch cost. +func BenchmarkNamespacedTriggerSink_Notify(b *testing.B) { + for _, n := range []int{100, 1000} { + b.Run(fmt.Sprintf("tenantresources=%d", n), func(b *testing.B) { + scheme := testScheme(b) + + objs := make([]client.Object, 0, n+10) + + for i := 0; i < 10; i++ { + objs = append(objs, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("tenant-%d", i), + Labels: map[string]string{capsulemeta.TenantLabel: fmt.Sprintf("tenant-%d", i)}, + }}) + } + + for i := 0; i < n; i++ { + tr := &capsulev1beta2.TenantResource{ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("tr-%d", i), + Namespace: fmt.Sprintf("tenant-%d", i%10), + }} + tr.Spec.Triggers = []capsulev1beta2.TriggerSpec{{ + VersionKinds: capruntime.VersionKinds{Kinds: []string{"ConfigMap"}}, + }} + objs = append(objs, tr) + } + + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objs...). + WithIndex(&capsulev1beta2.TenantResource{}, tenantresource.TriggersIndexerFieldName, tenantresource.NamespacedTriggers{}.Func()). + Build() + + sink := &namespacedTriggerSink{ + reader: cl, + enqueue: func(types.NamespacedName) {}, + log: logr.Discard(), + } + + obj := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "cm", Namespace: "tenant-3"}} + ctx := context.Background() + + b.ReportAllocs() + b.ResetTimer() + + for i := 0; i < b.N; i++ { + sink.Notify(ctx, configMapGVK, watch.OperationUpdate, obj) + } + }) + } +} diff --git a/internal/controllers/resources/triggers_test.go b/internal/controllers/resources/triggers_test.go new file mode 100644 index 000000000..afbdf6ea0 --- /dev/null +++ b/internal/controllers/resources/triggers_test.go @@ -0,0 +1,349 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package resources + +import ( + "context" + "testing" + + "github.com/go-logr/logr" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" + capsulemeta "github.com/projectcapsule/capsule/pkg/api/meta" + capruntime "github.com/projectcapsule/capsule/pkg/api/runtime" + "github.com/projectcapsule/capsule/pkg/runtime/indexers/tenantresource" + "github.com/projectcapsule/capsule/pkg/runtime/watch" +) + +var ( + secretGVK = schema.GroupVersionKind{Version: "v1", Kind: "Secret"} + configMapGVK = schema.GroupVersionKind{Version: "v1", Kind: "ConfigMap"} +) + +func testScheme(t testing.TB) *runtime.Scheme { + t.Helper() + + scheme := runtime.NewScheme() + if err := capsulev1beta2.AddToScheme(scheme); err != nil { + t.Fatalf("add capsule scheme: %v", err) + } + + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("add corev1 scheme: %v", err) + } + + return scheme +} + +func secretTrigger(ops ...capsulev1beta2.TriggerOperation) capsulev1beta2.TriggerSpec { + return capsulev1beta2.TriggerSpec{ + VersionKinds: capruntime.VersionKinds{Kinds: []string{"Secret"}}, + Operations: ops, + } +} + +// collectEnqueued returns an enqueue func that records keys so each Notify's +// output is observable synchronously. +func collectEnqueued() (func(types.NamespacedName), *[]types.NamespacedName) { + var got []types.NamespacedName + + return func(key types.NamespacedName) { got = append(got, key) }, &got +} + +// fakeResolver resolves Secret and ConfigMap as namespaced and Namespace as +// cluster-scoped, mirroring the real REST mapper for the test kinds. +type fakeResolver struct{} + +func (fakeResolver) ResolveVersionKinds(vks []capruntime.VersionKind) (namespaced, clusterScoped []schema.GroupVersionKind, _ error) { + for _, vk := range vks { + switch vk.Kind { + case "Namespace": + clusterScoped = append(clusterScoped, schema.GroupVersionKind{Version: "v1", Kind: "Namespace"}) + default: + namespaced = append(namespaced, schema.GroupVersionKind{Version: "v1", Kind: vk.Kind}) + } + } + + return namespaced, clusterScoped, nil +} + +func TestTriggerWatchSpecs(t *testing.T) { + namespaceGVK := schema.GroupVersionKind{Version: "v1", Kind: "Namespace"} + selA := &metav1.LabelSelector{MatchLabels: map[string]string{"app": "a"}} + selB := &metav1.LabelSelector{MatchLabels: map[string]string{"app": "b"}} + + triggers := []capsulev1beta2.TriggerSpec{ + // Trigger selectors are matched in the sinks, never pushed down: all + // triggers of a kind share one watch. + {VersionKinds: capruntime.VersionKinds{Kinds: []string{"Secret"}}, Selector: selA}, + {VersionKinds: capruntime.VersionKinds{Kinds: []string{"Secret"}}, Selector: selB}, + {VersionKinds: capruntime.VersionKinds{Kinds: []string{"ConfigMap"}}, Selector: selA}, + {VersionKinds: capruntime.VersionKinds{Kinds: []string{"ConfigMap"}}}, + // Cluster-scoped kind: watched for a cluster-scoped owner. + {VersionKinds: capruntime.VersionKinds{Kinds: []string{"Namespace"}}}, + } + + specs, rejected, err := triggerWatchSpecs(fakeResolver{}, triggers, false) + if err != nil || len(rejected) != 0 { + t.Fatalf("unexpected error/rejection: %v %v", err, rejected) + } + + if len(specs) != 3 { + t.Fatalf("expected one spec per distinct kind (3), got %d", len(specs)) + } + + seen := map[schema.GroupVersionKind]bool{} + + for _, s := range specs { + if seen[s.GVK] { + t.Fatalf("kind %s must be watched exactly once", s.GVK) + } + + seen[s.GVK] = true + + if s.Selector != nil { + t.Fatalf("global watches must be unfiltered, got %v for %v", s.Selector, s.GVK) + } + } + + for _, g := range []schema.GroupVersionKind{secretGVK, configMapGVK, namespaceGVK} { + if !seen[g] { + t.Fatalf("missing watch for %v, got %v", g, seen) + } + } +} + +func TestTriggerWatchSpecs_NamespacedOwner(t *testing.T) { + userSel := &metav1.LabelSelector{MatchLabels: map[string]string{"app": "a"}} + + triggers := []capsulev1beta2.TriggerSpec{ + {VersionKinds: capruntime.VersionKinds{Kinds: []string{"Secret"}}, Selector: userSel}, + {VersionKinds: capruntime.VersionKinds{Kinds: []string{"ConfigMap"}}}, + // Cluster-scoped kind, referenced twice: rejected once. + {VersionKinds: capruntime.VersionKinds{Kinds: []string{"Namespace"}}}, + {VersionKinds: capruntime.VersionKinds{Kinds: []string{"Namespace"}}, Selector: userSel}, + } + + specs, rejected, err := triggerWatchSpecs(fakeResolver{}, triggers, true) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(rejected) != 1 || rejected[0] != "/v1, Kind=Namespace" { + t.Fatalf("expected namespace rejected once, got %v", rejected) + } + + if len(specs) != 2 { + t.Fatalf("expected 2 specs, got %d", len(specs)) + } + + // Every tenant-scoped watch carries exactly the tenant-label-exists + // selector: the user selector stays sink-side, so all TenantResources + // share one informer per kind across all tenants. + for _, s := range specs { + if s.Selector == nil { + t.Fatalf("every tenant-scoped watch must carry a selector, got nil for %v", s.GVK) + } + + if len(s.Selector.MatchLabels) != 0 || len(s.Selector.MatchExpressions) != 1 { + t.Fatalf("watch must carry only the tenant-label requirement, got %v for %v", s.Selector, s.GVK) + } + + req := s.Selector.MatchExpressions[0] + if req.Key != capsulemeta.NewTenantLabel || req.Operator != metav1.LabelSelectorOpExists || len(req.Values) != 0 { + t.Fatalf("watch must require the tenant label to exist, got %v for %v", req, s.GVK) + } + } +} + +func TestGlobalTriggerSink_Matching(t *testing.T) { + scheme := testScheme(t) + + gtr := &capsulev1beta2.GlobalTenantResource{ + ObjectMeta: metav1.ObjectMeta{Name: "g1"}, + } + gtr.Spec.Triggers = []capsulev1beta2.TriggerSpec{ + func() capsulev1beta2.TriggerSpec { + t := secretTrigger(capsulev1beta2.TriggerOperationCreate, capsulev1beta2.TriggerOperationUpdate) + t.Selector = &metav1.LabelSelector{MatchLabels: map[string]string{"app": "x"}} + + return t + }(), + } + + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(gtr). + WithIndex(&capsulev1beta2.GlobalTenantResource{}, tenantresource.TriggersIndexerFieldName, tenantresource.GlobalTriggers{}.Func()). + Build() + + enqueue, got := collectEnqueued() + sink := &globalTriggerSink{reader: cl, enqueue: enqueue, log: logr.Discard()} + + matching := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{ + Name: "s1", Namespace: "ns", Labels: map[string]string{"app": "x"}, + }} + + // Matching label + allowed op enqueues. + sink.Notify(context.Background(), secretGVK, watch.OperationCreate, matching) + if len(*got) != 1 || (*got)[0].Name != "g1" { + t.Fatalf("expected g1 enqueued, got %v", *got) + } + + *got = nil + + // Disallowed operation is ignored. + sink.Notify(context.Background(), secretGVK, watch.OperationDelete, matching) + if len(*got) != 0 { + t.Fatalf("DELETE must not enqueue, got %v", *got) + } + + // Non-matching label is ignored. + nonMatching := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "s2", Namespace: "ns"}} + sink.Notify(context.Background(), secretGVK, watch.OperationCreate, nonMatching) + + if len(*got) != 0 { + t.Fatalf("non-matching label must not enqueue, got %v", *got) + } + + // A capsule-replicated object enqueues too: triggers may watch the rendered + // objects themselves (restore on delete, revert on tampering). The + // self-trigger echo is not filtered here; it converges once the rendered + // output stops changing. + owned := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{ + Name: "owned", + Namespace: "ns", + Labels: map[string]string{ + "app": "x", + capsulemeta.CreatedByCapsuleLabel: capsulemeta.ValueControllerReplications, + }, + }} + sink.Notify(context.Background(), secretGVK, watch.OperationCreate, owned) + + if len(*got) != 1 { + t.Fatalf("capsule-replicated object must enqueue, got %v", *got) + } +} + +func TestGlobalTriggerSink_NamespaceSelector(t *testing.T) { + scheme := testScheme(t) + + trig := secretTrigger() + trig.NamespaceSelector = &metav1.LabelSelector{MatchLabels: map[string]string{"env": "prod"}} + + gtr := &capsulev1beta2.GlobalTenantResource{ObjectMeta: metav1.ObjectMeta{Name: "g1"}} + gtr.Spec.Triggers = []capsulev1beta2.TriggerSpec{trig} + + prod := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ns-prod", Labels: map[string]string{"env": "prod"}}} + dev := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ns-dev", Labels: map[string]string{"env": "dev"}}} + + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(gtr, prod, dev). + WithIndex(&capsulev1beta2.GlobalTenantResource{}, tenantresource.TriggersIndexerFieldName, tenantresource.GlobalTriggers{}.Func()). + Build() + + enqueue, got := collectEnqueued() + sink := &globalTriggerSink{reader: cl, enqueue: enqueue, log: logr.Discard()} + + inProd := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "s", Namespace: "ns-prod"}} + sink.Notify(context.Background(), secretGVK, watch.OperationUpdate, inProd) + + if len(*got) != 1 { + t.Fatalf("object in prod namespace must enqueue, got %v", *got) + } + + *got = nil + + inDev := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "s", Namespace: "ns-dev"}} + sink.Notify(context.Background(), secretGVK, watch.OperationUpdate, inDev) + + if len(*got) != 0 { + t.Fatalf("object in dev namespace must not enqueue, got %v", *got) + } +} + +func TestNamespacedTriggerSink_TenantScoping(t *testing.T) { + scheme := testScheme(t) + + tr := &capsulev1beta2.TenantResource{ObjectMeta: metav1.ObjectMeta{Name: "t1", Namespace: "tenant-a-ns1"}} + tr.Spec.Triggers = []capsulev1beta2.TriggerSpec{{ + VersionKinds: capruntime.VersionKinds{APIGroups: []string{"v1"}, Kinds: []string{"ConfigMap"}}, + }} + + nsA1 := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "tenant-a-ns1", Labels: map[string]string{capsulemeta.TenantLabel: "a"}}} + nsA2 := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "tenant-a-ns2", Labels: map[string]string{capsulemeta.TenantLabel: "a"}}} + nsB1 := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "tenant-b-ns1", Labels: map[string]string{capsulemeta.TenantLabel: "b"}}} + + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(tr, nsA1, nsA2, nsB1). + WithIndex(&capsulev1beta2.TenantResource{}, tenantresource.TriggersIndexerFieldName, tenantresource.NamespacedTriggers{}.Func()). + Build() + + enqueue, got := collectEnqueued() + sink := &namespacedTriggerSink{reader: cl, enqueue: enqueue, log: logr.Discard()} + + // Object in another namespace of the SAME tenant fires the TenantResource. + sameTenant := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "cm", Namespace: "tenant-a-ns2"}} + sink.Notify(context.Background(), configMapGVK, watch.OperationUpdate, sameTenant) + + if len(*got) != 1 || (*got)[0].Name != "t1" || (*got)[0].Namespace != "tenant-a-ns1" { + t.Fatalf("same-tenant change must enqueue t1, got %v", *got) + } + + *got = nil + + // Object in another tenant's namespace must NOT fire. + otherTenant := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "cm", Namespace: "tenant-b-ns1"}} + sink.Notify(context.Background(), configMapGVK, watch.OperationUpdate, otherTenant) + + if len(*got) != 0 { + t.Fatalf("cross-tenant change must not enqueue, got %v", *got) + } + + // Cluster-scoped object (no namespace) must NOT fire. + clusterScoped := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "cm"}} + sink.Notify(context.Background(), configMapGVK, watch.OperationUpdate, clusterScoped) + + if len(*got) != 0 { + t.Fatalf("cluster-scoped change must not enqueue, got %v", *got) + } + + // A capsule-replicated object enqueues too: triggers may watch the rendered + // objects themselves. The self-trigger echo converges once the rendered + // output stops changing. + owned := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{ + Name: "cm", + Namespace: "tenant-a-ns2", + Labels: map[string]string{capsulemeta.CreatedByCapsuleLabel: capsulemeta.ValueControllerReplications}, + }} + sink.Notify(context.Background(), configMapGVK, watch.OperationUpdate, owned) + + if len(*got) != 1 { + t.Fatalf("capsule-replicated object must enqueue, got %v", *got) + } +} + +func TestNewTriggersCondition(t *testing.T) { + obj := &capsulev1beta2.GlobalTenantResource{} + + if c := newTriggersCondition(obj, 0, nil); c.Status != metav1.ConditionTrue || c.Message != "no triggers configured" { + t.Fatalf("unexpected empty condition: %+v", c) + } + + if c := newTriggersCondition(obj, 3, nil); c.Status != metav1.ConditionTrue || c.Message != "watching 3 kind(s)" { + t.Fatalf("unexpected active condition: %+v", c) + } + + if c := newTriggersCondition(obj, 0, context.DeadlineExceeded); c.Status != metav1.ConditionFalse || c.Reason != capsulemeta.FailedReason { + t.Fatalf("unexpected error condition: %+v", c) + } +} diff --git a/pkg/api/meta/conditions.go b/pkg/api/meta/conditions.go index de15f7a79..7bbcc0c85 100644 --- a/pkg/api/meta/conditions.go +++ b/pkg/api/meta/conditions.go @@ -19,6 +19,10 @@ const ( BoundCondition string = "Bound" ExhaustedCondition string = "Exhausted" + // TriggersCondition reports the state of the dynamic trigger watches armed + // for a TenantResource / GlobalTenantResource. + TriggersCondition string = "Triggers" + // FailedReason indicates a condition or event observed a failure (Claim Rejected). SucceededReason string = "Succeeded" FailedReason string = "Failed" @@ -153,6 +157,16 @@ func NewAssignedCondition(obj client.Object) Condition { } } +func NewTriggersCondition(obj client.Object) Condition { + return Condition{ + Type: TriggersCondition, + Status: metav1.ConditionTrue, + Reason: ActiveReason, + ObservedGeneration: obj.GetGeneration(), + LastTransitionTime: metav1.Now(), + } +} + func NewReadyConditionReconcilingReason(obj client.Object) Condition { return Condition{ Type: ReadyCondition, diff --git a/pkg/api/runtime/version_kind_types.go b/pkg/api/runtime/version_kind_types.go index e96150c71..0510b838d 100644 --- a/pkg/api/runtime/version_kind_types.go +++ b/pkg/api/runtime/version_kind_types.go @@ -117,6 +117,8 @@ type VersionKinds struct { // - ["*"] means all API groups and versions. // // +optional + // +kubebuilder:validation:MaxItems=100 + // +kubebuilder:validation:items:MaxLength=317 APIGroups []string `json:"apiGroups,omitempty"` // Kinds of the referents. @@ -124,7 +126,9 @@ type VersionKinds struct { // Use "*" to match all kinds. // // +kubebuilder:validation:MinItems=1 + // +kubebuilder:validation:MaxItems=100 // +kubebuilder:validation:items:MinLength=1 + // +kubebuilder:validation:items:MaxLength=63 Kinds []string `json:"kinds"` } diff --git a/pkg/runtime/indexers/indexer.go b/pkg/runtime/indexers/indexer.go index 164f15a44..159b36be1 100644 --- a/pkg/runtime/indexers/indexer.go +++ b/pkg/runtime/indexers/indexer.go @@ -42,6 +42,8 @@ func AddToManager(ctx context.Context, log logr.Logger, mgr manager.Manager) err tenantresource.NamespacedProcessedItems{}, tenantresource.NamespacedResourceNamespace{}, tenantresource.NamespacedCreatedItems{}, + tenantresource.GlobalTriggers{}, + tenantresource.NamespacedTriggers{}, customquota.NamespacedTargetReference{}, customquota.NamespacedObjectUIDReference{}, customquota.GlobalTargetReference{}, diff --git a/pkg/runtime/indexers/tenantresource/const.go b/pkg/runtime/indexers/tenantresource/const.go index b8470f19d..69e2aff18 100644 --- a/pkg/runtime/indexers/tenantresource/const.go +++ b/pkg/runtime/indexers/tenantresource/const.go @@ -3,9 +3,50 @@ package tenantresource +import ( + "fmt" + + "k8s.io/apimachinery/pkg/runtime/schema" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" +) + const ( ServiceAccountIndexerFieldName string = "spec.serviceaccount" ProcessedIndexerFieldName string = "status.items" CreatedIndexerFieldName string = "status.items.created" NamespaceIndexerFieldName string = "metadata.namespace" + TriggersIndexerFieldName string = "spec.triggers" ) + +// TriggerGKKey returns the index key used for a trigger kind. Triggers select +// kinds, optionally by bare API group (matching any version), so the key +// intentionally carries only group and kind; the sinks apply the per-trigger +// version match on top of the index lookup. +func TriggerGKKey(gk schema.GroupKind) string { + const sep = "\x1f" + + return fmt.Sprintf("%s%s%s", gk.Group, sep, gk.Kind) +} + +// triggerIndexKeys extracts the per-GroupKind index keys for a resource's +// triggers. Shared by the Global and Namespaced trigger indexers. +func triggerIndexKeys(spec capsulev1beta2.TenantResourceCommonSpec) []string { + vks := spec.TriggerVersionKinds() + + seen := make(map[string]struct{}, len(vks)) + out := make([]string, 0, len(vks)) + + for _, vk := range vks { + key := TriggerGKKey(vk.GroupVersionKind().GroupKind()) + if _, ok := seen[key]; ok { + continue + } + + seen[key] = struct{}{} + + out = append(out, key) + } + + return out +} diff --git a/pkg/runtime/indexers/tenantresource/global_triggers.go b/pkg/runtime/indexers/tenantresource/global_triggers.go new file mode 100644 index 000000000..70b09d2ca --- /dev/null +++ b/pkg/runtime/indexers/tenantresource/global_triggers.go @@ -0,0 +1,31 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package tenantresource + +import ( + "sigs.k8s.io/controller-runtime/pkg/client" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" +) + +// GlobalTriggers indexes GlobalTenantResources by the GroupVersionKind of each +// of their triggers, so the watch manager can resolve "which GlobalTenantResource +// cares about this kind" without listing every object. +type GlobalTriggers struct{} + +func (g GlobalTriggers) Object() client.Object { + return &capsulev1beta2.GlobalTenantResource{} +} + +func (g GlobalTriggers) Field() string { + return TriggersIndexerFieldName +} + +func (g GlobalTriggers) Func() client.IndexerFunc { + return func(object client.Object) []string { + tgr := object.(*capsulev1beta2.GlobalTenantResource) //nolint:forcetypeassert + + return triggerIndexKeys(tgr.Spec.TenantResourceCommonSpec) + } +} diff --git a/pkg/runtime/indexers/tenantresource/namespaced_triggers.go b/pkg/runtime/indexers/tenantresource/namespaced_triggers.go new file mode 100644 index 000000000..0132a0e53 --- /dev/null +++ b/pkg/runtime/indexers/tenantresource/namespaced_triggers.go @@ -0,0 +1,31 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package tenantresource + +import ( + "sigs.k8s.io/controller-runtime/pkg/client" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" +) + +// NamespacedTriggers indexes TenantResources by the GroupVersionKind of each of +// their triggers, so the watch manager can resolve "which TenantResource cares +// about this kind" without listing every object. +type NamespacedTriggers struct{} + +func (g NamespacedTriggers) Object() client.Object { + return &capsulev1beta2.TenantResource{} +} + +func (g NamespacedTriggers) Field() string { + return TriggersIndexerFieldName +} + +func (g NamespacedTriggers) Func() client.IndexerFunc { + return func(object client.Object) []string { + tgr := object.(*capsulev1beta2.TenantResource) //nolint:forcetypeassert + + return triggerIndexKeys(tgr.Spec.TenantResourceCommonSpec) + } +} diff --git a/pkg/runtime/indexers/tenantresource/triggers_test.go b/pkg/runtime/indexers/tenantresource/triggers_test.go new file mode 100644 index 000000000..4e9826325 --- /dev/null +++ b/pkg/runtime/indexers/tenantresource/triggers_test.go @@ -0,0 +1,99 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package tenantresource_test + +import ( + "reflect" + "sort" + "testing" + + "k8s.io/apimachinery/pkg/runtime/schema" + + capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" + capruntime "github.com/projectcapsule/capsule/pkg/api/runtime" + "github.com/projectcapsule/capsule/pkg/runtime/indexers/tenantresource" +) + +func TestTriggerGKKey(t *testing.T) { + a := tenantresource.TriggerGKKey(schema.GroupKind{Kind: "Secret"}) + b := tenantresource.TriggerGKKey(schema.GroupKind{Kind: "Secret"}) + c := tenantresource.TriggerGKKey(schema.GroupKind{Group: "apps", Kind: "Deployment"}) + + if a != b { + t.Fatalf("equal GroupKinds must produce equal keys: %q != %q", a, b) + } + + if a == c { + t.Fatalf("different GroupKinds must produce different keys, both %q", a) + } +} + +func triggerSpec(apiGroups []string, kinds ...string) capsulev1beta2.TriggerSpec { + return capsulev1beta2.TriggerSpec{VersionKinds: capruntime.VersionKinds{APIGroups: apiGroups, Kinds: kinds}} +} + +func wantKeys(gks ...schema.GroupKind) []string { + out := make([]string, 0, len(gks)) + for _, gk := range gks { + out = append(out, tenantresource.TriggerGKKey(gk)) + } + + sort.Strings(out) + + return out +} + +func TestGlobalTriggers_Func(t *testing.T) { + var idx tenantresource.GlobalTriggers + + if _, ok := idx.Object().(*capsulev1beta2.GlobalTenantResource); !ok { + t.Fatalf("unexpected object type %T", idx.Object()) + } + + if idx.Field() != tenantresource.TriggersIndexerFieldName { + t.Fatalf("unexpected field %q", idx.Field()) + } + + gtr := &capsulev1beta2.GlobalTenantResource{} + gtr.Spec.Triggers = []capsulev1beta2.TriggerSpec{ + triggerSpec(nil, "Secret"), + triggerSpec([]string{"v1"}, "Secret"), // duplicate collapses to one key + triggerSpec([]string{"apps"}, "Deployment"), // bare group keys by group+kind + triggerSpec([]string{"apps/v1"}, "Deployment"), + } + + got := idx.Func()(gtr) + sort.Strings(got) + + want := wantKeys( + schema.GroupKind{Kind: "Secret"}, + schema.GroupKind{Group: "apps", Kind: "Deployment"}, + ) + + if !reflect.DeepEqual(got, want) { + t.Fatalf("keys mismatch\n got: %v\nwant: %v", got, want) + } +} + +func TestNamespacedTriggers_Func(t *testing.T) { + var idx tenantresource.NamespacedTriggers + + if _, ok := idx.Object().(*capsulev1beta2.TenantResource); !ok { + t.Fatalf("unexpected object type %T", idx.Object()) + } + + if idx.Field() != tenantresource.TriggersIndexerFieldName { + t.Fatalf("unexpected field %q", idx.Field()) + } + + tr := &capsulev1beta2.TenantResource{} + tr.Spec.Triggers = []capsulev1beta2.TriggerSpec{triggerSpec(nil, "ConfigMap")} + + got := idx.Func()(tr) + want := wantKeys(schema.GroupKind{Kind: "ConfigMap"}) + + if !reflect.DeepEqual(got, want) { + t.Fatalf("keys mismatch\n got: %v\nwant: %v", got, want) + } +} diff --git a/pkg/runtime/watch/cache.go b/pkg/runtime/watch/cache.go new file mode 100644 index 000000000..e268cd970 --- /dev/null +++ b/pkg/runtime/watch/cache.go @@ -0,0 +1,59 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package watch + +import ( + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/labels" + ctrlcache "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/cluster" +) + +// MetadataCacheFactory builds the per-watch metadata-only caches for the +// manager. Trigger informers must not share the controller cache: they strip +// heavy metadata via a cache-wide transform, and consumers of the shared cache +// rely on exactly those fields (managedFields drives SSA ownership reads +// during prune and adoption). The watch's label selector is pushed down to the +// apiserver, so only matching objects are streamed and cached. Each cache is +// started and stopped by the Manager; nothing else needs to run it. +func MetadataCacheFactory(clu cluster.Cluster) CacheFactory { + return func(selector labels.Selector) (ctrlcache.Cache, error) { + return ctrlcache.New(clu.GetConfig(), ctrlcache.Options{ + HTTPClient: clu.GetHTTPClient(), + Scheme: clu.GetScheme(), + Mapper: clu.GetRESTMapper(), + DefaultTransform: TransformStripHeavyMetadata, + DefaultLabelSelector: selector, + }) + } +} + +// TransformStripHeavyMetadata drops the metadata nobody can match a trigger +// on and which dominates a metadata-only object's size: managedFields +// (~25% of cached bytes measured) and the kubectl last-applied annotation +// (a full copy of the object body on kubectl-applied resources). Sinks match +// on labels, namespace and deletion timestamp only; those pass through +// untouched. Tombstones are unwrapped and their inner object stripped; +// non-objects pass through for the informer machinery to handle. +func TransformStripHeavyMetadata(in any) (any, error) { + obj := metaObject(in) + if obj == nil { + return in, nil + } + + obj.SetManagedFields(nil) + + ann := obj.GetAnnotations() + if _, ok := ann[corev1.LastAppliedConfigAnnotation]; ok { + delete(ann, corev1.LastAppliedConfigAnnotation) + + if len(ann) == 0 { + ann = nil + } + + obj.SetAnnotations(ann) + } + + return in, nil +} diff --git a/pkg/runtime/watch/cache_test.go b/pkg/runtime/watch/cache_test.go new file mode 100644 index 000000000..bbb6fe80d --- /dev/null +++ b/pkg/runtime/watch/cache_test.go @@ -0,0 +1,85 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package watch + +import ( + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + toolscache "k8s.io/client-go/tools/cache" +) + +func TestTransformStripHeavyMetadata(t *testing.T) { + obj := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{ + Name: "s1", + Namespace: "ns", + Labels: map[string]string{"app": "x"}, + Annotations: map[string]string{ + corev1.LastAppliedConfigAnnotation: `{"apiVersion":"v1","kind":"Secret"}`, + "keep.me": "yes", + }, + ManagedFields: []metav1.ManagedFieldsEntry{{Manager: "kubectl"}}, + }} + + out, err := TransformStripHeavyMetadata(obj) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + got, ok := out.(*metav1.PartialObjectMetadata) + if !ok { + t.Fatalf("expected the same object back, got %T", out) + } + + if got.ManagedFields != nil { + t.Fatalf("managedFields must be stripped, got %v", got.ManagedFields) + } + + if _, ok := got.Annotations[corev1.LastAppliedConfigAnnotation]; ok { + t.Fatalf("last-applied annotation must be stripped") + } + + // Fields sinks match on survive. + if got.Annotations["keep.me"] != "yes" || got.Labels["app"] != "x" { + t.Fatalf("unrelated annotations and labels must survive, got %v / %v", got.Annotations, got.Labels) + } + + // Annotations reduced to only the stripped key collapse to nil. + onlyStripped := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{ + Name: "s2", + Annotations: map[string]string{corev1.LastAppliedConfigAnnotation: "{}"}, + }} + + out, err = TransformStripHeavyMetadata(onlyStripped) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if ann := out.(*metav1.PartialObjectMetadata).Annotations; ann != nil { + t.Fatalf("annotations should collapse to nil, got %v", ann) + } + + // Tombstones are unwrapped and their inner object stripped. + inner := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{ + Name: "s3", + ManagedFields: []metav1.ManagedFieldsEntry{{Manager: "kubectl"}}, + }} + tombstone := toolscache.DeletedFinalStateUnknown{Key: "ns/s3", Obj: inner} + + out, err = TransformStripHeavyMetadata(tombstone) + if err != nil || out != any(tombstone) { + t.Fatalf("tombstones must pass through as-is, got %v / %v", out, err) + } + + if inner.ManagedFields != nil { + t.Fatalf("tombstone's inner object must be stripped, got %v", inner.ManagedFields) + } + + // Non-objects pass through untouched. + notAnObject := struct{ any }{} + if out, err := TransformStripHeavyMetadata(notAnObject); err != nil || out != notAnObject { + t.Fatalf("non-objects must pass through, got %v / %v", out, err) + } +} diff --git a/pkg/runtime/watch/manager.go b/pkg/runtime/watch/manager.go new file mode 100644 index 000000000..daa408c05 --- /dev/null +++ b/pkg/runtime/watch/manager.go @@ -0,0 +1,692 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +// Package watch manages dynamic, metadata-only informers for object kinds that +// are only known at runtime (e.g. declared in a spec). Watches are keyed by +// kind plus an optional server-side label selector, reference counted across +// owners and shared between consumers; a fully released watch is kept warm for +// a grace period (absorbing delete/recreate cycles without a relist) and then +// torn down, so its cached objects are not held indefinitely. Object change +// events are fanned out to registered sinks. Sinks still apply their own +// matching: the server-side selector is a bandwidth and memory optimization, +// not the source of truth for which events matter. +package watch + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/go-logr/logr" + "golang.org/x/sync/errgroup" + authorizationv1 "k8s.io/api/authorization/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime/schema" + toolscache "k8s.io/client-go/tools/cache" + ctrlcache "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" + + capruntime "github.com/projectcapsule/capsule/pkg/api/runtime" +) + +const ( + // defaultMaxWatches bounds the number of distinct watches (kind + selector + // combinations) the manager will run concurrently, guarding against a + // runaway set of owners exhausting informer memory. + defaultMaxWatches = 50 + + // defaultWarmGracePeriod is how long an ownerless watch is kept warm before + // it is torn down. Long enough to absorb delete/recreate cycles (GitOps + // prune-and-apply, CI) without a relist storm, short enough that a released + // watch cannot hold its cached objects — which, for cluster-wide kinds like + // Secret or ConfigMap, dominate the watch's memory — indefinitely. + defaultWarmGracePeriod = 10 * time.Minute + + // defaultSweepInterval is how often the manager checks for expired warm + // watches. + defaultSweepInterval = time.Minute + + // accessCheckTimeout bounds the arm-time permission check. It runs under + // the manager lock, so it must not hang on a slow apiserver. + accessCheckTimeout = 5 * time.Second +) + +// Operation is the object lifecycle event a watch reports. +type Operation string + +const ( + // OperationCreate reports creations of watched objects. + OperationCreate Operation = "CREATE" + // OperationUpdate reports updates of watched objects. + OperationUpdate Operation = "UPDATE" + // OperationDelete reports deletions of watched objects. + OperationDelete Operation = "DELETE" +) + +// Spec selects one kind to watch. A non-nil Selector is pushed down to the +// apiserver: only matching objects are streamed and cached, so memory and +// traffic scale with the selection instead of the cluster. Two owners asking +// for the same kind with the same (canonicalized) selector share one watch. +type Spec struct { + GVK schema.GroupVersionKind + Selector *metav1.LabelSelector +} + +// Sink receives raw object change events for every watched kind. Sinks decide +// themselves which events are relevant to them. When the same kind is watched +// under overlapping selectors, a sink can receive the same event once per +// watch; consumers must stay idempotent (the trigger sinks are: the workqueue +// dedups by owner key). +type Sink interface { + Notify(ctx context.Context, gvk schema.GroupVersionKind, op Operation, obj metav1.Object) +} + +// AccessReviewer creates access-review objects against the apiserver. Any +// controller-runtime client.Client satisfies it — writes always bypass the +// cache, so the regular manager client is fine here. +type AccessReviewer interface { + Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error +} + +// CacheFactory builds the cache backing a single watch, scoped to the given +// label selector (labels.Everything() when the watch is unfiltered). Use +// MetadataCacheFactory in production; tests inject fakes. +type CacheFactory func(selector labels.Selector) (ctrlcache.Cache, error) + +// watchKey identifies a watch: a kind plus the canonical string form of its +// selector (labels.Selector sorts requirements, so semantically equal +// selectors share a key; "" means unfiltered). +type watchKey struct { + gvk schema.GroupVersionKind + selector string +} + +func (k watchKey) String() string { + s := k.gvk.String() + if k.selector != "" { + s += " [" + k.selector + "]" + } + + return s +} + +type entry struct { + // owners is the set of owner keys keeping this watch alive. An entry whose + // owner set drained is kept warm for the grace period so delete/recreate + // cycles of an owner do not pay a relist, then swept. A warm watch is not + // free: beyond the informer's fixed ~37 KiB it retains every matched + // object's stripped metadata (~0.75 KiB each measured), which for a + // cluster-wide kind is the dominant cost — hence the bounded warm window. + owners map[string]struct{} + // releasedAt is when the last owner released this watch; zero while owned. + // It bounds the warm window (sweepExpired) and orders evictions (oldest + // released goes first) under limit pressure. + releasedAt time.Time + // cache is the dedicated single-kind cache backing this watch; stopping it + // (stop) tears the informer down. It is created unstarted and started by + // Start, or immediately when the manager is already running; stop is nil + // until then. + cache ctrlcache.Cache + stop context.CancelFunc +} + +// Manager installs and tears down metadata-only watches for dynamically +// requested kind+selector combinations and fans object change events out to +// the registered sinks. Each watch runs in its own dedicated cache built by +// the factory, so tearing one down is a context cancellation and can never +// affect informers other consumers use. Watches are reference counted across +// all owners; a fully released watch is kept warm for a grace period so a +// returning owner re-arms without a relist, then swept — a warm watch retains +// every matched object's metadata (~0.75 KiB each), so it is not held longer +// than the grace period, and the watch limit can evict a warm watch sooner to +// free a slot. It is safe for concurrent use. +type Manager struct { + newCache CacheFactory + authz AccessReviewer + mapper apimeta.RESTMapper + log logr.Logger + + maxWatches int + warmGracePeriod time.Duration + sweepInterval time.Duration + now func() time.Time + + mu sync.Mutex + entries map[watchKey]*entry + // runCtx is the manager's Start context: watch caches must live and die + // with the manager runnable, not with the reconcile call that armed them, + // so entries created before Start are started by Start and entries created + // later start immediately from it. + runCtx context.Context //nolint:containedctx + + // sinks has its own lock so that dispatch — which runs for every event of + // every watched kind — never contends with Sync holding mu across informer + // arming (SSAR round trips, informer creation). + sinksMu sync.RWMutex + sinks []Sink +} + +// NewManager builds a manager that creates a dedicated cache per watch via the +// given factory (use MetadataCacheFactory), verifies permissions through the +// access reviewer (the regular manager client), and resolves kinds via the +// REST mapper. Add it to the controller manager (mgr.Add) so the watch caches +// run and the warm-watch sweeper runs. +func NewManager(newCache CacheFactory, authz AccessReviewer, mapper apimeta.RESTMapper, log logr.Logger) *Manager { + return &Manager{ + newCache: newCache, + authz: authz, + mapper: mapper, + log: log, + maxWatches: defaultMaxWatches, + warmGracePeriod: defaultWarmGracePeriod, + sweepInterval: defaultSweepInterval, + now: time.Now, + entries: make(map[watchKey]*entry), + } +} + +// Start implements manager.Runnable: it runs the caches backing the armed +// watches and periodically tears down watches that have been ownerless for +// longer than the warm grace period. It runs on the leader only, alongside the +// reconcilers that arm the watches; cancelling its context stops every watch +// cache. +func (m *Manager) Start(ctx context.Context) error { + m.mu.Lock() + m.runCtx = ctx + + for _, e := range m.entries { + m.startEntryLocked(ctx, e) + } + m.mu.Unlock() + + ticker := time.NewTicker(m.sweepInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + m.sweepExpired() + } + } +} + +// RegisterSink adds a sink that receives every watched object change event. +// Sinks must not call RegisterSink from Notify. +func (m *Manager) RegisterSink(s Sink) { + m.sinksMu.Lock() + defer m.sinksMu.Unlock() + + m.sinks = append(m.sinks, s) +} + +// ResolveVersionKinds resolves kind selectors to concrete, de-duplicated +// GroupVersionKinds via the REST mapper, split by scope. Selectors without a +// concrete version (e.g. a bare API group) resolve to the mapper's preferred +// version for that group. Selectors that cannot be resolved contribute an +// error; the resolvable remainder is returned regardless. +func (m *Manager) ResolveVersionKinds(vks []capruntime.VersionKind) (namespaced, clusterScoped []schema.GroupVersionKind, _ error) { + seen := make(map[schema.GroupVersionKind]struct{}, len(vks)) + + var errs []error + + for _, vk := range vks { + gvk := vk.GroupVersionKind() + + var versions []string + if gvk.Version != "" && gvk.Version != capruntime.WildcardVersionKindMatcher { + versions = append(versions, gvk.Version) + } + + mapping, err := m.mapper.RESTMapping(gvk.GroupKind(), versions...) + if err != nil { + errs = append(errs, fmt.Errorf("cannot resolve kind %s: %w", gvk, err)) + + continue + } + + resolved := mapping.GroupVersionKind + if _, ok := seen[resolved]; ok { + continue + } + + seen[resolved] = struct{}{} + + if mapping.Scope.Name() == apimeta.RESTScopeNameNamespace { + namespaced = append(namespaced, resolved) + } else { + clusterScoped = append(clusterScoped, resolved) + } + } + + return namespaced, clusterScoped, errors.Join(errs...) +} + +// specKey canonicalizes a Spec into its watch key and parsed selector. A nil +// Selector means unfiltered (labels.Everything()), never labels.Nothing(). +func specKey(s Spec) (watchKey, labels.Selector, error) { + sel := labels.Everything() + + if s.Selector != nil { + parsed, err := metav1.LabelSelectorAsSelector(s.Selector) + if err != nil { + return watchKey{}, nil, fmt.Errorf("invalid selector for %s: %w", s.GVK, err) + } + + sel = parsed + } + + return watchKey{gvk: s.GVK, selector: sel.String()}, sel, nil +} + +// Sync reconciles the exact set of watches an owner wants. New kind+selector +// combinations get a dedicated informer + event handler; watches the owner no +// longer references are released and kept warm for the grace period (or until +// the watch limit needs their slot) in case the owner comes back. Passing an +// empty slice releases every watch held by the owner (used on delete). Errors +// from individual specs (e.g. unresolvable GVK, watch limit reached) are joined +// and returned; successfully armed watches are unaffected. +// +// Success means a watch is armed with list/watch permission verified, not that +// the informer synced: arming never blocks on the potentially slow initial +// list. Failures after arming (e.g. permissions revoked later) are retried by +// the informer in the background and surface only in the controller logs. +func (m *Manager) Sync(ctx context.Context, ownerKey string, specs []Spec) error { + m.mu.Lock() + defer m.mu.Unlock() + + var errs []error + + desired := make(map[watchKey]labels.Selector, len(specs)) + + for _, s := range specs { + k, sel, err := specKey(s) + if err != nil { + errs = append(errs, err) + + continue + } + + desired[k] = sel + } + + checks := m.verifyAccessLocked(ctx, desired) + + for k, sel := range desired { + if err := m.ensureWatchLocked(ctx, k, sel, ownerKey, checks); err != nil { + errs = append(errs, err) + } + } + + // entry.owners is the single source of truth for ownership: release every + // watch this owner holds that is no longer desired. + for k, e := range m.entries { + if _, owned := e.owners[ownerKey]; !owned { + continue + } + + if _, ok := desired[k]; ok { + continue + } + + m.releaseLocked(k, ownerKey) + } + + return errors.Join(errs...) +} + +// startEntryLocked runs the entry's cache under the manager's run context so +// the watch lives and dies with the manager runnable, not with the reconcile +// call that armed it. +func (m *Manager) startEntryLocked(ctx context.Context, e *entry) { + if e.stop != nil { + return + } + + cctx, cancel := context.WithCancel(ctx) + e.stop = cancel + + go func() { + if err := e.cache.Start(cctx); err != nil { + m.log.Error(err, "watch cache terminated") + } + }() +} + +// verifyAccessLocked resolves and permission-checks every kind in the desired +// set that is not armed yet, concurrently. The SSAR round trips dominate cold +// arm latency: running them serially made arming N kinds pay 2N sequential +// round trips under the manager lock. Kinds appearing under several selectors +// are checked once. The result maps each checked kind to nil or its +// resolve/permission error; kinds already armed are absent. +func (m *Manager) verifyAccessLocked(ctx context.Context, desired map[watchKey]labels.Selector) map[schema.GroupVersionKind]error { + results := make(map[schema.GroupVersionKind]error) + + var ( + gvks []schema.GroupVersionKind + gvrs []schema.GroupVersionResource + ) + + for k := range desired { + if _, ok := m.entries[k]; ok { + continue + } + + if _, seen := results[k.gvk]; seen { + continue + } + + // Validate the kind is resolvable before creating an informer for it. + mapping, err := m.mapper.RESTMapping(k.gvk.GroupKind(), k.gvk.Version) + if err != nil { + results[k.gvk] = fmt.Errorf("cannot resolve kind %s: %w", k.gvk, err) + + continue + } + + results[k.gvk] = nil + + gvks = append(gvks, k.gvk) + gvrs = append(gvrs, mapping.Resource) + } + + if len(gvks) == 0 { + return results + } + + // Bound the burst against the apiserver; reviews are answered in memory by + // the authorizer, so a small fan-out already collapses the latency. Errors + // land in checkErrs per kind; the group itself never fails. + checkErrs := make([]error, len(gvks)) + + var eg errgroup.Group + + eg.SetLimit(10) + + for i := range gvks { + eg.Go(func() error { + if err := m.assertWatchable(ctx, gvrs[i]); err != nil { + checkErrs[i] = fmt.Errorf("cannot watch %s: %w", gvks[i], err) + } + + return nil + }) + } + + _ = eg.Wait() + + for i, gvk := range gvks { + if checkErrs[i] != nil { + results[gvk] = checkErrs[i] + } + } + + return results +} + +func (m *Manager) ensureWatchLocked(ctx context.Context, k watchKey, sel labels.Selector, ownerKey string, checks map[schema.GroupVersionKind]error) error { + if e, ok := m.entries[k]; ok { + e.owners[ownerKey] = struct{}{} + e.releasedAt = time.Time{} + + return nil + } + + if err := checks[k.gvk]; err != nil { + return err + } + + // A warm (ownerless) watch is torn down early when its slot is needed. + if len(m.entries) >= m.maxWatches && !m.evictWarmLocked() { + return fmt.Errorf("watch limit reached (%d), not watching %s", m.maxWatches, k) + } + + c, err := m.armCache(ctx, k.gvk, sel) + if err != nil { + return fmt.Errorf("failed to arm %s: %w", k, err) + } + + e := &entry{ + owners: map[string]struct{}{ownerKey: {}}, + cache: c, + } + + if m.runCtx != nil { + // The cache's lifetime is deliberately bound to the manager's run + // context, not this reconcile's ctx. + m.startEntryLocked(m.runCtx, e) //nolint:contextcheck + } + + m.entries[k] = e + + m.log.V(3).Info("armed watch", "watch", k.String()) + + return nil +} + +// armCache builds the dedicated cache backing one watch and registers the +// event handler. It never blocks on the initial cache sync: a large list can +// be slow, and holding the manager lock through it would stall every other +// reconcile arming watches. Events flow once the informer syncs. +func (m *Manager) armCache(ctx context.Context, gvk schema.GroupVersionKind, sel labels.Selector) (ctrlcache.Cache, error) { + c, err := m.newCache(sel) + if err != nil { + return nil, fmt.Errorf("failed to build cache: %w", err) + } + + informer, err := c.GetInformer(ctx, partialFor(gvk), ctrlcache.BlockUntilSynced(false)) + if err != nil { + return nil, fmt.Errorf("failed to get informer: %w", err) + } + + // Informer callbacks carry no context; dispatch synthesizes a background one. + if _, err := informer.AddEventHandler(m.handlerFor(gvk)); err != nil { //nolint:contextcheck + return nil, fmt.Errorf("failed to add event handler: %w", err) + } + + return c, nil +} + +// releaseLocked drops an owner from a watch. The watch is kept warm for the +// grace period: an owner that is deleted and quickly recreated (CI, GitOps +// prune-and-apply) re-arms instantly instead of paying a full LIST per cycle, +// while a watch nobody re-claims cannot hold its cached objects (the dominant +// cost for a cluster-wide kind) beyond the grace period. +func (m *Manager) releaseLocked(k watchKey, ownerKey string) { + e, ok := m.entries[k] + if !ok { + return + } + + delete(e.owners, ownerKey) + + if len(e.owners) == 0 { + e.releasedAt = m.now() + + m.log.V(3).Info("watch released, keeping warm", "watch", k.String(), "gracePeriod", m.warmGracePeriod.String()) + } +} + +// sweepExpired tears down watches that have been ownerless for longer than the +// warm grace period, freeing their cached objects. Deleting from m.entries +// while ranging over it is safe in Go. +func (m *Manager) sweepExpired() { + m.mu.Lock() + defer m.mu.Unlock() + + deadline := m.now().Add(-m.warmGracePeriod) + + for k, e := range m.entries { + if len(e.owners) > 0 || e.releasedAt.After(deadline) { + continue + } + + m.removeEntryLocked(k) + + m.log.V(3).Info("warm watch expired", "watch", k.String()) + } +} + +// evictWarmLocked tears down the longest-released ownerless watch to free a +// slot for a new one. Returns false when every watch is still owned. +func (m *Manager) evictWarmLocked() bool { + var ( + victim watchKey + oldest *entry + ) + + for k, e := range m.entries { + if len(e.owners) > 0 { + continue + } + + if oldest == nil || e.releasedAt.Before(oldest.releasedAt) { + victim, oldest = k, e + } + } + + if oldest == nil { + return false + } + + m.removeEntryLocked(victim) + + m.log.V(3).Info("evicted warm watch", "watch", victim.String()) + + return true +} + +// removeEntryLocked stops the watch's dedicated cache and forgets its entry. +// Cancelling the cache context halts its reflector and processor and releases +// the registered handler with it; a cache created before Start (never started) +// has nothing to stop. +func (m *Manager) removeEntryLocked(k watchKey) { + if e, ok := m.entries[k]; ok && e.stop != nil { + e.stop() + } + + delete(m.entries, k) +} + +// assertWatchable verifies via SelfSubjectAccessReview that the controller's +// own identity may list and watch the resource before an informer is created +// for it. An informer that lacks these permissions never errors anywhere a +// caller can see — the reflector retries forever and only logs — so this is +// the one place a permission problem can surface into the owner's status +// (same pattern Kyverno uses for its background controller). The reviews are +// answered by the apiserver's authorizer in memory: no objects are listed, and +// creating SelfSubjectAccessReviews is granted to every authenticated identity +// via the system:basic-user role. +func (m *Manager) assertWatchable(ctx context.Context, gvr schema.GroupVersionResource) error { + ctx, cancel := context.WithTimeout(ctx, accessCheckTimeout) + defer cancel() + + for _, verb := range []string{"list", "watch"} { + review := &authorizationv1.SelfSubjectAccessReview{ + Spec: authorizationv1.SelfSubjectAccessReviewSpec{ + ResourceAttributes: &authorizationv1.ResourceAttributes{ + Verb: verb, + Group: gvr.Group, + Version: gvr.Version, + Resource: gvr.Resource, + }, + }, + } + + if err := m.authz.Create(ctx, review); err != nil { + return fmt.Errorf("failed to verify %q permission: %w", verb, err) + } + + if !review.Status.Allowed { + reason := review.Status.Reason + if reason == "" { + reason = "RBAC denies it" + } + + return fmt.Errorf("controller is not permitted to %s %s: %s", verb, gvr.GroupResource(), reason) + } + } + + return nil +} + +// handlerFor builds the informer event handler for a watched kind. Informer +// callbacks carry no context, so dispatch synthesizes a background one. +func (m *Manager) handlerFor(gvk schema.GroupVersionKind) toolscache.ResourceEventHandlerDetailedFuncs { + return toolscache.ResourceEventHandlerDetailedFuncs{ + AddFunc: func(obj any, isInInitialList bool) { + // Skip the informer's initial LIST replay: those objects predate the + // watch and are not creations. The owner reconciles independently, so + // nothing is missed. + if isInInitialList { + return + } + + m.dispatch(gvk, OperationCreate, obj) + }, + UpdateFunc: func(oldObj, newObj any) { + n := metaObject(newObj) + if n == nil { + return + } + + // Periodic resyncs replay every cached object as an update with an + // unchanged resourceVersion; a real write always bumps it. Dropping + // those here keeps a resync from fanning no-op updates of every + // cached object out to every sink. + if o := metaObject(oldObj); o != nil && o.GetResourceVersion() == n.GetResourceVersion() { + return + } + + m.dispatchObj(gvk, OperationUpdate, n) + }, + DeleteFunc: func(obj any) { + m.dispatch(gvk, OperationDelete, obj) + }, + } +} + +func (m *Manager) dispatch(gvk schema.GroupVersionKind, op Operation, raw any) { + obj := metaObject(raw) + if obj == nil { + return + } + + m.dispatchObj(gvk, op, obj) +} + +func (m *Manager) dispatchObj(gvk schema.GroupVersionKind, op Operation, obj metav1.Object) { + m.sinksMu.RLock() + defer m.sinksMu.RUnlock() + + // Informer callbacks have no inbound context; a fresh one is correct here. + ctx := context.Background() + for _, s := range m.sinks { + s.Notify(ctx, gvk, op, obj) + } +} + +func metaObject(raw any) metav1.Object { + if tombstone, ok := raw.(toolscache.DeletedFinalStateUnknown); ok { + raw = tombstone.Obj + } + + obj, err := apimeta.Accessor(raw) + if err != nil { + return nil + } + + return obj +} + +func partialFor(gvk schema.GroupVersionKind) *metav1.PartialObjectMetadata { + obj := &metav1.PartialObjectMetadata{} + obj.SetGroupVersionKind(gvk) + + return obj +} diff --git a/pkg/runtime/watch/manager_test.go b/pkg/runtime/watch/manager_test.go new file mode 100644 index 000000000..c29c5f47a --- /dev/null +++ b/pkg/runtime/watch/manager_test.go @@ -0,0 +1,757 @@ +// Copyright 2020-2026 Project Capsule Authors +// SPDX-License-Identifier: Apache-2.0 + +package watch + +import ( + "context" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/go-logr/logr" + authorizationv1 "k8s.io/api/authorization/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime/schema" + toolscache "k8s.io/client-go/tools/cache" + ctrlcache "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" + + capruntime "github.com/projectcapsule/capsule/pkg/api/runtime" +) + +var ( + secretGVK = schema.GroupVersionKind{Version: "v1", Kind: "Secret"} + configMapGVK = schema.GroupVersionKind{Version: "v1", Kind: "ConfigMap"} + namespaceGVK = schema.GroupVersionKind{Version: "v1", Kind: "Namespace"} + deploymentGVK = schema.GroupVersionKind{Group: "apps", Version: "v1", Kind: "Deployment"} +) + +// --- fakes for the per-watch caches the factory hands out --- + +type fakeRegistration struct{} + +func (fakeRegistration) HasSynced() bool { return true } + +type fakeInformer struct { + handlers []toolscache.ResourceEventHandler +} + +func (f *fakeInformer) AddEventHandler(h toolscache.ResourceEventHandler) (toolscache.ResourceEventHandlerRegistration, error) { + f.handlers = append(f.handlers, h) + + return fakeRegistration{}, nil +} + +func (f *fakeInformer) AddEventHandlerWithResyncPeriod(h toolscache.ResourceEventHandler, _ time.Duration) (toolscache.ResourceEventHandlerRegistration, error) { + return f.AddEventHandler(h) +} + +func (f *fakeInformer) AddEventHandlerWithOptions(h toolscache.ResourceEventHandler, _ toolscache.HandlerOptions) (toolscache.ResourceEventHandlerRegistration, error) { + return f.AddEventHandler(h) +} + +func (f *fakeInformer) RemoveEventHandler(toolscache.ResourceEventHandlerRegistration) error { + return nil +} + +func (f *fakeInformer) AddIndexers(toolscache.Indexers) error { return nil } +func (f *fakeInformer) HasSynced() bool { return true } +func (f *fakeInformer) IsStopped() bool { return false } + +// fakeCache backs exactly one watch, like the real per-watch caches. +type fakeCache struct { + ctrlcache.Cache + + selector string + gvk schema.GroupVersionKind + informer *fakeInformer + startedCh chan struct{} +} + +func (c *fakeCache) GetInformer(_ context.Context, obj client.Object, _ ...ctrlcache.InformerGetOption) (ctrlcache.Informer, error) { + c.gvk = obj.GetObjectKind().GroupVersionKind() + + if c.informer == nil { + c.informer = &fakeInformer{} + } + + return c.informer, nil +} + +func (c *fakeCache) Start(ctx context.Context) error { + close(c.startedCh) + <-ctx.Done() + + return nil +} + +// fakeFactory records every cache it built, in creation order. +type fakeFactory struct { + caches []*fakeCache +} + +func (f *fakeFactory) new(sel labels.Selector) (ctrlcache.Cache, error) { + c := &fakeCache{ + selector: sel.String(), + startedCh: make(chan struct{}), + } + + f.caches = append(f.caches, c) + + return c, nil +} + +// createdFor counts how many caches (i.e. informers, i.e. LISTs against the +// apiserver) were ever built for a kind. +func (f *fakeFactory) createdFor(gvk schema.GroupVersionKind) int { + n := 0 + + for _, c := range f.caches { + if c.gvk == gvk { + n++ + } + } + + return n +} + +// handlerFor returns the event handler registered for a kind's first cache. +func (f *fakeFactory) handlerFor(t *testing.T, gvk schema.GroupVersionKind) toolscache.ResourceEventHandler { + t.Helper() + + for _, c := range f.caches { + if c.gvk == gvk && c.informer != nil && len(c.informer.handlers) > 0 { + return c.informer.handlers[0] + } + } + + t.Fatalf("no handler registered for %s", gvk) + + return nil +} + +// entryCount counts the manager's live watches for a kind (any selector). +func entryCount(m *Manager, gvk schema.GroupVersionKind) int { + m.mu.Lock() + defer m.mu.Unlock() + + n := 0 + + for k := range m.entries { + if k.gvk == gvk { + n++ + } + } + + return n +} + +// fakeAuthorizer answers SelfSubjectAccessReviews; resources listed in denied +// are refused. Reviews run concurrently, so it locks. +type fakeAuthorizer struct { + mu sync.Mutex + denied map[string]struct{} + calls int +} + +func (f *fakeAuthorizer) Create(_ context.Context, obj client.Object, _ ...client.CreateOption) error { + f.mu.Lock() + defer f.mu.Unlock() + + f.calls++ + + review, ok := obj.(*authorizationv1.SelfSubjectAccessReview) + if !ok { + return fmt.Errorf("unexpected object %T", obj) + } + + if _, deny := f.denied[review.Spec.ResourceAttributes.Resource]; deny { + review.Status.Allowed = false + review.Status.Reason = "RBAC: no rule grants access" + + return nil + } + + review.Status.Allowed = true + + return nil +} + +func allowAll() *fakeAuthorizer { return &fakeAuthorizer{} } + +func testMapper() apimeta.RESTMapper { + m := apimeta.NewDefaultRESTMapper([]schema.GroupVersion{ + {Version: "v1"}, + {Group: "apps", Version: "v1"}, + }) + m.Add(secretGVK, apimeta.RESTScopeNamespace) + m.Add(configMapGVK, apimeta.RESTScopeNamespace) + m.Add(namespaceGVK, apimeta.RESTScopeRoot) + m.Add(deploymentGVK, apimeta.RESTScopeNamespace) + + return m +} + +func newTestManager() (*fakeFactory, *Manager) { + f := &fakeFactory{} + + return f, NewManager(f.new, allowAll(), testMapper(), logr.Discard()) +} + +func unfiltered(gvks ...schema.GroupVersionKind) []Spec { + specs := make([]Spec, 0, len(gvks)) + for _, g := range gvks { + specs = append(specs, Spec{GVK: g}) + } + + return specs +} + +func syncSpecsOrFatal(t *testing.T, m *Manager, owner string, specs ...Spec) { + t.Helper() + + if err := m.Sync(context.Background(), owner, specs); err != nil { + t.Fatalf("Sync(%q) unexpected error: %v", owner, err) + } +} + +func syncOrFatal(t *testing.T, m *Manager, owner string, gvks ...schema.GroupVersionKind) { + t.Helper() + + syncSpecsOrFatal(t, m, owner, unfiltered(gvks...)...) +} + +// fakeClock drives the manager's release timestamps (eviction order) in tests. +type fakeClock struct { + t time.Time +} + +func (c *fakeClock) now() time.Time { return c.t } +func (c *fakeClock) advance(d time.Duration) { c.t = c.t.Add(d) } +func newFakeClock() *fakeClock { return &fakeClock{t: time.Unix(1000, 0)} } +func withFakeClock(m *Manager) *fakeClock { + c := newFakeClock() + m.now = c.now + + return c +} + +func TestManager_RefcountAndWarmRelease(t *testing.T) { + f, m := newTestManager() + + syncOrFatal(t, m, "a", secretGVK) + syncOrFatal(t, m, "b", secretGVK) + + if f.createdFor(secretGVK) != 1 { + t.Fatalf("watch should be created once, got %d", f.createdFor(secretGVK)) + } + + // Releasing one of two owners keeps the shared watch alive. + syncOrFatal(t, m, "a") + + if entryCount(m, secretGVK) != 1 { + t.Fatalf("watch must survive while owner b holds it") + } + + // Releasing the last owner keeps the watch warm instead of tearing it + // down, so delete/recreate cycles do not cause a full LIST per cycle. + syncOrFatal(t, m, "b") + + if entryCount(m, secretGVK) != 1 { + t.Fatalf("watch must stay warm after last release") + } + + // Re-arming reuses the warm watch without a new LIST. + syncOrFatal(t, m, "a", secretGVK) + + if f.createdFor(secretGVK) != 1 { + t.Fatalf("warm watch must be reused, created=%d", f.createdFor(secretGVK)) + } +} + +func TestManager_WarmGracePeriod(t *testing.T) { + f, m := newTestManager() + clock := withFakeClock(m) + + syncOrFatal(t, m, "a", secretGVK) + syncOrFatal(t, m, "a") + + // Within the grace period the warm watch survives sweeps. + clock.advance(m.warmGracePeriod / 2) + m.sweepExpired() + + if entryCount(m, secretGVK) != 1 { + t.Fatalf("watch must survive sweeps within the grace period") + } + + // Re-arming reuses the warm watch without a new LIST and resets the warm + // state: the entry is owned again and the old release timestamp must not + // count against it. + syncOrFatal(t, m, "a", secretGVK) + + if f.createdFor(secretGVK) != 1 { + t.Fatalf("warm watch must be reused, created=%d", f.createdFor(secretGVK)) + } + + clock.advance(m.warmGracePeriod * 2) + m.sweepExpired() + + if entryCount(m, secretGVK) != 1 { + t.Fatalf("owned watch must never be swept") + } + + // Once ownerless past the grace period, the sweep tears it down and frees + // its cached objects. + syncOrFatal(t, m, "a") + clock.advance(m.warmGracePeriod + time.Second) + m.sweepExpired() + + if entryCount(m, secretGVK) != 0 { + t.Fatalf("expired warm watch must be torn down") + } +} + +func TestManager_ReleasedWatchEvictedUnderSlotPressure(t *testing.T) { + f, m := newTestManager() + m.maxWatches = 2 + + // A fully released watch stays warm and a returning owner reuses it without + // a new LIST (the sweeper never runs in this test). + syncOrFatal(t, m, "a", secretGVK) + syncOrFatal(t, m, "a") + syncOrFatal(t, m, "a", secretGVK) + syncOrFatal(t, m, "a") + + if f.createdFor(secretGVK) != 1 { + t.Fatalf("warm watch must be reused, created=%d", f.createdFor(secretGVK)) + } + + // Slot pressure reclaims it before the grace period: filling the limit + // evicts the ownerless watch for a new kind. + syncOrFatal(t, m, "b", configMapGVK) + syncOrFatal(t, m, "b", configMapGVK, deploymentGVK) + + if entryCount(m, secretGVK) != 0 { + t.Fatalf("ownerless watch must be evicted under slot pressure") + } + + if entryCount(m, deploymentGVK) != 1 { + t.Fatalf("new watch must take the reclaimed slot") + } +} + +func TestManager_TenantScopedAndGlobalWatchesOfAKind(t *testing.T) { + f, m := newTestManager() + + // The trigger layer's two selector classes for one kind: the shared + // tenant-label-exists watch (namespaced TenantResources) and the + // unfiltered watch (GlobalTenantResources). + tenantScoped := &metav1.LabelSelector{MatchExpressions: []metav1.LabelSelectorRequirement{{ + Key: "projectcapsule.dev/tenant", + Operator: metav1.LabelSelectorOpExists, + }}} + + // Owners across different tenants share the single tenant-scoped watch. + syncSpecsOrFatal(t, m, "tenant-a/tr", Spec{GVK: secretGVK, Selector: tenantScoped}) + syncSpecsOrFatal(t, m, "tenant-b/tr", Spec{GVK: secretGVK, Selector: tenantScoped}) + + if entryCount(m, secretGVK) != 1 || f.createdFor(secretGVK) != 1 { + t.Fatalf("all tenant-scoped owners must share one watch, entries=%d created=%d", entryCount(m, secretGVK), f.createdFor(secretGVK)) + } + + // A global owner of the same kind gets its own unfiltered watch. + syncOrFatal(t, m, "global/gtr", secretGVK) + + if entryCount(m, secretGVK) != 2 || f.createdFor(secretGVK) != 2 { + t.Fatalf("the global watch must be a second, distinct entry, entries=%d created=%d", entryCount(m, secretGVK), f.createdFor(secretGVK)) + } + + // Releasing every tenant-scoped owner leaves the global watch owned; the + // tenant-scoped one goes warm. + syncOrFatal(t, m, "tenant-a/tr") + syncOrFatal(t, m, "tenant-b/tr") + + m.mu.Lock() + for k, e := range m.entries { + owned := len(e.owners) > 0 + if k.selector == "" && !owned { + t.Fatalf("the global watch must remain owned") + } + + if k.selector != "" && owned { + t.Fatalf("the tenant-scoped watch must be fully released, owners=%v", e.owners) + } + } + m.mu.Unlock() +} + +func TestManager_EvictsOldestWarmFirst(t *testing.T) { + f, m := newTestManager() + clock := withFakeClock(m) + m.maxWatches = 2 + + syncOrFatal(t, m, "a", secretGVK) + syncOrFatal(t, m, "b", configMapGVK) + + // Release secret first, configmap later: secret is the older warm entry. + syncOrFatal(t, m, "a") + clock.advance(time.Minute) + syncOrFatal(t, m, "b") + + syncOrFatal(t, m, "c", deploymentGVK) + + if entryCount(m, secretGVK) != 0 { + t.Fatalf("oldest warm watch must be evicted first") + } + + if entryCount(m, configMapGVK) != 1 { + t.Fatalf("younger warm watch must survive") + } + + if f.createdFor(deploymentGVK) != 1 { + t.Fatalf("deployment watch should have been created, created=%d", f.createdFor(deploymentGVK)) + } +} + +func TestManager_SyncSwapsGVKs(t *testing.T) { + f, m := newTestManager() + + syncOrFatal(t, m, "a", secretGVK) + // Same owner now wants a different kind: old goes warm, new is armed. + syncOrFatal(t, m, "a", configMapGVK) + + if entryCount(m, secretGVK) != 1 { + t.Fatalf("secret watch should stay warm") + } + + if f.createdFor(configMapGVK) != 1 { + t.Fatalf("configmap watch should have been created, created=%d", f.createdFor(configMapGVK)) + } +} + +func TestManager_MaxWatchesEvictsWarmOnly(t *testing.T) { + f, m := newTestManager() + m.maxWatches = 1 + + syncOrFatal(t, m, "a", secretGVK) + + // The slot is owned: arming another kind must fail. + err := m.Sync(context.Background(), "a", unfiltered(secretGVK, configMapGVK)) + if err == nil { + t.Fatalf("expected error when exceeding the watch limit") + } + + if f.createdFor(configMapGVK) != 0 { + t.Fatalf("configmap watch must not be created past the limit") + } + + // The already-armed secret watch must be unaffected. + if entryCount(m, secretGVK) != 1 { + t.Fatalf("secret watch must remain armed") + } + + // Once the secret watch is merely warm, its slot is reclaimed for new kinds. + syncOrFatal(t, m, "a") + syncOrFatal(t, m, "a", configMapGVK) + + if entryCount(m, secretGVK) != 0 { + t.Fatalf("warm secret watch must be evicted under limit pressure") + } + + if f.createdFor(configMapGVK) != 1 { + t.Fatalf("configmap watch should have been created after eviction, created=%d", f.createdFor(configMapGVK)) + } +} + +func TestManager_SelectorWatches(t *testing.T) { + f, m := newTestManager() + + selA := &metav1.LabelSelector{MatchLabels: map[string]string{"app": "a"}} + selAEquivalent := &metav1.LabelSelector{MatchLabels: map[string]string{"app": "a"}} + selB := &metav1.LabelSelector{MatchLabels: map[string]string{"app": "b"}} + + // Same kind under two different selectors: two independent watches. + syncSpecsOrFatal(t, m, "a", Spec{GVK: secretGVK, Selector: selA}, Spec{GVK: secretGVK, Selector: selB}) + + if entryCount(m, secretGVK) != 2 || len(f.caches) != 2 { + t.Fatalf("expected two watches for two selectors, entries=%d caches=%d", entryCount(m, secretGVK), len(f.caches)) + } + + // The selector must reach the cache factory (i.e. the apiserver). + got := map[string]bool{} + for _, c := range f.caches { + got[c.selector] = true + } + + if !got["app=a"] || !got["app=b"] { + t.Fatalf("expected caches for app=a and app=b, got %v", got) + } + + // An equal selector from another owner shares the existing watch. + syncSpecsOrFatal(t, m, "b", Spec{GVK: secretGVK, Selector: selAEquivalent}) + + if len(f.caches) != 2 { + t.Fatalf("equal selectors must share a watch, caches=%d", len(f.caches)) + } + + // Filtered and unfiltered watches of the same kind are distinct entries. + syncOrFatal(t, m, "c", secretGVK) + + if entryCount(m, secretGVK) != 3 { + t.Fatalf("unfiltered watch must be its own entry, entries=%d", entryCount(m, secretGVK)) + } + + // Invalid selectors surface as errors and arm nothing. + bad := &metav1.LabelSelector{MatchExpressions: []metav1.LabelSelectorRequirement{{Key: "k", Operator: "Bogus"}}} + if err := m.Sync(context.Background(), "d", []Spec{{GVK: configMapGVK, Selector: bad}}); err == nil { + t.Fatalf("expected error for invalid selector") + } + + if f.createdFor(configMapGVK) != 0 { + t.Fatalf("no watch must be created for an invalid selector") + } +} + +func TestManager_StartRunsCaches(t *testing.T) { + f, m := newTestManager() + + // Armed before Start: the cache exists but is not running. + syncOrFatal(t, m, "a", secretGVK) + + select { + case <-f.caches[0].startedCh: + t.Fatalf("cache must not start before the manager starts") + default: + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan error, 1) + + go func() { done <- m.Start(ctx) }() + + // Start runs pre-existing caches... + select { + case <-f.caches[0].startedCh: + case <-time.After(5 * time.Second): + t.Fatalf("pre-armed cache was not started") + } + + // ...and caches armed while running start immediately. + syncOrFatal(t, m, "a", secretGVK, configMapGVK) + + select { + case <-f.caches[1].startedCh: + case <-time.After(5 * time.Second): + t.Fatalf("cache armed after start was not started") + } + + cancel() + + select { + case err := <-done: + if err != nil { + t.Fatalf("Start returned error: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatalf("Start did not return on context cancellation") + } +} + +func TestManager_UnresolvableGVK(t *testing.T) { + f, m := newTestManager() + + bad := schema.GroupVersionKind{Group: "x", Version: "v1", Kind: "Nope"} + if err := m.Sync(context.Background(), "a", unfiltered(bad)); err == nil { + t.Fatalf("expected error for unresolvable GVK") + } + + if len(f.caches) != 0 { + t.Fatalf("no watch should be created for an unresolvable GVK, got %d", len(f.caches)) + } +} + +func TestManager_ResolveVersionKinds(t *testing.T) { + _, m := newTestManager() + + namespaced, clusterScoped, err := m.ResolveVersionKinds([]capruntime.VersionKind{ + {Kind: "Secret"}, // empty apiVersion means core v1 + {APIVersion: "v1", Kind: "Secret"}, // duplicate after resolution + {APIVersion: "apps/*", Kind: "Deployment"}, // bare group resolves to the preferred version + {APIVersion: "v1", Kind: "Namespace"}, // cluster-scoped + }) + if err != nil { + t.Fatalf("unexpected resolve error: %v", err) + } + + wantNamespaced := []schema.GroupVersionKind{secretGVK, deploymentGVK} + if len(namespaced) != len(wantNamespaced) { + t.Fatalf("expected namespaced %v, got %v", wantNamespaced, namespaced) + } + + for i, g := range wantNamespaced { + if namespaced[i] != g { + t.Fatalf("namespaced[%d] = %v, want %v", i, namespaced[i], g) + } + } + + if len(clusterScoped) != 1 || clusterScoped[0] != namespaceGVK { + t.Fatalf("expected cluster-scoped [%v], got %v", namespaceGVK, clusterScoped) + } + + // An unresolvable selector yields an error but keeps the resolvable remainder. + namespaced, _, err = m.ResolveVersionKinds([]capruntime.VersionKind{ + {Kind: "Nope"}, + {APIVersion: "v1", Kind: "Secret"}, + }) + if err == nil { + t.Fatalf("expected error for unresolvable selector") + } + + if len(namespaced) != 1 || namespaced[0] != secretGVK { + t.Fatalf("resolvable selectors must survive an error, got %v", namespaced) + } +} + +// recSink records what the manager dispatches. +type recSink struct { + ops []Operation + objs []string +} + +func (s *recSink) Notify(_ context.Context, _ schema.GroupVersionKind, op Operation, obj metav1.Object) { + s.ops = append(s.ops, op) + s.objs = append(s.objs, obj.GetName()) +} + +func TestManager_DispatchOperations(t *testing.T) { + f, m := newTestManager() + + sink := &recSink{} + m.RegisterSink(sink) + + syncOrFatal(t, m, "a", secretGVK) + + handler := f.handlerFor(t, secretGVK) + + obj := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "s1", Namespace: "ns", ResourceVersion: "1"}} + updated := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "s1", Namespace: "ns", ResourceVersion: "2"}} + handler.OnAdd(obj, false) + handler.OnUpdate(obj, updated) + handler.OnDelete(updated) + + wantOps := []Operation{OperationCreate, OperationUpdate, OperationDelete} + + if len(sink.ops) != len(wantOps) { + t.Fatalf("expected %d dispatches, got %d (%v)", len(wantOps), len(sink.ops), sink.ops) + } + + for i, op := range wantOps { + if sink.ops[i] != op { + t.Fatalf("op[%d] = %q, want %q", i, sink.ops[i], op) + } + + if sink.objs[i] != "s1" { + t.Fatalf("obj[%d] = %q, want s1", i, sink.objs[i]) + } + } +} + +func TestManager_InitialListNotDispatched(t *testing.T) { + f, m := newTestManager() + + sink := &recSink{} + m.RegisterSink(sink) + + syncOrFatal(t, m, "a", secretGVK) + + handler := f.handlerFor(t, secretGVK) + + // The informer replays every pre-existing object as an add with + // isInInitialList=true when it syncs (startup and re-arm relist). Those + // are not creations and must not be dispatched. + obj := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "s1", Namespace: "ns"}} + handler.OnAdd(obj, true) + + if len(sink.ops) != 0 { + t.Fatalf("expected no dispatch for initial-list add, got %v", sink.ops) + } + + // A genuine create afterwards still dispatches. + handler.OnAdd(obj, false) + + if len(sink.ops) != 1 || sink.ops[0] != OperationCreate { + t.Fatalf("expected a single OperationCreate, got %v", sink.ops) + } +} + +func TestManager_ResyncUpdatesNotDispatched(t *testing.T) { + f, m := newTestManager() + + sink := &recSink{} + m.RegisterSink(sink) + + syncOrFatal(t, m, "a", secretGVK) + + handler := f.handlerFor(t, secretGVK) + + // A periodic resync replays the cached object against itself: the + // resourceVersion is unchanged, so no sink must be notified. + obj := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "s1", Namespace: "ns", ResourceVersion: "7"}} + handler.OnUpdate(obj, obj) + + if len(sink.ops) != 0 { + t.Fatalf("expected no dispatch for a resync replay, got %v", sink.ops) + } + + // A genuine write bumps the resourceVersion and dispatches. + updated := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Name: "s1", Namespace: "ns", ResourceVersion: "8"}} + handler.OnUpdate(obj, updated) + + if len(sink.ops) != 1 || sink.ops[0] != OperationUpdate { + t.Fatalf("expected a single OperationUpdate, got %v", sink.ops) + } +} + +func TestManager_ForbiddenKindNotArmed(t *testing.T) { + f := &fakeFactory{} + authz := &fakeAuthorizer{denied: map[string]struct{}{"secrets": {}}} + m := NewManager(f.new, authz, testMapper(), logr.Discard()) + + err := m.Sync(context.Background(), "a", unfiltered(secretGVK, configMapGVK)) + if err == nil || !strings.Contains(err.Error(), "not permitted to list secrets") { + t.Fatalf("expected a permission error naming the resource, got %v", err) + } + + if f.createdFor(secretGVK) != 0 { + t.Fatalf("no watch must be created for a kind the controller cannot watch") + } + + // Permitted kinds in the same Sync are unaffected. + if f.createdFor(configMapGVK) != 1 { + t.Fatalf("configmap watch should have been created, created=%d", f.createdFor(configMapGVK)) + } + + // Already-armed kinds are not re-checked on subsequent Syncs. + calls := authz.calls + + syncOrFatal(t, m, "b", configMapGVK) + + if authz.calls != calls { + t.Fatalf("armed kinds must not be re-checked, calls went %d -> %d", calls, authz.calls) + } + + // Once permissions are granted, the next Sync arms the kind. + authz.denied = nil + syncOrFatal(t, m, "a", secretGVK, configMapGVK) + + if f.createdFor(secretGVK) != 1 { + t.Fatalf("secret watch should be created after permissions are granted, created=%d", f.createdFor(secretGVK)) + } +}