diff --git a/cmd/manager/main.go b/cmd/manager/main.go index f69cc9005..e7dd070f9 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -65,6 +65,7 @@ import ( "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations/failover" "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations/inflight" "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations/quota" + "github.com/cobaltcore-dev/cortex/pkg/cache" "github.com/cobaltcore-dev/cortex/pkg/conf" "github.com/cobaltcore-dev/cortex/pkg/monitoring" "github.com/cobaltcore-dev/cortex/pkg/multicluster" @@ -391,6 +392,10 @@ func main() { }, } multiclusterClientConfig := conf.GetConfigOrDie[multicluster.ClientConfig]() + + if c := conf.GetConfigOrDie[cache.RootConfig](); c.Cache.Enabled { + multiclusterClient.Wrappers = append(multiclusterClient.Wrappers, cache.NewWrapper(c.Cache)) + } if err := multiclusterClient.InitFromConf(ctx, mgr, multiclusterClientConfig); err != nil { setupLog.Error(err, "unable to initialize multicluster client") os.Exit(1) diff --git a/helm/bundles/cortex-nova/values.yaml b/helm/bundles/cortex-nova/values.yaml index cec441eaa..82cc898b1 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/helm/bundles/cortex-nova/values.yaml @@ -74,6 +74,12 @@ kvm: # Useful when rolling out a new KVM region before it is production-ready. criticalAlerts: true +# Cache configuration for the transparent in-process overlay cache. +cache: + enabled: true + gvks: + - cortex.cloud/v1alpha1/Reservation + cortex: &cortex crd: {enable: false} # Disable the default ServiceMonitor and metrics service from the kubebuilder stack. @@ -185,7 +191,7 @@ cortex-scheduling-controllers: # How long after a VM is allocated to a reservation before it is expected to appear # on the target host; allocations not confirmed within this window are removed allocationGracePeriod: "15m" - # How long to wait after detecting host over-subscription before evicting reservation slots. + # How long to wait after detecting host over-subscription before evicting reservation slots. # Gives other controllers (e.g. failover) time to self-heal. oversubscriptionGracePeriod: "3m" # Minimum time between consecutive over-subscription checks for the same host. diff --git a/internal/shim/placement/field_index_test.go b/internal/shim/placement/field_index_test.go index 1f74ad9a3..a81ec2879 100644 --- a/internal/shim/placement/field_index_test.go +++ b/internal/shim/placement/field_index_test.go @@ -50,6 +50,10 @@ type stubCluster struct { func (s *stubCluster) GetClient() client.Client { return s.cl } func (s *stubCluster) GetCache() cache.Cache { return s.cache } +// GetFieldIndexer returns the cache, which implements client.FieldIndexer via +// its IndexField method — mirroring an unwrapped controller-runtime cluster. +func (s *stubCluster) GetFieldIndexer() client.FieldIndexer { return s.cache } + type stubManager struct { manager.Manager } diff --git a/pkg/cache/client.go b/pkg/cache/client.go new file mode 100644 index 000000000..6fb5745de --- /dev/null +++ b/pkg/cache/client.go @@ -0,0 +1,578 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package cache + +import ( + "context" + "errors" + "strconv" + "sync" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/fields" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/cluster" + "sigs.k8s.io/controller-runtime/pkg/manager" +) + +// entry is a single overlaid object with the bookkeeping needed for eviction. +type entry struct { + obj client.Object + uid types.UID + resourceVersion string + deleted bool // tombstone: object was deleted through the caching client + expiresAt time.Time +} + +// resourceVersionAtLeast reports whether observed >= cached, treating +// ResourceVersions as opaque monotonically increasing integers (as the +// kubernetes apiserver guarantees per resource). Unparsable or empty values +// are treated conservatively: an empty cached RV means "evict on any sighting". +func resourceVersionAtLeast(observed, cached string) bool { + if cached == "" { + return true + } + if observed == "" { + return false + } + oi, oerr := strconv.ParseUint(observed, 10, 64) + ci, cerr := strconv.ParseUint(cached, 10, 64) + if oerr != nil || cerr != nil { + // Fall back to string comparison if not integers. + return observed >= cached + } + return oi >= ci +} + +// defaultTTL is used when Config.TTL is zero. +const defaultTTL = 2 * time.Minute + +// Overlay wraps an inner client.Client with a transparent in-process +// overlay. Writes populate the overlay; reads merge the overlay with the inner +// (informer-backed) result; entries are evicted once the real object appears in +// an informer (see runnable.go) or after TTL expiry. +// +// It embeds client.Client so all methods not overridden below are delegated to +// the inner client unchanged. +// +// The local overlay maps each cached GVK to a set of entries keyed by +// namespace/name. An entry is either live (a pending write not yet visible in +// the informer) or a tombstone (a Delete that has not yet propagated). Reads +// merge the inner result with these entries: tombstones suppress objects; +// live entries override or supplement the inner result. +type Overlay struct { + client.Client // inner client, used for delegation + + informerCache cache.Cache // used by Start to attach eviction event handlers + scheme *runtime.Scheme + ttl time.Duration + gvks map[schema.GroupVersionKind]bool + + mu sync.RWMutex + byGVK map[schema.GroupVersionKind]map[client.ObjectKey]*entry + indexers map[schema.GroupVersionKind]map[string]client.IndexerFunc + + // writeLocks serializes writes to the same object so the inner call and the + // overlay mutation are atomic per object, keeping the overlay from falling + // behind the apiserver under concurrent writes. + writeLocks *keyedMutex +} + +// WrapCluster wraps inner with a transparent in-process overlay cache and +// returns the wrapped cluster.Cluster together with the overlay's lifecycle +// Runnable. +// +// The returned cluster.Cluster delegates everything to inner except GetClient +// and GetFieldIndexer, which return the overlay client so per-cluster reads and +// writes flow through the cache. GVK strings in conf are formatted as +// "//" and resolved against inner.GetScheme(). +// +// The returned manager.Runnable is the overlay itself: it attaches informer +// eviction handlers (reading informers from inner.GetCache()) and runs the TTL +// cleanup loop. The caller MUST add it to the manager. It does NOT re-Start the +// inner cluster — the manager owns the inner cluster's lifecycle separately. +func WrapCluster(inner cluster.Cluster, conf Config) (cluster.Cluster, manager.Runnable, error) { + scheme := inner.GetScheme() + gvks, err := resolveGVKs(scheme, conf.GVKs) + if err != nil { + return nil, nil, err + } + ttl := conf.TTL.Duration + if ttl <= 0 { + ttl = defaultTTL + } + cc := &Overlay{ + Client: inner.GetClient(), + informerCache: inner.GetCache(), + scheme: scheme, + ttl: ttl, + gvks: gvks, + byGVK: make(map[schema.GroupVersionKind]map[client.ObjectKey]*entry), + indexers: make(map[schema.GroupVersionKind]map[string]client.IndexerFunc), + writeLocks: newKeyedMutex(), + } + return &overlayCluster{Cluster: inner, cache: cc}, cc, nil +} + +// resolveGVKs maps "//" strings to GVKs via the scheme's +// known types. Mirrors the resolution logic of multicluster.InitFromConf, but +// stays local to this package. +func resolveGVKs(scheme *runtime.Scheme, gvkStrs []string) (map[schema.GroupVersionKind]bool, error) { + byStr := make(map[string]schema.GroupVersionKind) + for gvk := range scheme.AllKnownTypes() { + byStr[gvk.GroupVersion().String()+"/"+gvk.Kind] = gvk + } + out := make(map[schema.GroupVersionKind]bool, len(gvkStrs)) + for _, s := range gvkStrs { + gvk, ok := byStr[s] + if !ok { + return nil, errors.New("cache: no gvk registered in scheme for " + s) + } + out[gvk] = true + } + return out, nil +} + +// gvkFor resolves the GVK of obj and reports whether it is cached. +func (c *Overlay) gvkFor(obj runtime.Object) (schema.GroupVersionKind, bool) { + gvks, _, err := c.scheme.ObjectKinds(obj) + if err != nil || len(gvks) != 1 { + return schema.GroupVersionKind{}, false + } + gvk := gvks[0] + return gvk, c.gvks[gvk] +} + +// itemGVKForList resolves the singular item GVK for a list object and reports +// whether that item GVK is cached. The list GVK's Kind ends with "List". +func (c *Overlay) itemGVKForList(list client.ObjectList) (schema.GroupVersionKind, bool) { + gvks, _, err := c.scheme.ObjectKinds(list) + if err != nil || len(gvks) != 1 { + return schema.GroupVersionKind{}, false + } + gvk := gvks[0] + if kind, ok := trimListSuffix(gvk.Kind); ok { + gvk.Kind = kind + } + return gvk, c.gvks[gvk] +} + +// trimListSuffix strips a trailing "List" from a Kind, reporting whether it did. +func trimListSuffix(kind string) (string, bool) { + const suffix = "List" + if len(kind) > len(suffix) && kind[len(kind)-len(suffix):] == suffix { + return kind[:len(kind)-len(suffix)], true + } + return kind, false +} + +// upsert stores a live (non-tombstone) entry for the object. +func (c *Overlay) upsert(gvk schema.GroupVersionKind, obj client.Object) { + c.mu.Lock() + defer c.mu.Unlock() + c.ensureGVK(gvk) + c.byGVK[gvk][client.ObjectKeyFromObject(obj)] = &entry{ + obj: obj.DeepCopyObject().(client.Object), + uid: obj.GetUID(), + resourceVersion: obj.GetResourceVersion(), + deleted: false, + expiresAt: time.Now().Add(c.ttl), + } +} + +// tombstone marks the object as deleted in the overlay so it is filtered out +// of reads until the deletion is observed in an informer. +func (c *Overlay) tombstone(gvk schema.GroupVersionKind, obj client.Object) { + c.mu.Lock() + defer c.mu.Unlock() + c.ensureGVK(gvk) + c.byGVK[gvk][client.ObjectKeyFromObject(obj)] = &entry{ + obj: obj.DeepCopyObject().(client.Object), + uid: obj.GetUID(), + resourceVersion: obj.GetResourceVersion(), + deleted: true, + expiresAt: time.Now().Add(c.ttl), + } +} + +// evictIfSeen removes the overlay entry for obj if the informer-observed object +// matches by UID and its ResourceVersion is at least as new as the cached one. +func (c *Overlay) evictIfSeen(gvk schema.GroupVersionKind, obj client.Object) { + c.mu.Lock() + defer c.mu.Unlock() + entries, ok := c.byGVK[gvk] + if !ok { + return + } + key := client.ObjectKeyFromObject(obj) + e, ok := entries[key] + if !ok { + return + } + // Only evict when the informer sees the same object generation (by UID) at + // a ResourceVersion >= the one we cached. Otherwise the informer might be + // showing an older revision than our pending write. + if e.uid != "" && obj.GetUID() != "" && e.uid != obj.GetUID() { + return + } + if !resourceVersionAtLeast(obj.GetResourceVersion(), e.resourceVersion) { + return + } + delete(entries, key) +} + +// getEntry returns the overlay entry for the key, if present. +func (c *Overlay) getEntry(gvk schema.GroupVersionKind, key client.ObjectKey) (*entry, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + entries, ok := c.byGVK[gvk] + if !ok { + return nil, false + } + e, ok := entries[key] + return e, ok +} + +// cleanupExpired removes entries whose TTL has passed. +func (c *Overlay) cleanupExpired(now time.Time) { + c.mu.Lock() + defer c.mu.Unlock() + for _, entries := range c.byGVK { + for key, e := range entries { + if now.After(e.expiresAt) { + delete(entries, key) + } + } + } +} + +// registerIndex captures an IndexerFunc for a field so overlay entries can be +// matched against MatchingFields queries. +func (c *Overlay) registerIndex(gvk schema.GroupVersionKind, field string, fn client.IndexerFunc) { + c.mu.Lock() + defer c.mu.Unlock() + if c.indexers[gvk] == nil { + c.indexers[gvk] = make(map[string]client.IndexerFunc) + } + c.indexers[gvk][field] = fn +} + +// ensureGVK initialises the per-GVK entry map if absent. Callers must hold the write lock. +func (c *Overlay) ensureGVK(gvk schema.GroupVersionKind) { + if c.byGVK[gvk] == nil { + c.byGVK[gvk] = make(map[client.ObjectKey]*entry) + } +} + +// overlayList merges the overlay entries for the GVK into the informer result, +// deduplicating by client.ObjectKey (overlay wins), dropping tombstones, and filtering +// overlay-only entries against the list options' label and field selectors. +func (c *Overlay) overlayList(gvk schema.GroupVersionKind, existing []runtime.Object, lo *client.ListOptions) []runtime.Object { + c.mu.RLock() + defer c.mu.RUnlock() + entries := c.byGVK[gvk] + if len(entries) == 0 { + return existing + } + + result := make([]runtime.Object, 0, len(existing)+len(entries)) + // Track which overlay keys are handled so overlay-only entries can be added. + handled := make(map[client.ObjectKey]bool, len(entries)) + + for _, item := range existing { + obj, ok := item.(client.Object) + if !ok { + result = append(result, item) + continue + } + key := client.ObjectKeyFromObject(obj) + e, present := entries[key] + if !present { + result = append(result, item) + continue + } + handled[key] = true + // Overlay wins over the informer result for the same key. + if e.deleted { + // Tombstone: drop the object entirely. + continue + } + // The inner result matched the query against the informer's (old) field + // values. Re-check the overlay version: if a write changed a queried + // field, the overlay object no longer belongs in this result set. + if !c.matchesLocked(gvk, e.obj, lo) { + continue + } + result = append(result, e.obj.DeepCopyObject()) + } + + // Add overlay-only entries (not present in the informer result) that match + // the list options. + for key, e := range entries { + if handled[key] { + continue + } + if e.deleted { + continue + } + if !c.matchesLocked(gvk, e.obj, lo) { + continue + } + result = append(result, e.obj.DeepCopyObject()) + } + return result +} + +// matchesLocked reports whether obj satisfies the list options' namespace, +// label and field selectors. Callers must hold at least the read lock. +func (c *Overlay) matchesLocked(gvk schema.GroupVersionKind, obj client.Object, lo *client.ListOptions) bool { + if lo == nil { + return true + } + if lo.Namespace != "" && obj.GetNamespace() != lo.Namespace { + return false + } + if lo.LabelSelector != nil && !lo.LabelSelector.Matches(labels.Set(obj.GetLabels())) { + return false + } + if lo.FieldSelector != nil && !lo.FieldSelector.Empty() { + set := c.fieldSetLocked(gvk, obj) + if !lo.FieldSelector.Matches(set) { + return false + } + } + return true +} + +// fieldSetLocked builds a fields.Set for obj using the registered IndexerFuncs +// for the GVK. Callers must hold at least the read lock. +func (c *Overlay) fieldSetLocked(gvk schema.GroupVersionKind, obj client.Object) fields.Set { + set := fields.Set{} + for field, fn := range c.indexers[gvk] { + for _, v := range fn(obj) { + // A field selector matches a single value; take the first indexed + // value for the field (mirrors controller-runtime cache behaviour). + set[field] = v + break + } + } + return set +} + +// Create delegates to the inner client and, on success for a cached GVK, adds +// the object to the overlay. +func (c *Overlay) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error { + gvk, cached := c.gvkFor(obj) + if !cached { + return c.Client.Create(ctx, obj, opts...) + } + unlock := c.writeLocks.lock(lockKey{gvk: gvk, key: client.ObjectKeyFromObject(obj)}) + defer unlock() + if err := c.Client.Create(ctx, obj, opts...); err != nil { + return err + } + c.upsert(gvk, obj) + return nil +} + +// Update delegates to the inner client and, on success for a cached GVK, +// refreshes the overlay entry. +func (c *Overlay) Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error { + gvk, cached := c.gvkFor(obj) + if !cached { + return c.Client.Update(ctx, obj, opts...) + } + unlock := c.writeLocks.lock(lockKey{gvk: gvk, key: client.ObjectKeyFromObject(obj)}) + defer unlock() + if err := c.Client.Update(ctx, obj, opts...); err != nil { + return err + } + c.upsert(gvk, obj) + return nil +} + +// Patch delegates to the inner client and, on success for a cached GVK, +// refreshes the overlay entry with the patched object. +func (c *Overlay) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + gvk, cached := c.gvkFor(obj) + if !cached { + return c.Client.Patch(ctx, obj, patch, opts...) + } + unlock := c.writeLocks.lock(lockKey{gvk: gvk, key: client.ObjectKeyFromObject(obj)}) + defer unlock() + if err := c.Client.Patch(ctx, obj, patch, opts...); err != nil { + return err + } + c.upsert(gvk, obj) + return nil +} + +// Delete delegates to the inner client and, on success for a cached GVK, stores +// a tombstone in the overlay. The object is NOT immediately removed from the +// local map; it stays as a deleted=true entry until the deletion propagates +// through the informer (which triggers eviction) or the TTL expires. This +// ensures that reads between the Delete call and the informer event correctly +// return NotFound rather than serving a stale object from the informer cache. +func (c *Overlay) Delete(ctx context.Context, obj client.Object, opts ...client.DeleteOption) error { + gvk, cached := c.gvkFor(obj) + if !cached { + return c.Client.Delete(ctx, obj, opts...) + } + unlock := c.writeLocks.lock(lockKey{gvk: gvk, key: client.ObjectKeyFromObject(obj)}) + defer unlock() + if err := c.Client.Delete(ctx, obj, opts...); err != nil { + return err + } + c.tombstone(gvk, obj) + return nil +} + +// DeleteAllOf delegates to the inner client and, on success for a cached GVK, +// tombstones all overlay entries that match the delete options. Objects that +// live only in the informer cache (not in the overlay) will be evicted +// naturally once the deletion propagates through the informer. +func (c *Overlay) DeleteAllOf(ctx context.Context, obj client.Object, opts ...client.DeleteAllOfOption) error { + gvk, cached := c.gvkFor(obj) + if !cached { + return c.Client.DeleteAllOf(ctx, obj, opts...) + } + if err := c.Client.DeleteAllOf(ctx, obj, opts...); err != nil { + return err + } + dao := &client.DeleteAllOfOptions{} + dao.ApplyOptions(opts) + c.mu.Lock() + defer c.mu.Unlock() + for key, e := range c.byGVK[gvk] { + if !c.matchesLocked(gvk, e.obj, &dao.ListOptions) { + continue + } + c.byGVK[gvk][key] = &entry{ + obj: e.obj, + uid: e.uid, + resourceVersion: e.resourceVersion, + deleted: true, + expiresAt: time.Now().Add(c.ttl), + } + } + return nil +} + +// Get delegates to the inner client, then applies the overlay: a tombstone +// yields NotFound; a live overlay entry overrides the inner result; and an +// overlay entry can satisfy a Get that the inner client reports as NotFound. +func (c *Overlay) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + gvk, cached := c.gvkFor(obj) + if !cached { + return c.Client.Get(ctx, key, obj, opts...) + } + err := c.Client.Get(ctx, key, obj, opts...) + if err != nil && !apierrors.IsNotFound(err) { + return err + } + e, ok := c.getEntry(gvk, key) + if !ok { + // No overlay entry: return the inner result (value or NotFound) as-is. + return err + } + if e.deleted { + return apierrors.NewNotFound(schema.GroupResource{Group: gvk.Group, Resource: gvk.Kind}, key.Name) + } + // Live overlay entry: copy it into obj, overriding the inner result. + // Deep-copy the cached object first so scheme.Convert cannot alias the + // overlay entry's maps, slices, or metadata into the caller's obj (which + // would let callers mutate the shared cache entry). + if cpErr := c.scheme.Convert(e.obj.DeepCopyObject(), obj, nil); cpErr != nil { + return cpErr + } + return nil +} + +// List delegates to the inner client, then merges the overlay into the result. +func (c *Overlay) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { + itemGVK, cached := c.itemGVKForList(list) + if !cached { + return c.Client.List(ctx, list, opts...) + } + if err := c.Client.List(ctx, list, opts...); err != nil { + return err + } + items, err := meta.ExtractList(list) + if err != nil { + return err + } + lo := &client.ListOptions{} + lo.ApplyOptions(opts) + merged := c.overlayList(itemGVK, items, lo) + return meta.SetList(list, merged) +} + +// IndexField delegates to the inner client (if it is a FieldIndexer) and also +// registers the IndexerFunc with the overlay so overlay entries can be matched +// against MatchingFields. +func (c *Overlay) IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error { + if err := c.informerCache.IndexField(ctx, obj, field, extractValue); err != nil { + return err + } + if gvk, cached := c.gvkFor(obj); cached { + c.registerIndex(gvk, field, extractValue) + } + return nil +} + +// Status returns a status writer that mirrors status Update/Patch writes for +// cached GVKs into the overlay. +func (c *Overlay) Status() client.StatusWriter { + return &statusWriter{c: c, inner: c.Client.Status()} +} + +// statusWriter wraps the inner status writer and reflects status writes into +// the overlay for cached GVKs. +type statusWriter struct { + c *Overlay + inner client.StatusWriter +} + +func (s *statusWriter) Create(ctx context.Context, obj, subResource client.Object, opts ...client.SubResourceCreateOption) error { + return s.inner.Create(ctx, obj, subResource, opts...) +} + +func (s *statusWriter) Update(ctx context.Context, obj client.Object, opts ...client.SubResourceUpdateOption) error { + gvk, cached := s.c.gvkFor(obj) + if !cached { + return s.inner.Update(ctx, obj, opts...) + } + unlock := s.c.writeLocks.lock(lockKey{gvk: gvk, key: client.ObjectKeyFromObject(obj)}) + defer unlock() + if err := s.inner.Update(ctx, obj, opts...); err != nil { + return err + } + s.c.upsert(gvk, obj) + return nil +} + +func (s *statusWriter) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + gvk, cached := s.c.gvkFor(obj) + if !cached { + return s.inner.Patch(ctx, obj, patch, opts...) + } + unlock := s.c.writeLocks.lock(lockKey{gvk: gvk, key: client.ObjectKeyFromObject(obj)}) + defer unlock() + if err := s.inner.Patch(ctx, obj, patch, opts...); err != nil { + return err + } + s.c.upsert(gvk, obj) + return nil +} + +func (s *statusWriter) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...client.SubResourceApplyOption) error { + return s.inner.Apply(ctx, obj, opts...) +} diff --git a/pkg/cache/client_test.go b/pkg/cache/client_test.go new file mode 100644 index 000000000..f1d43dac6 --- /dev/null +++ b/pkg/cache/client_test.go @@ -0,0 +1,1000 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package cache + +import ( + "context" + "errors" + goruntime "runtime" + "strconv" + "sync" + "testing" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + toolscachek8s "k8s.io/client-go/tools/cache" + ccache "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/cluster" + "sigs.k8s.io/controller-runtime/pkg/manager" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" +) + +const azIndexField = "spec.availabilityZone" + +// --- shared test infrastructure --- + +func testScheme(t *testing.T) *runtime.Scheme { + t.Helper() + scheme := runtime.NewScheme() + if err := v1alpha1.AddToScheme(scheme); err != nil { + t.Fatalf("add to scheme: %v", err) + } + return scheme +} + +func newReservation(name, az, rv string) *v1alpha1.Reservation { + return &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + UID: types.UID("uid-" + name), + ResourceVersion: rv, + }, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + AvailabilityZone: az, + }, + } +} + +// fakeInformer is a controllable informer that records handlers and lets tests +// fire Add/Update events to trigger eviction. +type fakeInformer struct { + ccache.Informer + mu sync.Mutex + handlers []toolscachek8s.ResourceEventHandler +} + +func (f *fakeInformer) AddEventHandler(h toolscachek8s.ResourceEventHandler) (toolscachek8s.ResourceEventHandlerRegistration, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.handlers = append(f.handlers, h) + return nil, nil +} + +func (f *fakeInformer) fireAdd(obj any) { + f.mu.Lock() + handlers := append([]toolscachek8s.ResourceEventHandler(nil), f.handlers...) + f.mu.Unlock() + for _, h := range handlers { + h.OnAdd(obj, false) + } +} + +func (f *fakeInformer) fireUpdate(oldObj, newObj any) { + f.mu.Lock() + handlers := append([]toolscachek8s.ResourceEventHandler(nil), f.handlers...) + f.mu.Unlock() + for _, h := range handlers { + h.OnUpdate(oldObj, newObj) + } +} + +// fakeCache is a minimal cache.Cache that returns a single fakeInformer for any +// GetInformer call, so the Overlay runnable can wire eviction handlers. +type fakeCache struct { + ccache.Cache + inf *fakeInformer +} + +func (f *fakeCache) GetInformer(_ context.Context, _ client.Object, _ ...ccache.InformerGetOption) (ccache.Informer, error) { + return f.inf, nil +} + +func (f *fakeCache) IndexField(_ context.Context, _ client.Object, _ string, _ client.IndexerFunc) error { + return nil +} + +// fakeCluster composes a fake client.Client with a fakeCache to satisfy the +// cluster.Cluster interface consumed by cache.WrapCluster. +type fakeCluster struct { + cluster.Cluster + client client.Client + cache *fakeCache + scheme *runtime.Scheme +} + +func (f *fakeCluster) GetClient() client.Client { return f.client } +func (f *fakeCluster) GetCache() ccache.Cache { return f.cache } +func (f *fakeCluster) GetScheme() *runtime.Scheme { return f.scheme } + +// fakeClient exposes the underlying informer so eviction tests can fire events, +// while itself acting as a cluster.Cluster wrapping the fake client.Client. +type fakeClient struct { + *fakeCluster + inf *fakeInformer +} + +func newTestClient(t *testing.T, objs ...client.Object) *fakeClient { + t.Helper() + scheme := testScheme(t) + inf := &fakeInformer{} + inner := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objs...). + WithStatusSubresource(&v1alpha1.Reservation{}). + WithIndex(&v1alpha1.Reservation{}, azIndexField, func(obj client.Object) []string { + res, ok := obj.(*v1alpha1.Reservation) + if !ok || res.Spec.AvailabilityZone == "" { + return nil + } + return []string{res.Spec.AvailabilityZone} + }). + Build() + return &fakeClient{ + fakeCluster: &fakeCluster{ + client: inner, + cache: &fakeCache{inf: inf}, + scheme: scheme, + }, + inf: inf, + } +} + +// errClient wraps a client.Client and injects configurable errors into +// mutating/read operations, so the error-propagation paths of Overlay +// (which must not touch the overlay on failure) can be exercised. +type errClient struct { + client.Client + createErr error + updateErr error + patchErr error + deleteErr error + deleteAllOfErr error + getErr error + listErr error +} + +func (e *errClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error { + if e.createErr != nil { + return e.createErr + } + return e.Client.Create(ctx, obj, opts...) +} + +func (e *errClient) Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error { + if e.updateErr != nil { + return e.updateErr + } + return e.Client.Update(ctx, obj, opts...) +} + +func (e *errClient) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if e.patchErr != nil { + return e.patchErr + } + return e.Client.Patch(ctx, obj, patch, opts...) +} + +func (e *errClient) Delete(ctx context.Context, obj client.Object, opts ...client.DeleteOption) error { + if e.deleteErr != nil { + return e.deleteErr + } + return e.Client.Delete(ctx, obj, opts...) +} + +func (e *errClient) DeleteAllOf(ctx context.Context, obj client.Object, opts ...client.DeleteAllOfOption) error { + if e.deleteAllOfErr != nil { + return e.deleteAllOfErr + } + return e.Client.DeleteAllOf(ctx, obj, opts...) +} + +func (e *errClient) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if e.getErr != nil { + return e.getErr + } + return e.Client.Get(ctx, key, obj, opts...) +} + +func (e *errClient) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { + if e.listErr != nil { + return e.listErr + } + return e.Client.List(ctx, list, opts...) +} + +// forceInnerAZ writes a divergent AvailabilityZone directly to the inner +// client, bypassing the caching wrapper. Used to prove reads are served from +// the overlay, not the inner client. +func forceInnerAZ(t *testing.T, inner client.Client, name, az string) { + t.Helper() + var cur v1alpha1.Reservation + if err := inner.Get(context.Background(), types.NamespacedName{Name: name}, &cur); err != nil { + t.Fatalf("forceInnerAZ get: %v", err) + } + cur.Spec.AvailabilityZone = az + if err := inner.Update(context.Background(), &cur); err != nil { + t.Fatalf("forceInnerAZ update: %v", err) + } +} + +// forceInnerStatusHost writes a divergent status Host directly to the inner client. +func forceInnerStatusHost(t *testing.T, inner client.Client, name, host string) { + t.Helper() + var cur v1alpha1.Reservation + if err := inner.Get(context.Background(), types.NamespacedName{Name: name}, &cur); err != nil { + t.Fatalf("forceInnerStatusHost get: %v", err) + } + cur.Status.Host = host + if err := inner.Status().Update(context.Background(), &cur); err != nil { + t.Fatalf("forceInnerStatusHost update: %v", err) + } +} + +// unknownObject is a client.Object whose type is not registered in the test +// scheme, used to exercise the "unresolvable GVK" branches. +type unknownObject struct { + metav1.TypeMeta + metav1.ObjectMeta +} + +func (u *unknownObject) DeepCopyObject() runtime.Object { return u } + +func reservationConfig() Config { + return Config{ + GVKs: []string{"cortex.cloud/v1alpha1/Reservation"}, + TTL: metav1.Duration{Duration: 2 * time.Minute}, + } +} + +// clusterFor wraps a client.Client as a cluster.Cluster so it can be passed to +// WrapCluster. Used by tests that inject a custom client.Client (e.g. errClient, +// orderingClient) below the overlay. +func clusterFor(t *testing.T, inner client.Client) cluster.Cluster { + t.Helper() + return &fakeCluster{ + client: inner, + cache: &fakeCache{inf: &fakeInformer{}}, + scheme: testScheme(t), + } +} + +// cachingFrom builds a *Overlay over the given cluster.Cluster. +func cachingFrom(t *testing.T, cl cluster.Cluster, conf Config) *Overlay { + t.Helper() + wrapped, runnable, err := WrapCluster(cl, conf) + if err != nil { + t.Fatalf("New: %v", err) + } + cc, ok := wrapped.GetClient().(*Overlay) + if !ok { + t.Fatalf("WrapCluster did not return an Overlay") + } + if _, ok := runnable.(*Overlay); !ok { + t.Fatalf("WrapCluster did not return a *Overlay runnable") + } + return cc +} + +// newCaching builds a *Overlay over the given fake cluster using the +// default reservation config. +func newCaching(t *testing.T, inner *fakeClient) *Overlay { + t.Helper() + return cachingFrom(t, inner.fakeCluster, reservationConfig()) +} + +func listReservations(t *testing.T, c client.Client, opts ...client.ListOption) []v1alpha1.Reservation { + t.Helper() + var list v1alpha1.ReservationList + if err := c.List(context.Background(), &list, opts...); err != nil { + t.Fatalf("List: %v", err) + } + return list.Items +} + +func reservationGVK() schema.GroupVersionKind { + return v1alpha1.GroupVersion.WithKind("Reservation") +} + +func waitFor(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("condition not met within timeout") +} + +// --- constructor tests --- + +func TestNewUnknownGVKError(t *testing.T) { + _, _, err := WrapCluster(newTestClient(t).fakeCluster, Config{ + GVKs: []string{"cortex.cloud/v1alpha1/DoesNotExist"}, + }) + if err == nil { + t.Fatalf("expected error for unknown GVK, got nil") + } +} + +func TestNewDefaultTTL(t *testing.T) { + c := cachingFrom(t, newTestClient(t).fakeCluster, Config{ + GVKs: []string{"cortex.cloud/v1alpha1/Reservation"}, + }) + if c.ttl != defaultTTL { + t.Fatalf("expected ttl %v, got %v", defaultTTL, c.ttl) + } +} + +func TestNewExplicitTTL(t *testing.T) { + c := cachingFrom(t, newTestClient(t).fakeCluster, Config{ + GVKs: []string{"cortex.cloud/v1alpha1/Reservation"}, + TTL: metav1.Duration{Duration: 90 * time.Second}, + }) + if c.ttl != 90*time.Second { + t.Fatalf("expected ttl 90s, got %v", c.ttl) + } +} + +func TestNewWrapsClusterClientAndIndexer(t *testing.T) { + inner := newTestClient(t) + wrapped, runnable, err := WrapCluster(inner.fakeCluster, reservationConfig()) + if err != nil { + t.Fatalf("New: %v", err) + } + // GetClient and GetFieldIndexer must both return the overlay so per-cluster + // reads/writes and IndexField flow through the cache. + cc, ok := wrapped.GetClient().(*Overlay) + if !ok { + t.Fatalf("GetClient did not return the overlay") + } + if wrapped.GetFieldIndexer() != client.FieldIndexer(cc) { + t.Fatalf("GetFieldIndexer did not return the overlay") + } + // The returned Runnable is the same overlay. + if runnable != manager.Runnable(cc) { + t.Fatalf("returned Runnable is not the overlay") + } + // Everything else delegates to the inner cluster. + if wrapped.GetScheme() != inner.GetScheme() { + t.Fatalf("GetScheme did not delegate to inner") + } + if wrapped.GetCache() != inner.GetCache() { + t.Fatalf("GetCache did not delegate to inner") + } +} + +// --- overlay behaviour tests --- + +func TestCreateThenGetVisible(t *testing.T) { + c := newCaching(t, newTestClient(t)) + + r := newReservation("res-1", "az-1", "") + if err := c.Create(context.Background(), r); err != nil { + t.Fatalf("Create: %v", err) + } + + var got v1alpha1.Reservation + if err := c.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err != nil { + t.Fatalf("Get after create: %v", err) + } + if got.Spec.AvailabilityZone != "az-1" { + t.Fatalf("expected az-1, got %q", got.Spec.AvailabilityZone) + } +} + +func TestOverlayWhenInnerEmpty(t *testing.T) { + c := newCaching(t, newTestClient(t)) + c.upsert(reservationGVK(), newReservation("res-2", "az-1", "5")) + + if items := listReservations(t, c); len(items) != 1 || items[0].Name != "res-2" { + t.Fatalf("expected overlay entry res-2, got %+v", items) + } +} + +func TestEviction(t *testing.T) { + const cachedRV = "10" + cases := []struct { + name string + useUpdate bool + observedUID string + observedRV string + wantEvicted bool + }{ + {name: "add older RV keeps", observedRV: "9", wantEvicted: false}, + {name: "add equal RV evicts", observedRV: "10", wantEvicted: true}, + {name: "add newer RV evicts", observedRV: "11", wantEvicted: true}, + {name: "update newer RV evicts", useUpdate: true, observedRV: "11", wantEvicted: true}, + {name: "update older RV keeps", useUpdate: true, observedRV: "9", wantEvicted: false}, + {name: "uid mismatch keeps", observedUID: "uid-other", observedRV: "11", wantEvicted: false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + inner := newTestClient(t) + c := newCaching(t, inner) + + ctx := t.Context() + go func() { + if err := c.Start(ctx); err != nil && ctx.Err() == nil { + t.Errorf("c.Start: %v", err) + } + }() + waitFor(t, func() bool { + inner.inf.mu.Lock() + defer inner.inf.mu.Unlock() + return len(inner.inf.handlers) > 0 + }) + + c.upsert(reservationGVK(), newReservation("res-3", "az-1", cachedRV)) + + observed := newReservation("res-3", "az-1", tc.observedRV) + if tc.observedUID != "" { + observed.UID = types.UID(tc.observedUID) + } + if tc.useUpdate { + inner.inf.fireUpdate(nil, observed) + } else { + inner.inf.fireAdd(observed) + } + + _, present := c.getEntry(reservationGVK(), client.ObjectKey{Name: "res-3"}) + if present == tc.wantEvicted { + t.Fatalf("evicted=%v, want evicted=%v", !present, tc.wantEvicted) + } + }) + } +} + +func TestEvictionIgnoresNonObject(t *testing.T) { + inner := newTestClient(t) + c := newCaching(t, inner) + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go func() { + if err := c.Start(ctx); err != nil && ctx.Err() == nil { + t.Errorf("c.Start: %v", err) + } + }() + waitFor(t, func() bool { + inner.inf.mu.Lock() + defer inner.inf.mu.Unlock() + return len(inner.inf.handlers) > 0 + }) + + c.upsert(reservationGVK(), newReservation("res-x", "az-1", "1")) + inner.inf.fireAdd("not-an-object") + if _, ok := c.getEntry(reservationGVK(), client.ObjectKey{Name: "res-x"}); !ok { + t.Fatalf("non-object event must not evict the entry") + } +} + +func TestTTLCleanup(t *testing.T) { + c := newCaching(t, newTestClient(t)) + c.upsert(reservationGVK(), newReservation("res-4", "az-1", "1")) + c.mu.Lock() + for _, entries := range c.byGVK { + for _, e := range entries { + e.expiresAt = time.Now().Add(-time.Second) + } + } + c.mu.Unlock() + c.cleanupExpired(time.Now()) + if _, ok := c.getEntry(reservationGVK(), client.ObjectKey{Name: "res-4"}); ok { + t.Fatalf("expired entry should be removed") + } +} + +func TestTombstone(t *testing.T) { + r := newReservation("res-5", "az-1", "") + inner := newTestClient(t, r) + c := newCaching(t, inner) + + if err := c.Delete(context.Background(), r); err != nil { + t.Fatalf("Delete: %v", err) + } + // Re-create directly in inner to simulate informer lag. + if err := inner.GetClient().Create(context.Background(), newReservation("res-5", "az-1", "")); err != nil { + t.Fatalf("re-create inner: %v", err) + } + + var got v1alpha1.Reservation + if err := c.Get(context.Background(), types.NamespacedName{Name: "res-5"}, &got); !apierrors.IsNotFound(err) { + t.Fatalf("expected NotFound for tombstoned object, got %v", err) + } + if items := listReservations(t, c); len(items) != 0 { + t.Fatalf("expected tombstone to filter from list, got %+v", items) + } +} + +func TestDeleteAllOf(t *testing.T) { + r1 := newReservation("res-dao-1", "az-1", "1") + r1.Labels = map[string]string{"zone": "a"} + r2 := newReservation("res-dao-2", "az-2", "1") + r2.Labels = map[string]string{"zone": "b"} + inner := newTestClient(t, r1, r2) + c := newCaching(t, inner) + + // Populate overlay for both so we can verify tombstoning. + c.upsert(reservationGVK(), r1) + c.upsert(reservationGVK(), r2) + + // DeleteAllOf with a label selector — only r1 should be tombstoned. + if err := c.DeleteAllOf(context.Background(), &v1alpha1.Reservation{}, client.MatchingLabels{"zone": "a"}); err != nil { + t.Fatalf("DeleteAllOf: %v", err) + } + + var got v1alpha1.Reservation + if err := c.Get(context.Background(), types.NamespacedName{Name: "res-dao-1"}, &got); !apierrors.IsNotFound(err) { + t.Fatalf("expected NotFound for tombstoned res-dao-1, got %v", err) + } + if err := c.Get(context.Background(), types.NamespacedName{Name: "res-dao-2"}, &got); err != nil { + t.Fatalf("res-dao-2 should still be visible, got %v", err) + } +} + +func TestUpdateOverridesInner(t *testing.T) { + r := newReservation("res-6", "az-old", "1") + inner := newTestClient(t, r) + c := newCaching(t, inner) + + var cur v1alpha1.Reservation + if err := inner.GetClient().Get(context.Background(), types.NamespacedName{Name: "res-6"}, &cur); err != nil { + t.Fatalf("inner get: %v", err) + } + cur.Spec.AvailabilityZone = "az-new" + if err := c.Update(context.Background(), &cur); err != nil { + t.Fatalf("Update: %v", err) + } + + var got v1alpha1.Reservation + if err := c.Get(context.Background(), types.NamespacedName{Name: "res-6"}, &got); err != nil { + t.Fatalf("Get: %v", err) + } + if got.Spec.AvailabilityZone != "az-new" { + t.Fatalf("expected az-new from overlay, got %q", got.Spec.AvailabilityZone) + } +} + +func TestLabelMatchingAfterOverlayChange(t *testing.T) { + // Regression: if a write changes a field that is part of the query, the + // overlay version must be re-matched against the list options. Before the + // fix, the inner List would include the object (matching the old value in + // the informer), and overlayList would silently swap in the new version, + // returning an object that does not satisfy the query. + r := newReservation("res-overlay-label", "az-1", "1") + r.Labels = map[string]string{"team": "a"} + inner := newTestClient(t, r) + c := newCaching(t, inner) + + // Update label in overlay only (bypass inner to simulate informer lag). + updated := r.DeepCopy() + updated.Labels = map[string]string{"team": "b"} + c.upsert(reservationGVK(), updated) + + // Query for the OLD label value — informer still returns the object, but + // the overlay version has team=b, so it must be excluded. + if got := listReservations(t, c, client.MatchingLabels{"team": "a"}); len(got) != 0 { + t.Fatalf("expected no results for team=a after overlay changed label to b, got %+v", got) + } + // Query for the NEW label value — overlay-only path must include it. + if got := listReservations(t, c, client.MatchingLabels{"team": "b"}); len(got) != 1 { + t.Fatalf("expected one result for team=b, got %+v", got) + } +} + +func TestLabelMatching(t *testing.T) { + c := newCaching(t, newTestClient(t)) + r := newReservation("res-7", "az-1", "1") + r.Labels = map[string]string{"team": "a"} + c.upsert(reservationGVK(), r) + + if match := listReservations(t, c, client.MatchingLabels{"team": "a"}); len(match) != 1 { + t.Fatalf("expected match for team=a, got %+v", match) + } + if noMatch := listReservations(t, c, client.MatchingLabels{"team": "b"}); len(noMatch) != 0 { + t.Fatalf("expected no match for team=b, got %+v", noMatch) + } +} + +func TestFieldMatching(t *testing.T) { + c := newCaching(t, newTestClient(t)) + if err := c.IndexField(context.Background(), &v1alpha1.Reservation{}, azIndexField, func(obj client.Object) []string { + res := obj.(*v1alpha1.Reservation) + if res.Spec.AvailabilityZone == "" { + return nil + } + return []string{res.Spec.AvailabilityZone} + }); err != nil { + t.Fatalf("IndexField: %v", err) + } + c.upsert(reservationGVK(), newReservation("res-8", "az-1", "1")) + + if match := listReservations(t, c, client.MatchingFields{azIndexField: "az-1"}); len(match) != 1 { + t.Fatalf("expected field match az-1, got %+v", match) + } + if noMatch := listReservations(t, c, client.MatchingFields{azIndexField: "az-2"}); len(noMatch) != 0 { + t.Fatalf("expected no field match az-2, got %+v", noMatch) + } +} + +func TestNonCachedGVKPassthrough(t *testing.T) { + inner := newTestClient(t) + c := cachingFrom(t, inner.fakeCluster, Config{}) + r := newReservation("res-9", "az-1", "") + if err := c.Create(context.Background(), r); err != nil { + t.Fatalf("Create: %v", err) + } + if _, ok := c.getEntry(reservationGVK(), client.ObjectKey{Name: "res-9"}); ok { + t.Fatalf("non-cached GVK should not populate overlay") + } + if err := c.Delete(context.Background(), r); err != nil { + t.Fatalf("Delete: %v", err) + } + var got v1alpha1.Reservation + if err := c.Get(context.Background(), types.NamespacedName{Name: "res-9"}, &got); !apierrors.IsNotFound(err) { + t.Fatalf("expected NotFound, got %v", err) + } +} + +func TestDedup(t *testing.T) { + r := newReservation("res-10", "az-1", "1") + c := newCaching(t, newTestClient(t, r)) + + newer := newReservation("res-10", "az-1", "2") + newer.Spec.TargetHost = "host-x" + c.upsert(reservationGVK(), newer) + + items := listReservations(t, c) + if len(items) != 1 { + t.Fatalf("expected exactly one item after dedup, got %d: %+v", len(items), items) + } + if items[0].Spec.TargetHost != "host-x" { + t.Fatalf("expected overlay version to win, got %+v", items[0]) + } +} + +// --- error-propagation tests --- + +func TestWriteErrorLeavesOverlayUntouched(t *testing.T) { + sentinel := errors.New("boom") + cases := []struct { + name string + rv string + seed bool + mkClient func(inner *fakeClient) client.Client + op func(c *Overlay, r *v1alpha1.Reservation) error + }{ + { + name: "create", + mkClient: func(inner *fakeClient) client.Client { + return &errClient{Client: inner.GetClient(), createErr: sentinel} + }, + op: func(c *Overlay, r *v1alpha1.Reservation) error { return c.Create(context.Background(), r) }, + }, + { + name: "update", + rv: "1", + mkClient: func(inner *fakeClient) client.Client { + return &errClient{Client: inner.GetClient(), updateErr: sentinel} + }, + op: func(c *Overlay, r *v1alpha1.Reservation) error { return c.Update(context.Background(), r) }, + }, + { + name: "patch", + rv: "1", + mkClient: func(inner *fakeClient) client.Client { + return &errClient{Client: inner.GetClient(), patchErr: sentinel} + }, + op: func(c *Overlay, r *v1alpha1.Reservation) error { + p := r.DeepCopy() + p.Spec.AvailabilityZone = "az-new" + return c.Patch(context.Background(), p, client.MergeFrom(r)) + }, + }, + { + name: "delete", + seed: true, + mkClient: func(inner *fakeClient) client.Client { + return &errClient{Client: inner.GetClient(), deleteErr: sentinel} + }, + op: func(c *Overlay, r *v1alpha1.Reservation) error { return c.Delete(context.Background(), r) }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + r := newReservation("res-"+tc.name, "az-1", tc.rv) + var base *fakeClient + if tc.seed { + base = newTestClient(t, r) + } else { + base = newTestClient(t) + } + c := cachingFrom(t, clusterFor(t, tc.mkClient(base)), reservationConfig()) + + if err := tc.op(c, r); !errors.Is(err, sentinel) { + t.Fatalf("expected sentinel error, got %v", err) + } + if _, ok := c.getEntry(reservationGVK(), client.ObjectKey{Name: r.Name}); ok { + t.Fatalf("overlay must not be touched on %s failure", tc.name) + } + }) + } +} + +func TestWriteServedFromOverlay(t *testing.T) { + cases := []struct { + name string + write func(t *testing.T, c *Overlay, cur *v1alpha1.Reservation) string + diverge func(t *testing.T, inner client.Client, name string) + read func(*v1alpha1.Reservation) string + }{ + { + name: "patch spec", + write: func(t *testing.T, c *Overlay, cur *v1alpha1.Reservation) string { + base := cur.DeepCopy() + cur.Spec.AvailabilityZone = "az-new" + if err := c.Patch(context.Background(), cur, client.MergeFrom(base)); err != nil { + t.Fatalf("Patch: %v", err) + } + return "az-new" + }, + diverge: func(t *testing.T, inner client.Client, name string) { forceInnerAZ(t, inner, name, "az-stale") }, + read: func(r *v1alpha1.Reservation) string { return r.Spec.AvailabilityZone }, + }, + { + name: "status update", + write: func(t *testing.T, c *Overlay, cur *v1alpha1.Reservation) string { + cur.Status.Host = "host-active" + if err := c.Status().Update(context.Background(), cur); err != nil { + t.Fatalf("Status().Update: %v", err) + } + return "host-active" + }, + diverge: func(t *testing.T, inner client.Client, name string) { + forceInnerStatusHost(t, inner, name, "host-stale") + }, + read: func(r *v1alpha1.Reservation) string { return r.Status.Host }, + }, + { + name: "status patch", + write: func(t *testing.T, c *Overlay, cur *v1alpha1.Reservation) string { + base := cur.DeepCopy() + cur.Status.Host = "host-patched" + if err := c.Status().Patch(context.Background(), cur, client.MergeFrom(base)); err != nil { + t.Fatalf("Status().Patch: %v", err) + } + return "host-patched" + }, + diverge: func(t *testing.T, inner client.Client, name string) { + forceInnerStatusHost(t, inner, name, "host-stale") + }, + read: func(r *v1alpha1.Reservation) string { return r.Status.Host }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + r := newReservation("res-served", "az-1", "") + inner := newTestClient(t, r) + c := newCaching(t, inner) + + var cur v1alpha1.Reservation + if err := inner.GetClient().Get(context.Background(), types.NamespacedName{Name: r.Name}, &cur); err != nil { + t.Fatalf("inner get: %v", err) + } + want := tc.write(t, c, &cur) + tc.diverge(t, inner.GetClient(), r.Name) + + var got v1alpha1.Reservation + if err := c.Get(context.Background(), types.NamespacedName{Name: r.Name}, &got); err != nil { + t.Fatalf("Get: %v", err) + } + if tc.read(&got) != want { + t.Fatalf("expected overlay value %q to be served, got %q", want, tc.read(&got)) + } + }) + } +} + +func TestGetPropagatesNonNotFoundError(t *testing.T) { + sentinel := errors.New("get boom") + c := cachingFrom(t, clusterFor(t, &errClient{Client: newTestClient(t).GetClient(), getErr: sentinel}), reservationConfig()) + c.upsert(reservationGVK(), newReservation("res-ge", "az-1", "1")) + + var got v1alpha1.Reservation + if err := c.Get(context.Background(), types.NamespacedName{Name: "res-ge"}, &got); !errors.Is(err, sentinel) { + t.Fatalf("expected sentinel error, got %v", err) + } +} + +func TestGetOverlayResurrectsNotFound(t *testing.T) { + c := newCaching(t, newTestClient(t)) + c.upsert(reservationGVK(), newReservation("res-gr", "az-z", "1")) + + var got v1alpha1.Reservation + if err := c.Get(context.Background(), types.NamespacedName{Name: "res-gr"}, &got); err != nil { + t.Fatalf("expected overlay to satisfy Get, got %v", err) + } + if got.Spec.AvailabilityZone != "az-z" { + t.Fatalf("expected az-z from overlay, got %q", got.Spec.AvailabilityZone) + } +} + +func TestGetNotFoundWithNoOverlay(t *testing.T) { + c := newCaching(t, newTestClient(t)) + + var got v1alpha1.Reservation + if err := c.Get(context.Background(), types.NamespacedName{Name: "missing"}, &got); !apierrors.IsNotFound(err) { + t.Fatalf("expected NotFound, got %v", err) + } +} + +func TestGetNonCachedPropagatesError(t *testing.T) { + sentinel := errors.New("get boom") + c := cachingFrom(t, clusterFor(t, &errClient{Client: newTestClient(t).GetClient(), getErr: sentinel}), Config{}) + var got v1alpha1.Reservation + if gerr := c.Get(context.Background(), types.NamespacedName{Name: "x"}, &got); !errors.Is(gerr, sentinel) { + t.Fatalf("expected sentinel error, got %v", gerr) + } +} + +func TestListPropagatesError(t *testing.T) { + sentinel := errors.New("list boom") + c := cachingFrom(t, clusterFor(t, &errClient{Client: newTestClient(t).GetClient(), listErr: sentinel}), reservationConfig()) + c.upsert(reservationGVK(), newReservation("res-le", "az-1", "1")) + + var list v1alpha1.ReservationList + if err := c.List(context.Background(), &list); !errors.Is(err, sentinel) { + t.Fatalf("expected sentinel error, got %v", err) + } +} + +func TestStatusUpdateErrorLeavesOverlayUntouched(t *testing.T) { + c := newCaching(t, newTestClient(t)) + + r := newReservation("res-se", "az-1", "1") + if err := c.Status().Update(context.Background(), r); err == nil { + t.Fatalf("expected status update to fail for missing object") + } + if _, ok := c.getEntry(reservationGVK(), client.ObjectKey{Name: "res-se"}); ok { + t.Fatalf("overlay must not be populated on status update failure") + } +} + +func TestStatusCreateDelegates(t *testing.T) { + r := newReservation("res-sc", "az-1", "") + c := newCaching(t, newTestClient(t, r)) + + if err := c.Status().Create(context.Background(), r, r); err == nil { + t.Fatalf("expected Status().Create to fail on fake client") + } + if _, ok := c.getEntry(reservationGVK(), client.ObjectKey{Name: "res-sc"}); ok { + t.Fatalf("Status().Create must not populate the overlay") + } +} + +func TestStatusUpdateNonCachedNoOverlay(t *testing.T) { + r := newReservation("res-sn", "az-1", "") + inner := newTestClient(t, r) + c := cachingFrom(t, inner.fakeCluster, Config{}) + var cur v1alpha1.Reservation + if err := inner.GetClient().Get(context.Background(), types.NamespacedName{Name: "res-sn"}, &cur); err != nil { + t.Fatalf("inner get: %v", err) + } + cur.Status.Host = "host-active" + if err := c.Status().Update(context.Background(), &cur); err != nil { + t.Fatalf("Status().Update: %v", err) + } + if _, ok := c.getEntry(reservationGVK(), client.ObjectKey{Name: "res-sn"}); ok { + t.Fatalf("non-cached GVK status update should not populate overlay") + } +} + +// --- concurrency regression test --- + +// orderingClient is a fake inner client whose Update records the committed +// ResourceVersion (in inner-commit order) and then yields, widening the window +// between the inner commit and the overlay upsert. This exposes the ordering +// race the per-object write lock is meant to prevent: without the lock two +// concurrent writers to the same object can commit in one order yet upsert in +// the reverse order, leaving the overlay behind the apiserver. +// +// The per-object write lock makes each inner call + upsert atomic, so the inner +// commit order and the overlay upsert order are identical: the overlay always +// reflects the last write that reached the inner client (lastRV). +type orderingClient struct { + client.Client + mu sync.Mutex + lastRV string // ResourceVersion of the most recent inner commit +} + +func (o *orderingClient) Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error { + o.mu.Lock() + o.lastRV = obj.GetResourceVersion() + o.mu.Unlock() + // Yield after the inner commit but before the caller upserts, widening the + // commit-vs-overlay reorder window. + goruntime.Gosched() + return nil +} + +// TestConcurrentUpdatesOverlayNotBehind fires many concurrent Updates to the +// SAME object with distinct ResourceVersions. The per-object write lock must +// make each inner call + overlay update atomic, so once all writes settle the +// overlay entry reflects the last write that reached the inner client (overlay +// RV == last committed RV, never a reordered/stale one). Run under -race to +// also catch data races. +func TestConcurrentUpdatesOverlayNotBehind(t *testing.T) { + const ( + rounds = 50 + n = 8 + ) + for round := range rounds { + oc := &orderingClient{Client: newTestClient(t).GetClient()} + c := cachingFrom(t, clusterFor(t, oc), reservationConfig()) + + var wg sync.WaitGroup + for i := 1; i <= n; i++ { + wg.Add(1) + go func(rv int) { + defer wg.Done() + r := newReservation("res-conc", "az-1", strconv.Itoa(rv)) + if err := c.Update(context.Background(), r); err != nil { + t.Errorf("Update rv=%d: %v", rv, err) + } + }(i) + } + wg.Wait() + + oc.mu.Lock() + lastRV := oc.lastRV + oc.mu.Unlock() + + e, ok := c.getEntry(reservationGVK(), client.ObjectKey{Name: "res-conc"}) + if !ok { + t.Fatalf("round %d: expected overlay entry for res-conc", round) + } + if e.resourceVersion != lastRV { + t.Fatalf("round %d: overlay RV %q does not match last inner commit %q", + round, e.resourceVersion, lastRV) + } + } +} + +// --- helper / utility tests --- + +func TestGVKForUnknownType(t *testing.T) { + c := newCaching(t, newTestClient(t)) + if _, cached := c.gvkFor(&unknownObject{}); cached { + t.Fatalf("unknown type must not be reported as cached") + } +} + +func TestTrimListSuffix(t *testing.T) { + cases := []struct { + in string + wantKind string + wantOK bool + }{ + {"ReservationList", "Reservation", true}, + {"List", "List", false}, + {"Reservation", "Reservation", false}, + {"", "", false}, + } + for _, tc := range cases { + gotKind, gotOK := trimListSuffix(tc.in) + if gotKind != tc.wantKind || gotOK != tc.wantOK { + t.Errorf("trimListSuffix(%q) = (%q, %v), want (%q, %v)", tc.in, gotKind, gotOK, tc.wantKind, tc.wantOK) + } + } +} diff --git a/pkg/cache/cluster.go b/pkg/cache/cluster.go new file mode 100644 index 000000000..0fc51cdcd --- /dev/null +++ b/pkg/cache/cluster.go @@ -0,0 +1,55 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package cache + +import ( + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/cluster" + "sigs.k8s.io/controller-runtime/pkg/manager" +) + +// overlayCluster wraps a cluster.Cluster so its client and field indexer are +// served through the in-process overlay cache. Everything else (config, scheme, +// REST mapper, HTTP client, API reader, event recorder, and Start) is delegated +// to the embedded inner cluster unchanged. +// +// It deliberately does NOT override Start: the manager already owns the inner +// cluster's lifecycle (home cluster) or drives it via mgr.Add (remotes). The +// overlay's own lifecycle (eviction handlers + TTL cleanup) is registered with +// the manager directly by Wrapper.WrapCluster. +type overlayCluster struct { + cluster.Cluster + cache *Overlay +} + +// GetClient returns the overlay client so per-cluster reads and writes flow +// through the cache. +func (w *overlayCluster) GetClient() client.Client { return w.cache } + +// GetFieldIndexer returns the overlay so IndexField dual-registers the indexer +// with both the real cache (via Overlay.IndexField) and the overlay. +func (w *overlayCluster) GetFieldIndexer() client.FieldIndexer { return w.cache } + +// Wrapper implements the multicluster.ClusterWrapper interface for the overlay +// cache. It does not import the multicluster package; interface satisfaction is +// structural and checked at the call site (e.g. in main.go). +type Wrapper struct{ conf Config } + +// NewWrapper returns a Wrapper that applies the overlay cache to any cluster +// passed to WrapCluster. +func NewWrapper(conf Config) *Wrapper { return &Wrapper{conf} } + +// WrapCluster applies the overlay cache to cl, registers the overlay's lifecycle +// Runnable with mgr, and returns the wrapped cluster. +// Satisfies multicluster.ClusterWrapper structurally. +func (w *Wrapper) WrapCluster(mgr manager.Manager, cl cluster.Cluster) (cluster.Cluster, error) { + wrapped, cleanUpRunnable, err := WrapCluster(cl, w.conf) + if err != nil { + return nil, err + } + if err := mgr.Add(cleanUpRunnable); err != nil { + return nil, err + } + return wrapped, nil +} diff --git a/pkg/cache/config.go b/pkg/cache/config.go new file mode 100644 index 000000000..08c58d573 --- /dev/null +++ b/pkg/cache/config.go @@ -0,0 +1,27 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package cache + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// Config configures the transparent in-process overlay cache. +type Config struct { + // Enabled turns the overlay cache on. When false (the default), clusters are + // used unwrapped: no overlay, no eviction Runnable, zero overhead. + Enabled bool `json:"enabled"` + // GVKs the cache should overlay, formatted as "//". + // Calls for GVKs not listed here are passed through unchanged. + GVKs []string `json:"gvks"` + // TTL is the maximum lifetime of an overlay entry before it is evicted by + // the background cleanup goroutine, guarding against entries that never + // appear in the informer (e.g. after a crash). Defaults to 2m when zero. + TTL metav1.Duration `json:"ttl,omitzero"` +} + +type RootConfig struct { + // Cache configures the transparent in-process overlay cache. + Cache Config `json:"cache"` +} diff --git a/pkg/cache/lock.go b/pkg/cache/lock.go new file mode 100644 index 000000000..22d2a351a --- /dev/null +++ b/pkg/cache/lock.go @@ -0,0 +1,85 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package cache + +import ( + "sync" + + "k8s.io/apimachinery/pkg/runtime/schema" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// lockKey identifies the object whose write path a keyedMutex serializes. +type lockKey struct { + gvk schema.GroupVersionKind + key client.ObjectKey +} + +// refMutex is the actual per-object lock plus a reference count of how many +// callers currently hold or are waiting for it. The count lets keyedMutex know +// when the entry is unused so it can be deleted (see keyedMutex.lock). +type refMutex struct { + mu sync.Mutex + ref int +} + +// keyedMutex hands out one mutex per key, serializing operations that share a +// key while letting distinct keys proceed concurrently. +// +// A sync.Map (or a map we only ever insert into) would be simpler, but it would leak: +// this cache wraps a single client for the whole lifetime of the controller-manager process, +// and a controller reconciles a continuous stream of (often short-lived) objects. +// A map would therefore accumulate one mutex per distinct object ever written and +// grow without bound. To avoid that we reference count each entry and delete it +// once the last holder unlocks, so the map only ever holds locks for objects +// with writes currently in flight. +// +// The two mutexes have distinct, non-overlapping roles: +// - k.mu guards the locks map itself. It is only ever held for the tiny +// bookkeeping critical sections below (map lookup/insert/delete and the +// ref counter), never across the caller's I/O. +// - rm.mu is the real per-object lock, held by the caller across the inner +// client call and the overlay mutation. +// +// Because k.mu is never held while rm.mu is locked (we release k.mu before +// taking rm.mu), the two can never deadlock against each other. +type keyedMutex struct { + mu sync.Mutex + locks map[lockKey]*refMutex +} + +func newKeyedMutex() *keyedMutex { + return &keyedMutex{locks: make(map[lockKey]*refMutex)} +} + +// lock acquires the per-key mutex and returns a function that releases it. +func (k *keyedMutex) lock(lk lockKey) func() { + // Look up (or create) the entry for this key and register our interest by + // bumping ref, all under k.mu so the map stays consistent. We increment ref + // here, before taking rm.mu, so that a concurrent unlock cannot see ref==0 + // and delete the entry out from under us while we are blocked waiting on it. + k.mu.Lock() + rm := k.locks[lk] + if rm == nil { + rm = &refMutex{} + k.locks[lk] = rm + } + rm.ref++ + k.mu.Unlock() + + // Take the real per-object lock outside k.mu; this is where a second caller + // for the same key blocks (and where we may block across the caller's I/O). + rm.mu.Lock() + return func() { + rm.mu.Unlock() + // Drop our reference and, if we were the last holder, remove the entry + // so the map does not grow unbounded. + k.mu.Lock() + rm.ref-- + if rm.ref == 0 { + delete(k.locks, lk) + } + k.mu.Unlock() + } +} diff --git a/pkg/cache/runnable.go b/pkg/cache/runnable.go new file mode 100644 index 000000000..a4d8a9a21 --- /dev/null +++ b/pkg/cache/runnable.go @@ -0,0 +1,96 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package cache + +import ( + "context" + "fmt" + "time" + + "k8s.io/apimachinery/pkg/runtime/schema" + toolscachek8s "k8s.io/client-go/tools/cache" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// minCleanupInterval bounds the TTL cleanup ticker from below. +const minCleanupInterval = 30 * time.Second + +// Start implements manager.Runnable. It attaches informer event handlers for +// eviction and runs the TTL cleanup loop until ctx is done. +func (c *Overlay) Start(ctx context.Context) error { + log := ctrl.LoggerFrom(ctx).WithName("cache") + + for gvk := range c.gvks { + obj, err := c.newObjectForGVK(gvk) + if err != nil { + log.Error(err, "failed to build object for gvk; eviction disabled for it", "gvk", gvk) + continue + } + inf, err := c.informerCache.GetInformer(ctx, obj) + if err != nil { + log.Error(err, "failed to get informer for gvk; eviction disabled for it", "gvk", gvk) + continue + } + if _, err := inf.AddEventHandler(c.evictionHandler(gvk)); err != nil { + log.Error(err, "failed to add eviction event handler", "gvk", gvk) + } + } + + interval := max(c.ttl/4, minCleanupInterval) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return nil + case now := <-ticker.C: + c.cleanupExpired(now) + } + } +} + +// NeedLeaderElection reports that the Overlay's Start lifecycle must run +// on every replica, not only the elected leader. The overlay is per-process +// state, so its eviction handlers and TTL cleanup have to run wherever the +// client is used, regardless of leader election. +func (c *Overlay) NeedLeaderElection() bool { + return false +} + +// evictionHandler returns an informer event handler that evicts overlay entries +// for the GVK when the real object is observed at a >= ResourceVersion. +func (c *Overlay) evictionHandler(gvk schema.GroupVersionKind) toolscachek8s.ResourceEventHandler { + evict := func(o any) { + obj, ok := o.(client.Object) + if !ok { + return + } + c.evictIfSeen(gvk, obj) + } + return toolscachek8s.ResourceEventHandlerFuncs{ + AddFunc: func(o any) { evict(o) }, + UpdateFunc: func(_, o any) { evict(o) }, + DeleteFunc: func(o any) { + if d, ok := o.(toolscachek8s.DeletedFinalStateUnknown); ok { + evict(d.Obj) + return + } + evict(o) + }, + } +} + +// newObjectForGVK builds an empty typed object for the GVK using the scheme. +func (c *Overlay) newObjectForGVK(gvk schema.GroupVersionKind) (client.Object, error) { + ro, err := c.scheme.New(gvk) + if err != nil { + return nil, err + } + obj, ok := ro.(client.Object) + if !ok { + return nil, fmt.Errorf("cache: object for gvk %s does not implement client.Object", gvk) + } + return obj, nil +} diff --git a/pkg/multicluster/client.go b/pkg/multicluster/client.go index 8b3a65a60..cb8d1c0b5 100644 --- a/pkg/multicluster/client.go +++ b/pkg/multicluster/client.go @@ -20,8 +20,22 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/cluster" + "sigs.k8s.io/controller-runtime/pkg/manager" ) +// ClusterWrapper can be registered on the Client to transparently transform each +// cluster before it is stored for routing. WrapCluster receives the manager and +// the current cluster: raw for the first wrapper in the chain, or the previous +// wrapper's result for subsequent ones. It must return the (possibly wrapped) +// cluster to use for routing and is responsible for registering any lifecycle +// Runnables with mgr itself. Wrappers are applied in order for every home and +// remote cluster during InitFromConf. For remote clusters the original unwrapped +// cluster is always added to the manager separately so its informers start +// independently of wrapping. +type ClusterWrapper interface { + WrapCluster(mgr manager.Manager, cl cluster.Cluster) (cluster.Cluster, error) +} + // A remote cluster with routing labels used to match resources to clusters. type remoteCluster struct { cluster cluster.Cluster @@ -33,6 +47,12 @@ type Client struct { // when multiple clusters serve the same GVK. ResourceRouters map[schema.GroupVersionKind]ResourceRouter + // Wrappers are applied to every cluster (home and remotes) during InitFromConf. + // Each wrapper transforms the cluster before it is stored for routing. + // Applied in slice order; the raw inner cluster is always added to the manager + // so its informers start independently of wrapping. + Wrappers []ClusterWrapper + // The cluster in which cortex is deployed. HomeCluster cluster.Cluster // The REST config for the home cluster in which cortex is deployed. @@ -55,48 +75,6 @@ type Client struct { homeGVKs map[schema.GroupVersionKind]bool } -type ClientConfig struct { - // Apiserver configuration mapping GVKs to home or remote clusters. - // Every GVK used through the multicluster client must be listed - // in either Home or Remotes. Unknown GVKs will cause an error. - APIServers APIServersConfig `json:"apiservers"` -} - -// APIServersConfig separates resources into home and remote clusters. -type APIServersConfig struct { - // Resources managed in the cluster where cortex is deployed. - Home HomeConfig `json:"home"` - // Resources managed in remote clusters. - Remotes []RemoteConfig `json:"remotes,omitempty"` -} - -// HomeConfig lists GVKs that are managed in the home cluster. -type HomeConfig struct { - // The resource GVKs formatted as "//". - GVKs []string `json:"gvks"` -} - -// RemoteConfig maps multiple GVKs to a remote kubernetes apiserver with -// routing labels. It is assumed that the remote apiserver accepts the -// serviceaccount tokens issued by the local cluster. -type RemoteConfig struct { - // The remote kubernetes apiserver url, e.g. "https://my-apiserver:6443". - Host string `json:"host"` - // The root CA certificate to verify the remote apiserver. - // Ignored if InsecureSkipTLSVerify is true. - CACert string `json:"caCert,omitempty"` - // InsecureSkipTLSVerify disables verification of the remote apiserver's - // TLS certificate. Use this for apiservers whose CA certificate rotates - // frequently and does not chain to a stable root. Mutually exclusive - // with CACert: when true, CACert is ignored. - InsecureSkipTLSVerify bool `json:"insecureSkipTLSVerify,omitempty"` - // The resource GVKs this apiserver serves, formatted as "//". - GVKs []string `json:"gvks"` - // Labels used by ResourceRouters to match resources to this cluster - // for write operations (Create/Update/Delete/Patch). - Labels map[string]string `json:"labels,omitempty"` -} - // Helper function to initialize a new multicluster client during service startup, // using the conf module provided by cortex. func (c *Client) InitFromConf(ctx context.Context, mgr ctrl.Manager, conf ClientConfig) error { @@ -132,14 +110,27 @@ func (c *Client) InitFromConf(ctx context.Context, mgr ctrl.Manager, conf Client } resolvedGVKs = append(resolvedGVKs, gvk) } - cl, err := c.AddRemote(ctx, remote.Host, remote.CACert, remote.InsecureSkipTLSVerify, remote.Labels, resolvedGVKs...) + cl, err := c.AddRemote(ctx, mgr, remote.Host, remote.CACert, remote.InsecureSkipTLSVerify, remote.Labels, resolvedGVKs...) if err != nil { return err } + // Add the raw inner cluster so its informers/caches start. if err := mgr.Add(cl); err != nil { return err } } + // Apply wrappers to the home cluster. The manager already owns the home + // cluster's lifecycle, so we only apply wrappers (each wrapper registers its + // own Runnables) and must NOT re-Start the inner home cluster. + if c.HomeCluster != nil { + for _, w := range c.Wrappers { + wrapped, werr := w.WrapCluster(mgr, c.HomeCluster) + if werr != nil { + return werr + } + c.HomeCluster = wrapped + } + } return nil } @@ -153,7 +144,12 @@ func (c *Client) InitFromConf(ctx context.Context, mgr ctrl.Manager, conf Client // This can be used when the remote cluster accepts the home cluster's service // account tokens. See the kubernetes documentation on structured auth to // learn more about jwt-based authentication across clusters. -func (c *Client) AddRemote(ctx context.Context, host, caCert string, insecureSkipTLSVerify bool, labels map[string]string, gvks ...schema.GroupVersionKind) (cluster.Cluster, error) { +// AddRemote returns the raw inner cluster.Cluster (which the caller must add to +// the manager so its informers/caches start). Each registered Wrapper is +// responsible for adding its own lifecycle Runnables to mgr directly. +// The wrapped cluster is stored in remoteClusters so all routing goes through +// any per-cluster wrapper. +func (c *Client) AddRemote(ctx context.Context, mgr manager.Manager, host, caCert string, insecureSkipTLSVerify bool, labels map[string]string, gvks ...schema.GroupVersionKind) (cluster.Cluster, error) { log := ctrl.LoggerFrom(ctx) homeRestConfig := *c.HomeRestConfig restConfigCopy := homeRestConfig @@ -173,6 +169,17 @@ func (c *Client) AddRemote(ctx context.Context, host, caCert string, insecureSki if err != nil { return nil, err } + // Apply each registered wrapper in order. stored is the cluster placed in + // remoteClusters. Each wrapper is responsible for registering its own + // lifecycle Runnables with mgr directly. + stored := cl + for _, w := range c.Wrappers { + wrapped, werr := w.WrapCluster(mgr, stored) + if werr != nil { + return nil, werr + } + stored = wrapped + } c.remoteClustersMu.Lock() defer c.remoteClustersMu.Unlock() if c.remoteClusters == nil { @@ -181,10 +188,11 @@ func (c *Client) AddRemote(ctx context.Context, host, caCert string, insecureSki for _, gvk := range gvks { log.Info("adding remote cluster for resource", "gvk", gvk, "host", host, "labels", labels, "insecureSkipTLSVerify", insecureSkipTLSVerify) c.remoteClusters[gvk] = append(c.remoteClusters[gvk], remoteCluster{ - cluster: cl, + cluster: stored, labels: labels, }) } + // Return the raw inner cluster so the caller starts its informers/caches. return cl, nil } @@ -854,7 +862,7 @@ func (c *Client) IndexField(ctx context.Context, obj client.Object, list client. continue } indexed[ch] = true - if err := ch.IndexField(ctx, obj, field, extractValue); err != nil { + if err := cl.GetFieldIndexer().IndexField(ctx, obj, field, extractValue); err != nil { log.Error(err, "failed to register field index for cluster — objects from this cluster will be absent from index queries; restart required to recover", "field", field) continue } @@ -869,7 +877,7 @@ func (c *Client) IndexField(ctx context.Context, obj client.Object, list client. continue } indexed[ch] = true - if err := ch.IndexField(ctx, obj, field, extractValue); err != nil { + if err := cl.GetFieldIndexer().IndexField(ctx, obj, field, extractValue); err != nil { log.Error(err, "failed to register field index for cluster — objects from this cluster will be absent from index queries; restart required to recover", "field", field) continue } diff --git a/pkg/multicluster/client_test.go b/pkg/multicluster/client_test.go index 41fd3a6b9..29ea8e0ca 100644 --- a/pkg/multicluster/client_test.go +++ b/pkg/multicluster/client_test.go @@ -82,16 +82,28 @@ type fakeCluster struct { fakeClient client.Client fakeCache *fakeCache fakeRecorder events.EventRecorder + scheme *runtime.Scheme } func (f *fakeCluster) GetClient() client.Client { return f.fakeClient } +func (f *fakeCluster) GetScheme() *runtime.Scheme { + return f.scheme +} + func (f *fakeCluster) GetCache() cache.Cache { return f.fakeCache } +// GetFieldIndexer returns the fake cache, which implements client.FieldIndexer +// via its IndexField method. This mirrors production, where an unwrapped +// cluster's field indexer is its cache. +func (f *fakeCluster) GetFieldIndexer() client.FieldIndexer { + return f.fakeCache +} + func (f *fakeCluster) GetEventRecorder(_ string) events.EventRecorder { if f.fakeRecorder != nil { return f.fakeRecorder @@ -107,6 +119,7 @@ func newFakeCluster(scheme *runtime.Scheme, objs ...client.Object) *fakeCluster return &fakeCluster{ fakeClient: fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build(), fakeCache: &fakeCache{}, + scheme: scheme, } } @@ -115,6 +128,7 @@ func newFakeClusterWithCache(scheme *runtime.Scheme, fakeCache *fakeCache, objs return &fakeCluster{ fakeClient: fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build(), fakeCache: fakeCache, + scheme: scheme, } } diff --git a/pkg/multicluster/config.go b/pkg/multicluster/config.go new file mode 100644 index 000000000..359d225cb --- /dev/null +++ b/pkg/multicluster/config.go @@ -0,0 +1,46 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package multicluster + +type ClientConfig struct { + // Apiserver configuration mapping GVKs to home or remote clusters. + // Every GVK used through the multicluster client must be listed + // in either Home or Remotes. Unknown GVKs will cause an error. + APIServers APIServersConfig `json:"apiservers"` +} + +// APIServersConfig separates resources into home and remote clusters. +type APIServersConfig struct { + // Resources managed in the cluster where cortex is deployed. + Home HomeConfig `json:"home"` + // Resources managed in remote clusters. + Remotes []RemoteConfig `json:"remotes,omitempty"` +} + +// HomeConfig lists GVKs that are managed in the home cluster. +type HomeConfig struct { + // The resource GVKs formatted as "//". + GVKs []string `json:"gvks"` +} + +// RemoteConfig maps multiple GVKs to a remote kubernetes apiserver with +// routing labels. It is assumed that the remote apiserver accepts the +// serviceaccount tokens issued by the local cluster. +type RemoteConfig struct { + // The remote kubernetes apiserver url, e.g. "https://my-apiserver:6443". + Host string `json:"host"` + // The root CA certificate to verify the remote apiserver. + // Ignored if InsecureSkipTLSVerify is true. + CACert string `json:"caCert,omitempty"` + // InsecureSkipTLSVerify disables verification of the remote apiserver's + // TLS certificate. Use this for apiservers whose CA certificate rotates + // frequently and does not chain to a stable root. Mutually exclusive + // with CACert: when true, CACert is ignored. + InsecureSkipTLSVerify bool `json:"insecureSkipTLSVerify,omitempty"` + // The resource GVKs this apiserver serves, formatted as "//". + GVKs []string `json:"gvks"` + // Labels used by ResourceRouters to match resources to this cluster + // for write operations (Create/Update/Delete/Patch). + Labels map[string]string `json:"labels,omitempty"` +}