Skip to content

feat: client cache for configurable list of gvks #1042

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
SoWieMarkus wants to merge 4 commits into
base: main
Choose a base branch
from
Open

feat: client cache for configurable list of gvks #1042

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
cd30053
feat: mutex-protected in-process map for scheduling safety
SoWieMarkus Jul 14, 2026
58a9a46
Merge branch 'main' into feat/client-cache
SoWieMarkus Jul 14, 2026
c201d3d
fix: simplify interval calculation for cleanup ticker
SoWieMarkus Jul 14, 2026
e164280
Merge branch 'main' into feat/client-cache
mblos Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 56 additions & 39 deletions cmd/manager/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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]()
Expand Down Expand Up @@ -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"))
}

Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -474,15 +491,15 @@ 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)
}
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")
Expand All @@ -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,
Expand All @@ -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 {

@PhilippMatthes PhilippMatthes Jul 20, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can this cause issues? Inside the SetupWithManager function, aren't there calls that would need to be tunneled by the caching client, such as defining indexes or resource handlers?
Is it possible to wrap it the other way around? Instead of ctrl.Client -> mcl.Client -> clientcache.Client, do ctrl.Client -> clientcache.Client -> mcl.Client? This wouldn't break the pattern here

setupLog.Error(err, "unable to create controller",
"controller", "HypervisorOvercommitController")
os.Exit(1)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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 {
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -658,15 +675,15 @@ 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 {
setupLog.Error(err, "unable to create controller", "controller", "OpenStackDatasourceReconciler")
os.Exit(1)
}
if err := (&prometheus.PrometheusDatasourceReconciler{
Client: multiclusterClient,
Client: cachingClient,
Scheme: mgr.GetScheme(),
Monitor: monitor,
}).SetupWithManager(mgr, multiclusterClient); err != nil {
Expand All @@ -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](),
Expand All @@ -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 {
Expand All @@ -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")
Expand Down Expand Up @@ -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)
Expand All @@ -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,
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -841,7 +858,7 @@ func main() {

// Create the quota controller
quotaController := quota.NewQuotaController(
multiclusterClient,
cachingClient,
vmSource,
quotaConfig,
quotaMetrics,
Expand Down Expand Up @@ -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) },
Expand All @@ -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")
Expand All @@ -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")
Expand All @@ -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")
Expand Down
3 changes: 3 additions & 0 deletions helm/bundles/cortex-nova/values.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:

@PhilippMatthes PhilippMatthes Jul 20, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it expensive to cache resources? If not, why not cache all resources w/o extra configuration?

name: cortex-nova-openstack-keystone
namespace: default
Expand Down
8 changes: 1 addition & 7 deletions internal/scheduling/nova/hypervisor_overcommit_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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{},
Expand Down
2 changes: 1 addition & 1 deletion internal/scheduling/nova/hypervisor_overcommit_controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
}
Expand Down
Loading