diff --git a/cmd/manager/main.go b/cmd/manager/main.go index fca0c0550..7e09976b2 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/clientcache" "github.com/cobaltcore-dev/cortex/pkg/conf" "github.com/cobaltcore-dev/cortex/pkg/monitoring" "github.com/cobaltcore-dev/cortex/pkg/multicluster" @@ -396,6 +397,22 @@ func main() { os.Exit(1) } + // Transparent in-process overlay cache for CRDs that are eventually + // consistent across in-pod clients (e.g. Reservations). Writes populate an + // overlay; reads merge it with the informer result until the real object is + // observed. *multicluster.Client satisfies clientcache.Client, providing + // both the inner client.Client and informer access for eviction. + clientCacheConfig := conf.GetConfigOrDie[clientcache.RootConfig]() + cachingClient, err := clientcache.New(multiclusterClient, scheme, clientCacheConfig.ClientCache) + if err != nil { + setupLog.Error(err, "unable to create client cache") + os.Exit(1) + } + if err := mgr.Add(cachingClient); err != nil { + setupLog.Error(err, "unable to add client cache to manager") + os.Exit(1) + } + // Our custom monitoring registry can add prometheus labels to all metrics. // This is useful to distinguish metrics from different deployments. metricsConfig := conf.GetConfigOrDie[monitoring.Config]() @@ -428,10 +445,10 @@ func main() { commitmentsConfig := conf.GetConfigOrDie[commitments.Config]() var commitmentsVMSource reservations.VMSource if commitmentsConfig.DatasourceName != "" { - commitmentsVMSource = reservations.NewPostgresVMSource(multiclusterClient, commitmentsConfig.DatasourceName) + commitmentsVMSource = reservations.NewPostgresVMSource(cachingClient, commitmentsConfig.DatasourceName) } if slices.Contains(mainConfig.EnabledControllers, "committed-resource-reservations-controller") { - commitmentsAPI := commitmentsapi.NewAPIWithConfig(multiclusterClient, commitmentsConfig.API, commitmentsVMSource) + commitmentsAPI := commitmentsapi.NewAPIWithConfig(cachingClient, commitmentsConfig.API, commitmentsVMSource) commitmentsAPI.Init(mux, metrics.Registry, ctrl.Log.WithName("commitments-api")) } @@ -451,8 +468,8 @@ func main() { metrics.Registry.MustRegister(noHostFoundCounter) metrics.Registry.MustRegister(placementCounter) // Inferred through the base controller. - filterWeigherController.Client = multiclusterClient - filterWeigherController.CRRecorder.Client = multiclusterClient + filterWeigherController.Client = cachingClient + filterWeigherController.CRRecorder.Client = cachingClient if err := filterWeigherController.SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "nova FilterWeigherPipelineController") os.Exit(1) @@ -467,7 +484,7 @@ func main() { novaClient := nova.NewNovaClient() novaClientConfig := conf.GetConfigOrDie[nova.NovaClientConfig]() if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error { - return novaClient.Init(ctx, multiclusterClient, novaClientConfig) + return novaClient.Init(ctx, cachingClient, novaClientConfig) })); err != nil { setupLog.Error(err, "unable to initialize nova client") os.Exit(1) @@ -478,7 +495,7 @@ func main() { Breaker: &nova.DetectorCycleBreaker{NovaClient: novaClient}, } // Inferred through the base controller. - deschedulingsController.Client = multiclusterClient + deschedulingsController.Client = cachingClient if err := (deschedulingsController).SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "nova DetectorPipelineController") os.Exit(1) @@ -486,7 +503,7 @@ func main() { go deschedulingsController.CreateDeschedulingsPeriodically(ctx) // Deschedulings cleanup on startup if err := (&nova.DeschedulingsCleanup{ - Client: multiclusterClient, + Client: cachingClient, Scheme: mgr.GetScheme(), }).SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "Cleanup") @@ -518,13 +535,13 @@ func main() { novaClient := nova.NewNovaClient() novaClientConfig := conf.GetConfigOrDie[nova.NovaClientConfig]() if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error { - return novaClient.Init(ctx, multiclusterClient, novaClientConfig) + return novaClient.Init(ctx, cachingClient, novaClientConfig) })); err != nil { setupLog.Error(err, "unable to initialize nova client") os.Exit(1) } if err := (&nova.DeschedulingsExecutor{ - Client: multiclusterClient, + Client: cachingClient, Scheme: mgr.GetScheme(), Conf: executorConfig, NovaClient: novaClient, @@ -535,8 +552,8 @@ func main() { } if slices.Contains(mainConfig.EnabledControllers, "hypervisor-overcommit-controller") { hypervisorOvercommitController := &nova.HypervisorOvercommitController{} - hypervisorOvercommitController.Client = multiclusterClient - if err := hypervisorOvercommitController.SetupWithManager(mgr); err != nil { + hypervisorOvercommitController.Client = cachingClient + if err := hypervisorOvercommitController.SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "HypervisorOvercommitController") os.Exit(1) @@ -548,7 +565,7 @@ func main() { Monitor: filterWeigherPipelineMonitor, } // Inferred through the base controller. - controller.Client = multiclusterClient + controller.Client = cachingClient if err := (controller).SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "DecisionReconciler") os.Exit(1) @@ -568,7 +585,7 @@ func main() { Monitor: filterWeigherPipelineMonitor, } // Inferred through the base controller. - controller.Client = multiclusterClient + controller.Client = cachingClient if err := (controller).SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "DecisionReconciler") os.Exit(1) @@ -588,7 +605,7 @@ func main() { Monitor: filterWeigherPipelineMonitor, } // Inferred through the base controller. - controller.Client = multiclusterClient + controller.Client = cachingClient if err := (controller).SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "DecisionReconciler") os.Exit(1) @@ -607,7 +624,7 @@ func main() { Monitor: filterWeigherPipelineMonitor, } // Inferred through the base controller. - controller.Client = multiclusterClient + controller.Client = cachingClient if err := (controller).SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "DecisionReconciler") os.Exit(1) @@ -623,11 +640,11 @@ func main() { if slices.Contains(mainConfig.EnabledControllers, "committed-resource-reservations-controller") { setupLog.Info("enabling controller", "controller", "committed-resource-reservations-controller") - monitor := reservations.NewMonitor(multiclusterClient) + monitor := reservations.NewMonitor(cachingClient) metrics.Registry.MustRegister(&monitor) if err := (&commitments.CommitmentReservationController{ - Client: multiclusterClient, + Client: cachingClient, Scheme: mgr.GetScheme(), Conf: commitmentsConfig.ReservationController, }).SetupWithManager(mgr, multiclusterClient); err != nil { @@ -637,11 +654,11 @@ func main() { crControllerConf := commitmentsConfig.CommittedResourceController - crControllerMonitor := commitments.NewCRControllerMonitor(multiclusterClient) + crControllerMonitor := commitments.NewCRControllerMonitor(cachingClient) metrics.Registry.MustRegister(&crControllerMonitor) if err := (&commitments.CommittedResourceController{ - Client: multiclusterClient, + Client: cachingClient, Scheme: mgr.GetScheme(), Conf: crControllerConf, Monitor: &crControllerMonitor, @@ -659,7 +676,7 @@ func main() { usageReconcilerConf := commitmentsConfig.UsageReconciler usageReconcilerConf.ApplyDefaults() if err := (&commitments.UsageReconciler{ - Client: multiclusterClient, + Client: cachingClient, Conf: usageReconcilerConf, VMSource: commitmentsVMSource, Monitor: usageReconcilerMonitor, @@ -674,7 +691,7 @@ func main() { monitor := datasources.NewMonitor() metrics.Registry.MustRegister(&monitor) if err := (&openstack.OpenStackDatasourceReconciler{ - Client: multiclusterClient, + Client: cachingClient, Scheme: mgr.GetScheme(), Monitor: monitor, }).SetupWithManager(mgr, multiclusterClient); err != nil { @@ -682,7 +699,7 @@ func main() { os.Exit(1) } if err := (&prometheus.PrometheusDatasourceReconciler{ - Client: multiclusterClient, + Client: cachingClient, Scheme: mgr.GetScheme(), Monitor: monitor, }).SetupWithManager(mgr, multiclusterClient); err != nil { @@ -695,7 +712,7 @@ func main() { monitor := extractor.NewMonitor() metrics.Registry.MustRegister(&monitor) if err := (&extractor.KnowledgeReconciler{ - Client: multiclusterClient, + Client: cachingClient, Scheme: mgr.GetScheme(), Monitor: monitor, Conf: conf.GetConfigOrDie[extractor.KnowledgeReconcilerConfig](), @@ -704,7 +721,7 @@ func main() { os.Exit(1) } if err := (&extractor.TriggerReconciler{ - Client: multiclusterClient, + Client: cachingClient, Scheme: mgr.GetScheme(), Conf: conf.GetConfigOrDie[extractor.TriggerReconcilerConfig](), }).SetupWithManager(mgr, multiclusterClient); err != nil { @@ -716,7 +733,7 @@ func main() { setupLog.Info("enabling controller", "controller", "kpis-controller") kpisControllerConfig := conf.GetConfigOrDie[kpis.ControllerConfig]() if err := (&kpis.Controller{ - Client: multiclusterClient, + Client: cachingClient, Config: kpisControllerConfig, }).SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "KPIController") @@ -748,7 +765,7 @@ func main() { if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error { // Create PostgresReader from the configured Datasource CRD // This runs after the cache is started - postgresReader, err := external.NewPostgresReader(ctx, multiclusterClient, failoverConfig.DatasourceName) + postgresReader, err := external.NewPostgresReader(ctx, cachingClient, failoverConfig.DatasourceName) if err != nil { setupLog.Error(err, "unable to create postgres reader for failover controller", "datasourceName", failoverConfig.DatasourceName) @@ -764,7 +781,7 @@ func main() { // 1. Watch-based per-reservation reconciliation (acknowledgment, validation) // 2. Periodic bulk VM processing (creating/assigning reservations) failoverController := failover.NewFailoverReservationController( - multiclusterClient, + cachingClient, vmSource, failoverConfig, schedulerClient, @@ -807,12 +824,12 @@ func main() { os.Exit(1) } - capacityMonitor := capacity.NewMonitor(multiclusterClient) + capacityMonitor := capacity.NewMonitor(cachingClient) if err := metrics.Registry.Register(&capacityMonitor); err != nil { setupLog.Error(err, "failed to register capacity monitor metrics, continuing without metrics") } - if err := capacity.NewController(multiclusterClient, capacityConfig, commitmentsVMSource). + if err := capacity.NewController(cachingClient, capacityConfig, commitmentsVMSource). SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "capacity") os.Exit(1) @@ -844,7 +861,7 @@ func main() { // Defer initialization until the manager starts (cache must be ready for postgres reader) if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error { // Create PostgresReader from the configured Datasource CRD - postgresReader, err := external.NewPostgresReader(ctx, multiclusterClient, datasourceName) + postgresReader, err := external.NewPostgresReader(ctx, cachingClient, datasourceName) if err != nil { setupLog.Error(err, "unable to create postgres reader for quota controller", "datasourceName", datasourceName) @@ -857,7 +874,7 @@ func main() { // Create the quota controller quotaController := quota.NewQuotaController( - multiclusterClient, + cachingClient, vmSource, quotaConfig, quotaMetrics, @@ -919,11 +936,11 @@ func main() { setupLog.Info("starting commitments syncer") syncerMonitor := commitments.NewSyncerMonitor() must.Succeed(metrics.Registry.Register(syncerMonitor)) - syncer := commitments.NewSyncer(multiclusterClient, syncerMonitor) + syncer := commitments.NewSyncer(cachingClient, syncerMonitor) syncerConfig := conf.GetConfigOrDie[commitments.SyncerConfig]() syncerConfig.FlavorGroupResourceConfig = commitmentsConfig.API.FlavorGroupResourceConfig if err := (&task.Runner{ - Client: multiclusterClient, + Client: cachingClient, Interval: syncerConfig.SyncInterval.Duration, Name: "commitments-sync-task", Run: func(ctx context.Context) error { return syncer.SyncReservations(ctx) }, @@ -937,11 +954,11 @@ func main() { setupLog.Info("starting nova history cleanup task") historyCleanupConfig := conf.GetConfigOrDie[nova.HistoryCleanupConfig]() if err := (&task.Runner{ - Client: multiclusterClient, + Client: cachingClient, Interval: time.Hour, Name: "nova-history-cleanup-task", Run: func(ctx context.Context) error { - return nova.HistoryCleanup(ctx, multiclusterClient, historyCleanupConfig) + return nova.HistoryCleanup(ctx, cachingClient, historyCleanupConfig) }, }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "unable to add nova history cleanup task to manager") @@ -952,11 +969,11 @@ func main() { setupLog.Info("starting manila history cleanup task") historyCleanupConfig := conf.GetConfigOrDie[manila.HistoryCleanupConfig]() if err := (&task.Runner{ - Client: multiclusterClient, + Client: cachingClient, Interval: time.Hour, Name: "manila-history-cleanup-task", Run: func(ctx context.Context) error { - return manila.HistoryCleanup(ctx, multiclusterClient, historyCleanupConfig) + return manila.HistoryCleanup(ctx, cachingClient, historyCleanupConfig) }, }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "unable to add manila history cleanup task to manager") @@ -967,11 +984,11 @@ func main() { setupLog.Info("starting cinder history cleanup task") historyCleanupConfig := conf.GetConfigOrDie[cinder.HistoryCleanupConfig]() if err := (&task.Runner{ - Client: multiclusterClient, + Client: cachingClient, Interval: time.Hour, Name: "cinder-history-cleanup-task", Run: func(ctx context.Context) error { - return cinder.HistoryCleanup(ctx, multiclusterClient, historyCleanupConfig) + return cinder.HistoryCleanup(ctx, cachingClient, historyCleanupConfig) }, }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "unable to add cinder history cleanup task to manager") diff --git a/helm/bundles/cortex-nova/values.yaml b/helm/bundles/cortex-nova/values.yaml index 6524c095e..fb5387864 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/helm/bundles/cortex-nova/values.yaml @@ -110,6 +110,9 @@ cortex: &cortex - kvm.cloud.sap/v1/Hypervisor - kvm.cloud.sap/v1/HypervisorList - v1/Secret + clientcache: + gvks: + - cortex.cloud/v1alpha1/Reservation keystoneSecretRef: name: cortex-nova-openstack-keystone namespace: default diff --git a/internal/scheduling/nova/hypervisor_overcommit_controller.go b/internal/scheduling/nova/hypervisor_overcommit_controller.go index 72e4507fc..7df253849 100644 --- a/internal/scheduling/nova/hypervisor_overcommit_controller.go +++ b/internal/scheduling/nova/hypervisor_overcommit_controller.go @@ -217,7 +217,7 @@ func (c *HypervisorOvercommitController) predicateRemoteHypervisor() predicate.P // SetupWithManager sets up the controller with the Manager and a multicluster // client. The multicluster client is used to watch for changes in the // Hypervisor CRD across all clusters and trigger reconciliations accordingly. -func (c *HypervisorOvercommitController) SetupWithManager(mgr ctrl.Manager) (err error) { +func (c *HypervisorOvercommitController) SetupWithManager(mgr ctrl.Manager, mcl *multicluster.Client) (err error) { // This will load the config in a safe way and gracefully handle errors. c.config, err = conf.GetConfig[HypervisorOvercommitConfig]() if err != nil { @@ -227,12 +227,6 @@ func (c *HypervisorOvercommitController) SetupWithManager(mgr ctrl.Manager) (err if err := c.config.Validate(); err != nil { return err } - // Check that the provided client is a multicluster client, since we need - // that to watch for hypervisors across clusters. - mcl, ok := c.Client.(*multicluster.Client) - if !ok { - return errors.New("provided client must be a multicluster client") - } bldr := multicluster.BuildController(mcl, mgr) // The hypervisor crd may be distributed across multiple remote clusters. bldr, err = bldr.WatchesMulticluster(&hv1.Hypervisor{}, diff --git a/internal/scheduling/nova/hypervisor_overcommit_controller_test.go b/internal/scheduling/nova/hypervisor_overcommit_controller_test.go index e52669c3a..f122831eb 100644 --- a/internal/scheduling/nova/hypervisor_overcommit_controller_test.go +++ b/internal/scheduling/nova/hypervisor_overcommit_controller_test.go @@ -725,7 +725,7 @@ func TestHypervisorOvercommitController_SetupWithManager_InvalidClient(t *testin // SetupWithManager should fail - either because config loading fails // (in test environment without config files) or because the client // is not a multicluster client. - err := controller.SetupWithManager(mgr) + err := controller.SetupWithManager(mgr, nil) if err == nil { t.Error("expected error when calling SetupWithManager, got nil") } diff --git a/pkg/clientcache/client.go b/pkg/clientcache/client.go new file mode 100644 index 000000000..c80fde968 --- /dev/null +++ b/pkg/clientcache/client.go @@ -0,0 +1,650 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package clientcache + +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/client" +) + +// 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 +} + +// objectKey identifies an object within a GVK by namespace and name. +type objectKey struct { + namespace string + name string +} + +func keyForObject(obj client.Object) objectKey { + return objectKey{namespace: obj.GetNamespace(), name: obj.GetName()} +} + +// lockKey identifies the object whose write path a keyedMutex serializes. +type lockKey struct { + gvk schema.GroupVersionKind + key 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() + } +} + +// 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 + +// CachingClient 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 CachingClient struct { + client.Client // inner client, used for delegation + + inner Client + scheme *runtime.Scheme + ttl time.Duration + gvks map[schema.GroupVersionKind]bool + + mu sync.RWMutex + byGVK map[schema.GroupVersionKind]map[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 +} + +// New builds a CachingClient wrapping inner. informers supplies the informers +// used for eviction, scheme resolves object GVKs, and conf lists the GVKs to +// overlay and the TTL. GVK strings are formatted as "//" +// and are resolved against scheme. +func New(inner Client, scheme *runtime.Scheme, conf Config) (*CachingClient, error) { + gvks, err := resolveGVKs(scheme, conf.GVKs) + if err != nil { + return nil, err + } + ttl := conf.TTL.Duration + if ttl <= 0 { + ttl = defaultTTL + } + return &CachingClient{ + Client: inner, + inner: inner, + scheme: scheme, + ttl: ttl, + gvks: gvks, + byGVK: make(map[schema.GroupVersionKind]map[objectKey]*entry), + indexers: make(map[schema.GroupVersionKind]map[string]client.IndexerFunc), + writeLocks: newKeyedMutex(), + }, 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("clientcache: 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 *CachingClient) 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 *CachingClient) 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 *CachingClient) upsert(gvk schema.GroupVersionKind, obj client.Object) { + c.mu.Lock() + defer c.mu.Unlock() + c.ensureGVK(gvk) + c.byGVK[gvk][keyForObject(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 *CachingClient) tombstone(gvk schema.GroupVersionKind, obj client.Object) { + c.mu.Lock() + defer c.mu.Unlock() + c.ensureGVK(gvk) + c.byGVK[gvk][keyForObject(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 *CachingClient) evictIfSeen(gvk schema.GroupVersionKind, obj client.Object) { + c.mu.Lock() + defer c.mu.Unlock() + entries, ok := c.byGVK[gvk] + if !ok { + return + } + key := keyForObject(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 *CachingClient) getEntry(gvk schema.GroupVersionKind, key 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 *CachingClient) 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 *CachingClient) 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 *CachingClient) ensureGVK(gvk schema.GroupVersionKind) { + if c.byGVK[gvk] == nil { + c.byGVK[gvk] = make(map[objectKey]*entry) + } +} + +// overlayList merges the overlay entries for the GVK into the informer result, +// deduplicating by objectKey (overlay wins), dropping tombstones, and filtering +// overlay-only entries against the list options' label and field selectors. +func (c *CachingClient) 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[objectKey]bool, len(entries)) + + for _, item := range existing { + obj, ok := item.(client.Object) + if !ok { + result = append(result, item) + continue + } + key := keyForObject(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 *CachingClient) 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 *CachingClient) 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 *CachingClient) 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: keyForObject(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 *CachingClient) 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: keyForObject(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 *CachingClient) 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: keyForObject(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 *CachingClient) 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: keyForObject(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 *CachingClient) 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 *CachingClient) 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, objectKey{namespace: key.Namespace, name: key.Name}) + 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 *CachingClient) 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 *CachingClient) IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error { + if indexer, ok := c.Client.(client.FieldIndexer); ok { + if err := indexer.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 *CachingClient) 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 *CachingClient + 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: keyForObject(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: keyForObject(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/clientcache/client_test.go b/pkg/clientcache/client_test.go new file mode 100644 index 000000000..1d7934a89 --- /dev/null +++ b/pkg/clientcache/client_test.go @@ -0,0 +1,919 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package clientcache + +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" + + "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) + } +} + +// fakeClient composes a fake client.Client with a fakeInformer to satisfy the +// clientcache.Client interface. +type fakeClient struct { + client.Client + inf *fakeInformer +} + +func (f *fakeClient) GetInformersForKind(_ context.Context, _ client.Object) ([]ccache.Informer, error) { + return []ccache.Informer{f.inf}, nil +} + +func newTestClient(t *testing.T, objs ...client.Object) *fakeClient { + t.Helper() + return &fakeClient{ + Client: fake.NewClientBuilder(). + WithScheme(testScheme(t)). + 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(), + inf: &fakeInformer{}, + } +} + +// errClient wraps a Client and injects configurable errors into mutating/read +// operations, so the error-propagation paths of CachingClient (which must not +// touch the overlay on failure) can be exercised. +type errClient struct { + 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}, + } +} + +func newCaching(t *testing.T, inner Client) *CachingClient { + t.Helper() + c, err := New(inner, testScheme(t), reservationConfig()) + if err != nil { + t.Fatalf("New: %v", err) + } + return c +} + +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 := New(newTestClient(t), testScheme(t), 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, err := New(newTestClient(t), testScheme(t), Config{ + GVKs: []string{"cortex.cloud/v1alpha1/Reservation"}, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + if c.ttl != defaultTTL { + t.Fatalf("expected ttl %v, got %v", defaultTTL, c.ttl) + } +} + +func TestNewExplicitTTL(t *testing.T) { + c, err := New(newTestClient(t), testScheme(t), Config{ + GVKs: []string{"cortex.cloud/v1alpha1/Reservation"}, + TTL: metav1.Duration{Duration: 90 * time.Second}, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + if c.ttl != 90*time.Second { + t.Fatalf("expected ttl 90s, got %v", c.ttl) + } +} + +// --- 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(), 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(context.Background()) + 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(), 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(), 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.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.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, err := New(inner, testScheme(t), Config{}) + if err != nil { + t.Fatalf("New: %v", err) + } + 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(), 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 + op func(c *CachingClient, r *v1alpha1.Reservation) error + }{ + { + name: "create", + mkClient: func(inner *fakeClient) Client { return &errClient{Client: inner, createErr: sentinel} }, + op: func(c *CachingClient, r *v1alpha1.Reservation) error { return c.Create(context.Background(), r) }, + }, + { + name: "update", + rv: "1", + mkClient: func(inner *fakeClient) Client { return &errClient{Client: inner, updateErr: sentinel} }, + op: func(c *CachingClient, r *v1alpha1.Reservation) error { return c.Update(context.Background(), r) }, + }, + { + name: "patch", + rv: "1", + mkClient: func(inner *fakeClient) Client { return &errClient{Client: inner, patchErr: sentinel} }, + op: func(c *CachingClient, 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 { return &errClient{Client: inner, deleteErr: sentinel} }, + op: func(c *CachingClient, 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 := newCaching(t, tc.mkClient(base)) + + if err := tc.op(c, r); !errors.Is(err, sentinel) { + t.Fatalf("expected sentinel error, got %v", err) + } + if _, ok := c.getEntry(reservationGVK(), 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 *CachingClient, 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 *CachingClient, 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 *CachingClient, 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 *CachingClient, 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.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, 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 := newCaching(t, &errClient{Client: newTestClient(t), getErr: sentinel}) + 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, err := New(&errClient{Client: newTestClient(t), getErr: sentinel}, testScheme(t), Config{}) + if err != nil { + t.Fatalf("New: %v", err) + } + 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 := newCaching(t, &errClient{Client: newTestClient(t), listErr: sentinel}) + 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(), 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(), 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, err := New(inner, testScheme(t), Config{}) + if err != nil { + t.Fatalf("New: %v", err) + } + var cur v1alpha1.Reservation + if err := inner.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(), 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 + 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)} + c := newCaching(t, oc) + + 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(), 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/clientcache/config.go b/pkg/clientcache/config.go new file mode 100644 index 000000000..57d5d7a70 --- /dev/null +++ b/pkg/clientcache/config.go @@ -0,0 +1,24 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package clientcache + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// Config configures the transparent in-process overlay cache. +type Config struct { + // 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,omitempty"` +} + +// RootConfig is the top-level config key for the client cache. +type RootConfig struct { + ClientCache Config `json:"clientcache"` +} diff --git a/pkg/clientcache/interfaces.go b/pkg/clientcache/interfaces.go new file mode 100644 index 000000000..c2c3e6dd5 --- /dev/null +++ b/pkg/clientcache/interfaces.go @@ -0,0 +1,22 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package clientcache + +import ( + "context" + + "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// Client is the interface the CachingClient requires of its inner client. +// It extends client.Client with the informer access needed for overlay eviction. +// *multicluster.Client satisfies this interface. +type Client interface { + client.Client + // GetInformersForKind returns all informers serving the GVK of the given + // object. The cache attaches Add/Update event handlers to each informer to + // evict overlay entries once the real object appears in the informer cache. + GetInformersForKind(ctx context.Context, obj client.Object) ([]cache.Informer, error) +} diff --git a/pkg/clientcache/runnable.go b/pkg/clientcache/runnable.go new file mode 100644 index 000000000..7196dd787 --- /dev/null +++ b/pkg/clientcache/runnable.go @@ -0,0 +1,99 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package clientcache + +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 *CachingClient) Start(ctx context.Context) error { + log := ctrl.LoggerFrom(ctx).WithName("clientcache") + + 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 + } + informers, err := c.inner.GetInformersForKind(ctx, obj) + if err != nil { + log.Error(err, "failed to get informers for gvk; eviction disabled for it", "gvk", gvk) + continue + } + handler := c.evictionHandler(gvk) + for _, inf := range informers { + if _, err := inf.AddEventHandler(handler); 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 CachingClient'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 *CachingClient) 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 *CachingClient) 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 *CachingClient) 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("clientcache: 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..d5c74eeb0 100644 --- a/pkg/multicluster/client.go +++ b/pkg/multicluster/client.go @@ -18,6 +18,7 @@ import ( "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/rest" ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/cache" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/cluster" ) @@ -224,6 +225,30 @@ func (c *Client) ClustersForGVK(gvk schema.GroupVersionKind) ([]cluster.Cluster, return clusters, nil } +// GetInformersForKind returns the informers of all clusters serving the GVK of +// the given object. It is used by the in-process client cache to attach +// eviction event handlers. The GVK is resolved against the home scheme and must +// be explicitly configured in home or a remote cluster. +func (c *Client) GetInformersForKind(ctx context.Context, obj client.Object) ([]cache.Informer, error) { + gvk, err := c.GVKFromHomeScheme(obj) + if err != nil { + return nil, err + } + clusters, err := c.ClustersForGVK(gvk) + if err != nil { + return nil, err + } + informers := make([]cache.Informer, 0, len(clusters)) + for _, cl := range clusters { + inf, err := cl.GetCache().GetInformer(ctx, obj) + if err != nil { + return nil, err + } + informers = append(informers, inf) + } + return informers, nil +} + // clusterForWrite uses a ResourceRouter to determine which remote cluster // a resource should be written to based on the resource content and cluster labels. //