diff --git a/cmd/manager/main.go b/cmd/manager/main.go index a7ae683d3..7d8473fb7 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -64,6 +64,7 @@ import ( commitmentsapi "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations/commitments/api" "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations/failover" "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" @@ -393,6 +394,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 serves as both the inner client.Client and + // the InformerSource; the cache itself has no multicluster dependency. + clientCacheConfig := conf.GetConfigOrDie[clientcache.RootConfig]() + cachingClient, err := clientcache.New(multiclusterClient, 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]() @@ -424,10 +441,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")) } @@ -447,8 +464,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) @@ -463,7 +480,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) @@ -474,7 +491,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) @@ -482,7 +499,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") @@ -502,13 +519,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, @@ -519,8 +536,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) @@ -532,7 +549,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) @@ -552,7 +569,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) @@ -572,7 +589,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) @@ -591,7 +608,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,11 +624,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 { @@ -621,11 +638,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, @@ -643,7 +660,7 @@ func main() { usageReconcilerConf := commitmentsConfig.UsageReconciler usageReconcilerConf.ApplyDefaults() if err := (&commitments.UsageReconciler{ - Client: multiclusterClient, + Client: cachingClient, Conf: usageReconcilerConf, VMSource: commitmentsVMSource, Monitor: usageReconcilerMonitor, @@ -658,7 +675,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 { @@ -666,7 +683,7 @@ func main() { os.Exit(1) } if err := (&prometheus.PrometheusDatasourceReconciler{ - Client: multiclusterClient, + Client: cachingClient, Scheme: mgr.GetScheme(), Monitor: monitor, }).SetupWithManager(mgr, multiclusterClient); err != nil { @@ -679,7 +696,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](), @@ -688,7 +705,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 { @@ -700,7 +717,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") @@ -732,7 +749,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) @@ -748,7 +765,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, @@ -791,12 +808,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) @@ -828,7 +845,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) @@ -841,7 +858,7 @@ func main() { // Create the quota controller quotaController := quota.NewQuotaController( - multiclusterClient, + cachingClient, vmSource, quotaConfig, quotaMetrics, @@ -903,11 +920,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) }, @@ -921,11 +938,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") @@ -936,11 +953,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") @@ -951,11 +968,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 27acd10e1..287bdd205 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/cache.go b/pkg/clientcache/cache.go new file mode 100644 index 000000000..639fd7663 --- /dev/null +++ b/pkg/clientcache/cache.go @@ -0,0 +1,262 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package clientcache + +import ( + "strconv" + "sync" + "time" + + "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 +} + +// overlay is the generic, client-independent core of the cache. It stores +// pending writes keyed by GVK and objectKey and merges them into informer +// read results until the real object is observed in an informer (eviction). +type overlay struct { + mu sync.RWMutex + byGVK map[schema.GroupVersionKind]map[objectKey]*entry + ttl time.Duration + // indexers holds the IndexerFunc per field per GVK, captured from + // IndexField calls, so overlay entries can be matched against FieldSelectors. + indexers map[schema.GroupVersionKind]map[string]client.IndexerFunc +} + +func newOverlay(ttl time.Duration) *overlay { + return &overlay{ + byGVK: make(map[schema.GroupVersionKind]map[objectKey]*entry), + ttl: ttl, + indexers: make(map[schema.GroupVersionKind]map[string]client.IndexerFunc), + } +} + +func keyForObject(obj client.Object) objectKey { + return objectKey{namespace: obj.GetNamespace(), name: obj.GetName()} +} + +// upsert stores a live (non-tombstone) entry for the object. +func (o *overlay) upsert(gvk schema.GroupVersionKind, obj client.Object) { + o.mu.Lock() + defer o.mu.Unlock() + o.ensureGVK(gvk) + o.byGVK[gvk][keyForObject(obj)] = &entry{ + obj: obj.DeepCopyObject().(client.Object), + uid: obj.GetUID(), + resourceVersion: obj.GetResourceVersion(), + deleted: false, + expiresAt: time.Now().Add(o.ttl), + } +} + +// remove stores a tombstone so the object is filtered out of reads until the +// deletion is observed in an informer. +func (o *overlay) remove(gvk schema.GroupVersionKind, obj client.Object) { + o.mu.Lock() + defer o.mu.Unlock() + o.ensureGVK(gvk) + o.byGVK[gvk][keyForObject(obj)] = &entry{ + obj: obj.DeepCopyObject().(client.Object), + uid: obj.GetUID(), + resourceVersion: obj.GetResourceVersion(), + deleted: true, + expiresAt: time.Now().Add(o.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 (o *overlay) evictIfSeen(gvk schema.GroupVersionKind, obj client.Object) { + o.mu.Lock() + defer o.mu.Unlock() + entries, ok := o.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) +} + +// get returns the overlay entry for the key, if present. +func (o *overlay) get(gvk schema.GroupVersionKind, key objectKey) (*entry, bool) { + o.mu.RLock() + defer o.mu.RUnlock() + entries, ok := o.byGVK[gvk] + if !ok { + return nil, false + } + e, ok := entries[key] + return e, ok +} + +// 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 (o *overlay) overlayList(gvk schema.GroupVersionKind, existing []runtime.Object, lo *client.ListOptions) []runtime.Object { + o.mu.RLock() + defer o.mu.RUnlock() + entries := o.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 + } + 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 !o.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 (o *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 := o.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 (o *overlay) fieldSetLocked(gvk schema.GroupVersionKind, obj client.Object) fields.Set { + set := fields.Set{} + for field, fn := range o.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 +} + +// registerIndex captures an IndexerFunc for a field so overlay entries can be +// matched against MatchingFields queries. +func (o *overlay) registerIndex(gvk schema.GroupVersionKind, field string, fn client.IndexerFunc) { + o.mu.Lock() + defer o.mu.Unlock() + if o.indexers[gvk] == nil { + o.indexers[gvk] = make(map[string]client.IndexerFunc) + } + o.indexers[gvk][field] = fn +} + +// cleanupExpired removes entries whose TTL has passed. +func (o *overlay) cleanupExpired(now time.Time) { + o.mu.Lock() + defer o.mu.Unlock() + for _, entries := range o.byGVK { + for key, e := range entries { + if now.After(e.expiresAt) { + delete(entries, key) + } + } + } +} + +func (o *overlay) ensureGVK(gvk schema.GroupVersionKind) { + if o.byGVK[gvk] == nil { + o.byGVK[gvk] = make(map[objectKey]*entry) + } +} + +// 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 +} diff --git a/pkg/clientcache/cache_test.go b/pkg/clientcache/cache_test.go new file mode 100644 index 000000000..5d3cceead --- /dev/null +++ b/pkg/clientcache/cache_test.go @@ -0,0 +1,436 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package clientcache + +import ( + "context" + "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" + +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 { + r := &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + UID: types.UID("uid-" + name), + ResourceVersion: rv, + }, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + AvailabilityZone: az, + }, + } + return r +} + +func newTestClient(t *testing.T, objs ...client.Object) client.Client { + t.Helper() + return 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() +} + +// 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) + } +} + +// fakeInformerSource returns a single shared fakeInformer for all kinds. +type fakeInformerSource struct { + inf *fakeInformer +} + +func (s *fakeInformerSource) GetInformersForKind(ctx context.Context, obj client.Object) ([]ccache.Informer, error) { + return []ccache.Informer{s.inf}, nil +} + +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.Client, src InformerSource) *CachingClient { + t.Helper() + c, err := New(inner, src, 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 +} + +// 1. Write/Read: Create then immediate List/Get shows the object despite an +// empty informer (fake inner client without the object pre-loaded... but the +// fake client persists creates, so we simulate informer lag by deleting from +// inner after caching — instead we verify overlay independently below). +func TestCreateThenGetVisible(t *testing.T) { + inner := newTestClient(t) + src := &fakeInformerSource{inf: &fakeInformer{}} + c := newCaching(t, inner, src) + + 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) + } +} + +// 2. Overlay with empty informer: inner List returns [], overlay entry still in +// the result. +func TestOverlayWhenInnerEmpty(t *testing.T) { + inner := newTestClient(t) // no objects + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + // Directly seed the overlay to simulate a write whose object is not yet in + // the (empty) inner client. + c.overlay.upsert(reservationGVK(), newReservation("res-2", "az-1", "5")) + + items := listReservations(t, c) + if len(items) != 1 || items[0].Name != "res-2" { + t.Fatalf("expected overlay entry res-2, got %+v", items) + } +} + +// 3. Eviction: an informer sighting (Add or Update) evicts the overlay entry +// only when it matches by UID and carries a ResourceVersion >= the cached one. +func TestEviction(t *testing.T) { + // Cached entry is always uid-res-3 @ RV 10. + const cachedRV = "10" + cases := []struct { + name string + useUpdate bool // fire OnUpdate instead of OnAdd + observedUID string // "" => reuse the cached object's UID + 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) + inf := &fakeInformer{} + c := newCaching(t, inner, &fakeInformerSource{inf: inf}) + + 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 { + inf.mu.Lock() + defer inf.mu.Unlock() + return len(inf.handlers) > 0 + }) + + c.overlay.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 { + inf.fireUpdate(nil, observed) + } else { + inf.fireAdd(observed) + } + + _, present := c.overlay.get(reservationGVK(), objectKey{name: "res-3"}) + if present == tc.wantEvicted { + t.Fatalf("evicted=%v, want evicted=%v", !present, tc.wantEvicted) + } + }) + } +} + +// TestEvictionIgnoresNonObject: an informer event carrying a non-client.Object +// payload is ignored and does not panic or evict. +func TestEvictionIgnoresNonObject(t *testing.T) { + inner := newTestClient(t) + inf := &fakeInformer{} + c := newCaching(t, inner, &fakeInformerSource{inf: inf}) + + 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 { + inf.mu.Lock() + defer inf.mu.Unlock() + return len(inf.handlers) > 0 + }) + + c.overlay.upsert(reservationGVK(), newReservation("res-x", "az-1", "1")) + inf.fireAdd("not-an-object") + if _, ok := c.overlay.get(reservationGVK(), objectKey{name: "res-x"}); !ok { + t.Fatalf("non-object event must not evict the entry") + } +} + +// 4. TTL: cleanupExpired removes expired entries. +func TestTTLCleanup(t *testing.T) { + o := newOverlay(time.Minute) + o.upsert(reservationGVK(), newReservation("res-4", "az-1", "1")) + // Force expiry. + o.mu.Lock() + for _, entries := range o.byGVK { + for _, e := range entries { + e.expiresAt = time.Now().Add(-time.Second) + } + } + o.mu.Unlock() + o.cleanupExpired(time.Now()) + if _, ok := o.get(reservationGVK(), objectKey{name: "res-4"}); ok { + t.Fatalf("expired entry should be removed") + } +} + +// 5. Tombstone: Delete filters the object from List/Get even though inner still +// has it. +func TestTombstone(t *testing.T) { + r := newReservation("res-5", "az-1", "") + inner := newTestClient(t, r) + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + if err := c.Delete(context.Background(), r); err != nil { + t.Fatalf("Delete: %v", err) + } + // Re-add to inner to simulate informer lag: fake client already removed it, + // so re-create via inner directly (bypassing overlay). + if err := inner.Create(context.Background(), newReservation("res-5", "az-1", "")); err != nil { + t.Fatalf("re-create inner: %v", err) + } + + var got v1alpha1.Reservation + err := c.Get(context.Background(), types.NamespacedName{Name: "res-5"}, &got) + if !apierrors.IsNotFound(err) { + t.Fatalf("expected NotFound for tombstoned object, got %v", err) + } + items := listReservations(t, c) + if len(items) != 0 { + t.Fatalf("expected tombstone to filter from list, got %+v", items) + } +} + +// 6. Update/Patch: newer overlay version overrides stale inner read. +func TestUpdateOverridesInner(t *testing.T) { + r := newReservation("res-6", "az-old", "1") + inner := newTestClient(t, r) + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + // Read current to obtain the up-to-date ResourceVersion for update. + 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) + } +} + +// 7. Label-matching: overlay-only entry appears only for matching MatchingLabels. +func TestLabelMatching(t *testing.T) { + inner := newTestClient(t) + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + r := newReservation("res-7", "az-1", "1") + r.Labels = map[string]string{"team": "a"} + c.overlay.upsert(reservationGVK(), r) + + match := listReservations(t, c, client.MatchingLabels{"team": "a"}) + if len(match) != 1 { + t.Fatalf("expected match for team=a, got %+v", match) + } + noMatch := listReservations(t, c, client.MatchingLabels{"team": "b"}) + if len(noMatch) != 0 { + t.Fatalf("expected no match for team=b, got %+v", noMatch) + } +} + +// 8. Field-matching: after IndexField registration, overlay-only entry appears +// only for matching MatchingFields. +func TestFieldMatching(t *testing.T) { + inner := newTestClient(t) + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + 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.overlay.upsert(reservationGVK(), newReservation("res-8", "az-1", "1")) + + match := listReservations(t, c, client.MatchingFields{azIndexField: "az-1"}) + if len(match) != 1 { + t.Fatalf("expected field match az-1, got %+v", match) + } + noMatch := listReservations(t, c, client.MatchingFields{azIndexField: "az-2"}) + if len(noMatch) != 0 { + t.Fatalf("expected no field match az-2, got %+v", noMatch) + } +} + +// 9. Non-cached GVK: calls pass through unchanged (no overlay effect). +func TestNonCachedGVKPassthrough(t *testing.T) { + inner := newTestClient(t) + // Config with no GVKs → Reservation is not cached. + c, err := New(inner, &fakeInformerSource{inf: &fakeInformer{}}, 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) + } + // Overlay must be empty for non-cached GVK. + if _, ok := c.overlay.get(reservationGVK(), objectKey{name: "res-9"}); ok { + t.Fatalf("non-cached GVK should not populate overlay") + } + // Delete it in inner, then Get should be NotFound (no overlay resurrection). + 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) + } +} + +// 10. Dedup: object present in both informer (inner) and overlay appears once. +func TestDedup(t *testing.T) { + r := newReservation("res-10", "az-1", "1") + inner := newTestClient(t, r) + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + // Overlay holds a newer version of the same object. + newer := newReservation("res-10", "az-1", "2") + newer.Spec.TargetHost = "host-x" + c.overlay.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]) + } +} + +// helpers + +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") +} diff --git a/pkg/clientcache/client.go b/pkg/clientcache/client.go new file mode 100644 index 000000000..e9a4b01f6 --- /dev/null +++ b/pkg/clientcache/client.go @@ -0,0 +1,265 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package clientcache + +import ( + "context" + "errors" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// 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. +type CachingClient struct { + client.Client // inner client, used for delegation + + informers InformerSource + scheme *runtime.Scheme + overlay *overlay + ttl time.Duration + gvks map[schema.GroupVersionKind]bool +} + +// 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.Client, informers InformerSource, 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, + informers: informers, + scheme: scheme, + overlay: newOverlay(ttl), + ttl: ttl, + gvks: gvks, + }, 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 +} + +// Inner returns the wrapped client, e.g. for use with a controller Builder that +// needs the raw client rather than the caching wrapper. +func (c *CachingClient) Inner() client.Client { return c.Client } + +// 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 +} + +// 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 { + if err := c.Client.Create(ctx, obj, opts...); err != nil { + return err + } + if gvk, cached := c.gvkFor(obj); cached { + c.overlay.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 { + if err := c.Client.Update(ctx, obj, opts...); err != nil { + return err + } + if gvk, cached := c.gvkFor(obj); cached { + c.overlay.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 { + if err := c.Client.Patch(ctx, obj, patch, opts...); err != nil { + return err + } + if gvk, cached := c.gvkFor(obj); cached { + c.overlay.upsert(gvk, obj) + } + return nil +} + +// Delete delegates to the inner client and, on success for a cached GVK, stores +// a tombstone in the overlay. +func (c *CachingClient) Delete(ctx context.Context, obj client.Object, opts ...client.DeleteOption) error { + if err := c.Client.Delete(ctx, obj, opts...); err != nil { + return err + } + if gvk, cached := c.gvkFor(obj); cached { + c.overlay.remove(gvk, obj) + } + 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.overlay.get(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. + if cpErr := c.scheme.Convert(e.obj, 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.overlay.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.overlay.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 { + if err := s.inner.Update(ctx, obj, opts...); err != nil { + return err + } + if gvk, cached := s.c.gvkFor(obj); cached { + s.c.overlay.upsert(gvk, obj) + } + return nil +} + +func (s *statusWriter) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if err := s.inner.Patch(ctx, obj, patch, opts...); err != nil { + return err + } + if gvk, cached := s.c.gvkFor(obj); cached { + s.c.overlay.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..08d043f26 --- /dev/null +++ b/pkg/clientcache/client_test.go @@ -0,0 +1,470 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package clientcache + +import ( + "context" + "errors" + "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/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" +) + +// errClient wraps an inner client.Client and injects a configurable error into +// each mutating/read operation, so the error-propagation paths of +// CachingClient (which must not touch the overlay on failure) can be exercised. +type errClient struct { + client.Client + createErr error + updateErr error + patchErr error + deleteErr 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) 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 (and its overlay). Used to prove that reads +// through the caching client 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, bypassing the caching wrapper. +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) + } +} + +// TestNewUnknownGVKError: New fails when a configured GVK string is not +// registered in the scheme. +func TestNewUnknownGVKError(t *testing.T) { + inner := newTestClient(t) + _, err := New(inner, &fakeInformerSource{inf: &fakeInformer{}}, testScheme(t), Config{ + GVKs: []string{"cortex.cloud/v1alpha1/DoesNotExist"}, + }) + if err == nil { + t.Fatalf("expected error for unknown GVK, got nil") + } +} + +// TestNewDefaultTTL: a zero TTL in the config falls back to defaultTTL. +func TestNewDefaultTTL(t *testing.T) { + inner := newTestClient(t) + c, err := New(inner, &fakeInformerSource{inf: &fakeInformer{}}, 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) + } + if c.overlay.ttl != defaultTTL { + t.Fatalf("expected overlay ttl %v, got %v", defaultTTL, c.overlay.ttl) + } +} + +// TestNewExplicitTTL: a non-zero TTL is honoured verbatim. +func TestNewExplicitTTL(t *testing.T) { + inner := newTestClient(t) + c, err := New(inner, &fakeInformerSource{inf: &fakeInformer{}}, 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) + } +} + +// TestInnerReturnsWrappedClient: Inner returns the exact client passed to New. +func TestInnerReturnsWrappedClient(t *testing.T) { + inner := newTestClient(t) + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + if c.Inner() != inner { + t.Fatalf("Inner did not return the wrapped client") + } +} + +// TestWriteErrorLeavesOverlayUntouched: for every mutating method, a failing +// inner call surfaces the error and leaves the overlay untouched (no live entry +// and no tombstone). +func TestWriteErrorLeavesOverlayUntouched(t *testing.T) { + sentinel := errors.New("boom") + cases := []struct { + name string + rv string // ResourceVersion for the object passed to the op + seed bool // pre-seed the object in the inner client (delete needs it) + // mkClient wraps an inner client (which may already contain r) with the + // relevant injected error. + mkClient func(inner client.Client) client.Client + op func(c *CachingClient, r *v1alpha1.Reservation) error + }{ + { + name: "create", + mkClient: func(inner client.Client) client.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 client.Client) client.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 client.Client) client.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 client.Client) client.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 client.Client + if tc.seed { + base = newTestClient(t, r) + } else { + base = newTestClient(t) + } + c := newCaching(t, tc.mkClient(base), &fakeInformerSource{inf: &fakeInformer{}}) + + if err := tc.op(c, r); !errors.Is(err, sentinel) { + t.Fatalf("expected sentinel error, got %v", err) + } + if _, ok := c.overlay.get(reservationGVK(), objectKey{name: r.Name}); ok { + t.Fatalf("overlay must not be touched on %s failure", tc.name) + } + }) + } +} + +// TestWriteServedFromOverlay: after a write through the caching client, a Get +// returns the written value even though the inner client has been forced to a +// divergent (stale) value behind the cache's back. This proves the read path is +// actually served from the overlay, not merely that the overlay was written. +func TestWriteServedFromOverlay(t *testing.T) { + cases := []struct { + name string + // write performs the write under test through c, given the current + // object cur fetched from inner, and returns the value it wrote. + write func(t *testing.T, c *CachingClient, cur *v1alpha1.Reservation) string + // diverge forces the inner client to a stale value behind the cache. + diverge func(t *testing.T, inner client.Client, name string) + // read extracts the field under test from a Get result. + 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, &fakeInformerSource{inf: &fakeInformer{}}) + + 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)) + } + }) + } +} + +// TestGetPropagatesNonNotFoundError: a cached-GVK Get surfaces inner errors +// other than NotFound without consulting the overlay. +func TestGetPropagatesNonNotFoundError(t *testing.T) { + sentinel := errors.New("get boom") + inner := &errClient{Client: newTestClient(t), getErr: sentinel} + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + // Seed a live overlay entry that must NOT mask the underlying error. + c.overlay.upsert(reservationGVK(), newReservation("res-ge", "az-1", "1")) + + var got v1alpha1.Reservation + err := c.Get(context.Background(), types.NamespacedName{Name: "res-ge"}, &got) + if !errors.Is(err, sentinel) { + t.Fatalf("expected sentinel error, got %v", err) + } +} + +// TestGetOverlayResurrectsNotFound: a live overlay entry satisfies a Get that +// the inner client reports as NotFound (write not yet visible in the informer). +func TestGetOverlayResurrectsNotFound(t *testing.T) { + inner := newTestClient(t) // empty: inner Get returns NotFound + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + c.overlay.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) + } +} + +// TestGetNotFoundWithNoOverlay: inner NotFound with no overlay entry propagates +// NotFound unchanged for a cached GVK. +func TestGetNotFoundWithNoOverlay(t *testing.T) { + inner := newTestClient(t) + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + var got v1alpha1.Reservation + err := c.Get(context.Background(), types.NamespacedName{Name: "missing"}, &got) + if !apierrors.IsNotFound(err) { + t.Fatalf("expected NotFound, got %v", err) + } +} + +// TestGetNonCachedPropagatesError: for a non-cached GVK, Get is a pure +// passthrough and surfaces the inner error verbatim. +func TestGetNonCachedPropagatesError(t *testing.T) { + sentinel := errors.New("get boom") + inner := &errClient{Client: newTestClient(t), getErr: sentinel} + c, err := New(inner, &fakeInformerSource{inf: &fakeInformer{}}, 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) + } +} + +// TestListPropagatesError: for a cached GVK, a failing inner List surfaces the +// error rather than returning a partial overlay merge. +func TestListPropagatesError(t *testing.T) { + sentinel := errors.New("list boom") + inner := &errClient{Client: newTestClient(t), listErr: sentinel} + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + c.overlay.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) + } +} + +// TestStatusUpdateErrorLeavesOverlayUntouched: a failed status update does not +// populate the overlay. +func TestStatusUpdateErrorLeavesOverlayUntouched(t *testing.T) { + inner := newTestClient(t) // object absent → status update fails + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + 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.overlay.get(reservationGVK(), objectKey{name: "res-se"}); ok { + t.Fatalf("overlay must not be populated on status update failure") + } +} + +// TestStatusCreateDelegates: Status().Create delegates to the inner status +// writer (fake client reports it unsupported) and never touches the overlay. +func TestStatusCreateDelegates(t *testing.T) { + r := newReservation("res-sc", "az-1", "") + inner := newTestClient(t, r) + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + + // The fake client does not support subresource Create; we only assert the + // call is delegated (returns an error) and the overlay stays empty. + if err := c.Status().Create(context.Background(), r, r); err == nil { + t.Fatalf("expected Status().Create to fail on fake client") + } + if _, ok := c.overlay.get(reservationGVK(), objectKey{name: "res-sc"}); ok { + t.Fatalf("Status().Create must not populate the overlay") + } +} + +// TestStatusUpdateNonCachedNoOverlay: Status().Update for a non-cached GVK does +// not touch the overlay. +func TestStatusUpdateNonCachedNoOverlay(t *testing.T) { + r := newReservation("res-sn", "az-1", "") + inner := newTestClient(t, r) + c, err := New(inner, &fakeInformerSource{inf: &fakeInformer{}}, 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.overlay.get(reservationGVK(), objectKey{name: "res-sn"}); ok { + t.Fatalf("non-cached GVK status update should not populate overlay") + } +} + +// TestGVKForUnknownType: gvkFor reports not-cached for a type not registered in +// the scheme. +func TestGVKForUnknownType(t *testing.T) { + inner := newTestClient(t) + c := newCaching(t, inner, &fakeInformerSource{inf: &fakeInformer{}}) + if _, cached := c.gvkFor(&unknownObject{}); cached { + t.Fatalf("unknown type must not be reported as cached") + } +} + +// TestTrimListSuffix exercises the list-kind suffix trimming helper. +func TestTrimListSuffix(t *testing.T) { + cases := []struct { + in string + wantKind string + wantOK bool + }{ + {"ReservationList", "Reservation", true}, + {"List", "List", false}, // len(kind) not > len("List") + {"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) + } + } +} + +// 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 } 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..8c2e87599 --- /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" +) + +// InformerSource provides, per object type, the informers the cache attaches +// to for eviction purposes. It is satisfied structurally e.g. by +// *multicluster.Client (via ClustersForGVK + cluster.GetCache().GetInformer), +// so that this package does not need to import pkg/multicluster. +type InformerSource interface { + // 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..836bee0e6 --- /dev/null +++ b/pkg/clientcache/runnable.go @@ -0,0 +1,84 @@ +// 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.informers.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.overlay.cleanupExpired(now) + } + } +} + +// 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.overlay.evictIfSeen(gvk, obj) + } + return toolscachek8s.ResourceEventHandlerFuncs{ + AddFunc: func(o any) { evict(o) }, + UpdateFunc: func(_, o any) { 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 979258a44..11e35f033 100644 --- a/pkg/multicluster/client.go +++ b/pkg/multicluster/client.go @@ -17,6 +17,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" ) @@ -219,6 +220,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. //