Repository: kubernetes-sigs/kueue

Issue:
Unexport access to Cohorts and ClusterQueues maps in hierarchy.Manager
**What would you like to be cleaned**:
https://github.com/kubernetes-sigs/kueue/blob/be27bb83d087783d7a227e2068bb874110ac9580/pkg/hierarchy/manager.go#L22-L23

**Why is this needed**:
Current requires callers to avoid modifying this map, outside of using the Manager's add/delete methods. See https://github.com/kubernetes-sigs/kueue/pull/2971#discussion_r1745385475

Discussion:
@gabesaba: /assign@k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle rotten

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@tenzen-y: /remove-lifecycle rotten@gabesaba: /unassign@nasedil: /assign

Solution patch:
diff --git a/charts/kueue/templates/rbac/role.yaml b/charts/kueue/templates/rbac/role.yaml
index 62d4c6705..f860c94c4 100644
--- a/charts/kueue/templates/rbac/role.yaml
+++ b/charts/kueue/templates/rbac/role.yaml
@@ -199,6 +199,7 @@ rules:
     resources:
       - admissionchecks
       - clusterqueues
+      - cohorts
       - localqueues
       - workloads
     verbs:
diff --git a/config/components/manager/controller_manager_config.yaml b/config/components/manager/controller_manager_config.yaml
index c0a9f5973..866d3939d 100644
--- a/config/components/manager/controller_manager_config.yaml
+++ b/config/components/manager/controller_manager_config.yaml
@@ -16,6 +16,7 @@ controller:
     Pod: 5
     Workload.kueue.x-k8s.io: 5
     LocalQueue.kueue.x-k8s.io: 1
+    Cohort.kueue.x-k8s.io: 1
     ClusterQueue.kueue.x-k8s.io: 1
     ResourceFlavor.kueue.x-k8s.io: 1
 clientConnection:
diff --git a/config/components/rbac/role.yaml b/config/components/rbac/role.yaml
index 34a54abe5..d30ccd531 100644
--- a/config/components/rbac/role.yaml
+++ b/config/components/rbac/role.yaml
@@ -198,6 +198,7 @@ rules:
   resources:
   - admissionchecks
   - clusterqueues
+  - cohorts
   - localqueues
   - workloads
   verbs:
diff --git a/keps/79-hierarchical-cohorts/README.md b/keps/79-hierarchical-cohorts/README.md
index 3a935d155..3f69e3e88 100644
--- a/keps/79-hierarchical-cohorts/README.md
+++ b/keps/79-hierarchical-cohorts/README.md
@@ -9,6 +9,7 @@
   - [User Stories (Optional)](#user-stories-optional)
     - [Story 1](#story-1)
     - [Story 2](#story-2)
+  - [Notes](#notes)
   - [Risks and Mitigations](#risks-and-mitigations)
 - [Design Details](#design-details)
   - [Test Plan](#test-plan)
@@ -91,6 +92,22 @@ borrow unused capacity from any of the organizations.
 With this proposal, the cohorts for organizations will set borrowingLimit to 0. Top level Cohort will 
 contain all of these Cohorts, plus the "special" ClusterQueue, with borrowingLimit set to infinity. 
 
+### Notes
+
+As Cohorts do not have finalizers, Cohort's deletion will be processed
+immediately. If this Cohort is referenced by any Cohorts or ClusterQueues, the
+deleted Cohort will still exist in Kueue, but will no longer provide any
+resources of its own nor have a parent Cohort.
+
+This immediate deletion could result the tree suddenly losing capacity, which,
+depending on configuration, may result in preemptions. Fixing this is
+not as simple as just adding finalizers - non-deletions (nodes changing parents
+or resources) may trigger this capacity loss as well.  We may consider
+preventing a node from making a change which would result invalid balances -
+perhaps by requiring cordoning or draining some part of the tree before the
+change is executed. This requires further consideration and will not be included
+in the initial implementation.
+
 ### Risks and Mitigations
 
 * Users may create a cycle in the Cohort hierarchy - Kueue will stop all new admissions within
@@ -246,4 +263,4 @@ and quotas it may be hard for users to keep them under control.
 
 ## Alternatives
 
-* https://github.com/kubernetes-sigs/kueue/pull/1093 - Hierarchical ClusterQueues. 
\ No newline at end of file
+* https://github.com/kubernetes-sigs/kueue/pull/1093 - Hierarchical ClusterQueues.
diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go
index 900aac632..ac8591d7a 100644
--- a/pkg/cache/cache.go
+++ b/pkg/cache/cache.go
@@ -398,7 +398,7 @@ func (c *Cache) DeleteClusterQueue(cq *kueue.ClusterQueue) {
 	metrics.ClearCacheMetrics(cq.Name)
 }
 
-func (c *Cache) AddCohort(apiCohort *kueuealpha.Cohort) {
+func (c *Cache) AddOrUpdateCohort(apiCohort *kueuealpha.Cohort) {
 	c.Lock()
 	defer c.Unlock()
 	c.hm.AddCohort(apiCohort.Name)
@@ -406,15 +406,15 @@ func (c *Cache) AddCohort(apiCohort *kueuealpha.Cohort) {
 	cohort.updateCohort(apiCohort)
 }
 
-func (c *Cache) DeleteCohort(apiCohort *kueuealpha.Cohort) {
+func (c *Cache) DeleteCohort(cohortName string) {
 	c.Lock()
 	defer c.Unlock()
-	c.hm.DeleteCohort(apiCohort.Name)
+	c.hm.DeleteCohort(cohortName)
 
 	// If the cohort still exists after deletion, it means
 	// that it has one or more children referencing it.
 	// We need to run update algorithm.
-	if cohort, ok := c.hm.Cohorts[apiCohort.Name]; ok {
+	if cohort, ok := c.hm.Cohorts[cohortName]; ok {
 		updateCohortResourceNode(cohort)
 	}
 }
diff --git a/pkg/controller/core/cohort_controller.go b/pkg/controller/core/cohort_controller.go
new file mode 100644
index 000000000..b86812bf3
--- /dev/null
+++ b/pkg/controller/core/cohort_controller.go
@@ -0,0 +1,110 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package core
+
+import (
+	"context"
+
+	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/api/equality"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/klog/v2"
+	"k8s.io/utils/ptr"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
+	"sigs.k8s.io/kueue/pkg/cache"
+	"sigs.k8s.io/kueue/pkg/queue"
+)
+
+// CohortReconciler is responsible for synchronizing the in-memory
+// representation of Cohorts in cache.Cache and queue.Manager with
+// Cohort Kubernetes objects.
+type CohortReconciler struct {
+	client   client.Client
+	log      logr.Logger
+	cache    *cache.Cache
+	qManager *queue.Manager
+}
+
+func NewCohortReconciler(client client.Client, cache *cache.Cache, qManager *queue.Manager) CohortReconciler {
+	return CohortReconciler{client, ctrl.Log.WithName("cohort-reconciler"), cache, qManager}
+}
+
+func (r *CohortReconciler) SetupWithManager(mgr ctrl.Manager, cfg *config.Configuration) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&kueue.Cohort{}).
+		WithOptions(controller.Options{NeedLeaderElection: ptr.To(false)}).
+		WithEventFilter(r).
+		Complete(WithLeadingManager(mgr, r, &kueue.Cohort{}, cfg))
+}
+
+func (r *CohortReconciler) Create(e event.CreateEvent) bool {
+	return true
+}
+
+func (r *CohortReconciler) Update(e event.UpdateEvent) bool {
+	oldCohort, ok := e.ObjectOld.(*kueue.Cohort)
+	if !ok {
+		return false
+	}
+	newCohort, ok := e.ObjectNew.(*kueue.Cohort)
+	if !ok {
+		return false
+	}
+	log := r.log.WithValues("cohort", klog.KObj(newCohort))
+	if equality.Semantic.DeepEqual(oldCohort.Spec.ResourceGroups, newCohort.Spec.ResourceGroups) &&
+		oldCohort.Spec.Parent == newCohort.Spec.Parent {
+		log.V(2).Info("Skip Cohort update event as Cohort unchanged")
+		return false
+	}
+	log.V(2).Info("Processing Cohort update event")
+	return true
+}
+
+func (r *CohortReconciler) Delete(e event.DeleteEvent) bool {
+	return true
+}
+
+func (r *CohortReconciler) Generic(e event.GenericEvent) bool {
+	return true
+}
+
+//+kubebuilder:rbac:groups=kueue.x-k8s.io,resources=cohorts,verbs=get;list;watch;create;update;patch;delete
+
+func (r *CohortReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := ctrl.LoggerFrom(ctx).WithValues("cohort", req.Name)
+	log.V(2).Info("Reconciling Cohort")
+	var cohort kueue.Cohort
+	if err := r.client.Get(ctx, req.NamespacedName, &cohort); err != nil {
+		if apierrors.IsNotFound(err) {
+			log.V(2).Info("Cohort is being deleted")
+			r.cache.DeleteCohort(req.NamespacedName.Name)
+			r.qManager.DeleteCohort(req.NamespacedName.Name)
+		}
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+
+	log.V(2).Info("Cohort is being created or updated", "resources", cohort.Spec.ResourceGroups)
+	r.cache.AddOrUpdateCohort(&cohort)
+	r.qManager.AddOrUpdateCohort(ctx, &cohort)
+	return ctrl.Result{}, nil
+}
diff --git a/pkg/controller/core/core.go b/pkg/controller/core/core.go
index cb24eab9d..cc7a48a7b 100644
--- a/pkg/controller/core/core.go
+++ b/pkg/controller/core/core.go
@@ -71,6 +71,11 @@ func SetupControllers(mgr ctrl.Manager, qManager *queue.Manager, cc *cache.Cache
 		return "ClusterQueue", err
 	}
 
+	cohortRec := NewCohortReconciler(mgr.GetClient(), cc, qManager)
+	if err := cohortRec.SetupWithManager(mgr, cfg); err != nil {
+		return "Cohort", err
+	}
+
 	if err := NewWorkloadReconciler(mgr.GetClient(), qManager, cc,
 		mgr.GetEventRecorderFor(constants.WorkloadControllerName),
 		WithWorkloadUpdateWatchers(qRec, cqRec),
diff --git a/pkg/queue/manager.go b/pkg/queue/manager.go
index d6b9d7ea7..d20317896 100644
--- a/pkg/queue/manager.go
+++ b/pkg/queue/manager.go
@@ -29,6 +29,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	config "sigs.k8s.io/kueue/apis/config/v1beta1"
+	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	utilindexer "sigs.k8s.io/kueue/pkg/controller/core/indexer"
 	"sigs.k8s.io/kueue/pkg/hierarchy"
@@ -109,6 +110,21 @@ func NewManager(client client.Client, checker StatusChecker, opts ...Option) *Ma
 	return m
 }
 
+func (m *Manager) AddOrUpdateCohort(ctx context.Context, cohort *kueuealpha.Cohort) {
+	m.Lock()
+	defer m.Unlock()
+	m.hm.AddCohort(cohort.Name)
+	if m.moveWorkloadsCohort(ctx, m.hm.Cohorts[cohort.Name]) {
+		m.Broadcast()
+	}
+}
+
+func (m *Manager) DeleteCohort(cohortName string) {
+	m.Lock()
+	defer m.Unlock()
+	m.hm.DeleteCohort(cohortName)
+}
+
 func (m *Manager) AddClusterQueue(ctx context.Context, cq *kueue.ClusterQueue) error {
 	m.Lock()
 	defer m.Unlock()
@@ -139,7 +155,7 @@ func (m *Manager) AddClusterQueue(ctx context.Context, cq *kueue.ClusterQueue) e
 		}
 	}
 
-	queued := m.queueAllInadmissibleWorkloadsInCohort(ctx, cqImpl)
+	queued := m.moveWorkloadsCQ(ctx, cqImpl)
 	m.reportPendingWorkloads(cq.Name, cqImpl)
 	if queued || addedWorkloads {
 		m.Broadcast()
@@ -164,7 +180,7 @@ func (m *Manager) UpdateClusterQueue(ctx context.Context, cq *kueue.ClusterQueue
 
 	// TODO(#8): Selectively move workloads based on the exact event.
 	// If any workload becomes admissible or the queue becomes active.
-	if (specUpdated && m.queueAllInadmissibleWorkloadsInCohort(ctx, cqImpl)) || (!oldActive && cqImpl.Active()) {
+	if (specUpdated && m.moveWorkloadsCQ(ctx, cqImpl)) || (!oldActive && cqImpl.Active()) {
 		m.reportPendingWorkloads(cq.Name, cqImpl)
 		m.Broadcast()
 	}
@@ -394,7 +410,7 @@ func (m *Manager) QueueAssociatedInadmissibleWorkloadsAfter(ctx context.Context,
 		return
 	}
 
-	if m.queueAllInadmissibleWorkloadsInCohort(ctx, cq) {
+	if m.moveWorkloadsCQ(ctx, cq) {
 		m.Broadcast()
 	}
 }
@@ -415,7 +431,7 @@ func (m *Manager) QueueInadmissibleWorkloads(ctx context.Context, cqNames sets.S
 		if !exists {
 			continue
 		}
-		if m.queueAllInadmissibleWorkloadsInCohort(ctx, cq) {
+		if m.moveWorkloadsCQ(ctx, cq) {
 			queued = true
 		}
 	}
@@ -425,23 +441,27 @@ func (m *Manager) QueueInadmissibleWorkloads(ctx context.Context, cqNames sets.S
 	}
 }
 
-// queueAllInadmissibleWorkloadsInCohort moves all workloads in the same
+// moveWorkloadsCQ moves all workloads in the same
 // cohort with this ClusterQueue from inadmissibleWorkloads to heap. If the
 // cohort of this ClusterQueue is empty, it just moves all workloads in this
 // ClusterQueue. If at least one workload is moved, returns true, otherwise
 // returns false.
 // The events listed below could make workloads in the same cohort admissible.
-// Then queueAllInadmissibleWorkloadsInCohort need to be invoked.
+// Then moveWorkloadsCQ need to be invoked.
 // 1. delete events for any admitted workload in the cohort.
 // 2. add events of any cluster queue in the cohort.
 // 3. update events of any cluster queue in the cohort.
-func (m *Manager) queueAllInadmissibleWorkloadsInCohort(ctx context.Context, cq *ClusterQueue) bool {
-	if !cq.HasParent() {
-		return cq.QueueInadmissibleWorkloads(ctx, m.client)
+// 4. update of cohort.
+func (m *Manager) moveWorkloadsCQ(ctx context.Context, cq *ClusterQueue) bool {
+	if cq.HasParent() {
+		return m.moveWorkloadsCohort(ctx, cq.Parent())
 	}
+	return cq.QueueInadmissibleWorkloads(ctx, m.client)
+}
 
+func (m *Manager) moveWorkloadsCohort(ctx context.Context, cohort *cohort) bool {
 	queued := false
-	for _, clusterQueue := range cq.Parent().ChildCQs() {
+	for _, clusterQueue := range cohort.ChildCQs() {
 		queued = clusterQueue.QueueInadmissibleWorkloads(ctx, m.client) || queued
 	}
 	return queued



Test patch:
diff --git a/pkg/cache/cache_test.go b/pkg/cache/cache_test.go
index b7ee6c212..5e1435394 100644
--- a/pkg/cache/cache_test.go
+++ b/pkg/cache/cache_test.go
@@ -1055,7 +1055,7 @@ func TestCacheClusterQueueOperations(t *testing.T) {
 			name: "create cohort",
 			operation: func(cache *Cache) error {
 				cohort := utiltesting.MakeCohort("cohort").Obj()
-				cache.AddCohort(cohort)
+				cache.AddOrUpdateCohort(cohort)
 				return nil
 			},
 			wantCohorts: map[string]sets.Set[string]{
@@ -1066,8 +1066,8 @@ func TestCacheClusterQueueOperations(t *testing.T) {
 			name: "create and delete cohort",
 			operation: func(cache *Cache) error {
 				cohort := utiltesting.MakeCohort("cohort").Obj()
-				cache.AddCohort(cohort)
-				cache.DeleteCohort(cohort)
+				cache.AddOrUpdateCohort(cohort)
+				cache.DeleteCohort("cohort")
 				return nil
 			},
 			wantCohorts: nil,
@@ -1076,11 +1076,11 @@ func TestCacheClusterQueueOperations(t *testing.T) {
 			name: "cohort remains after deletion when child exists",
 			operation: func(cache *Cache) error {
 				cohort := utiltesting.MakeCohort("cohort").Obj()
-				cache.AddCohort(cohort)
+				cache.AddOrUpdateCohort(cohort)
 
 				_ = cache.AddClusterQueue(context.Background(),
 					utiltesting.MakeClusterQueue("cq").Cohort("cohort").Obj())
-				cache.DeleteCohort(cohort)
+				cache.DeleteCohort("cohort")
 				return nil
 			},
 			wantClusterQueues: map[string]*clusterQueue{
diff --git a/pkg/cache/resource_test.go b/pkg/cache/resource_test.go
index 0c7035d18..e609df09e 100644
--- a/pkg/cache/resource_test.go
+++ b/pkg/cache/resource_test.go
@@ -215,7 +215,7 @@ func TestAvailable(t *testing.T) {
 				_ = cache.AddClusterQueue(ctx, &cq)
 			}
 			for _, cohort := range tc.cohorts {
-				cache.AddCohort(&cohort)
+				cache.AddOrUpdateCohort(&cohort)
 			}
 
 			snapshot := cache.Snapshot()
diff --git a/pkg/cache/snapshot_test.go b/pkg/cache/snapshot_test.go
index b60e6bc7b..422fc43a5 100644
--- a/pkg/cache/snapshot_test.go
+++ b/pkg/cache/snapshot_test.go
@@ -813,7 +813,7 @@ func TestSnapshot(t *testing.T) {
 				}
 			}
 			for _, cohort := range tc.cohorts {
-				cache.AddCohort(cohort)
+				cache.AddOrUpdateCohort(cohort)
 			}
 			for _, rf := range tc.rfs {
 				cache.AddOrUpdateResourceFlavor(rf)
diff --git a/pkg/controller/core/cohort_controller_test.go b/pkg/controller/core/cohort_controller_test.go
new file mode 100644
index 000000000..cfa3eb876
--- /dev/null
+++ b/pkg/controller/core/cohort_controller_test.go
@@ -0,0 +1,283 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package core
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	"sigs.k8s.io/kueue/pkg/cache"
+	"sigs.k8s.io/kueue/pkg/queue"
+	"sigs.k8s.io/kueue/pkg/resources"
+	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+)
+
+func TestCohortReconcileCohortNotFoundDelete(t *testing.T) {
+	cl := utiltesting.NewClientBuilder().Build()
+	ctx := context.Background()
+	cache := cache.New(cl)
+	qManager := queue.NewManager(cl, cache)
+	reconciler := NewCohortReconciler(cl, cache, qManager)
+
+	cohort := utiltesting.MakeCohort("cohort").Obj()
+	cache.AddOrUpdateCohort(cohort)
+	qManager.AddOrUpdateCohort(ctx, cohort)
+	if _, ok := cache.Snapshot().Cohorts["cohort"]; !ok {
+		t.Fatal("expected Cohort in snapshot")
+	}
+
+	if _, err := reconciler.Reconcile(
+		ctx,
+		reconcile.Request{NamespacedName: client.ObjectKeyFromObject(cohort)},
+	); err != nil {
+		t.Fatal("unexpected error")
+	}
+
+	if _, ok := cache.Snapshot().Cohorts["cohort"]; ok {
+		t.Fatal("unexpected Cohort in snapshot")
+	}
+}
+
+func TestCohortReconcileCohortNotFoundIdempotentDelete(t *testing.T) {
+	cl := utiltesting.NewClientBuilder().
+		Build()
+	ctx := context.Background()
+	cache := cache.New(cl)
+	qManager := queue.NewManager(cl, cache)
+	reconciler := NewCohortReconciler(cl, cache, qManager)
+
+	if _, ok := cache.Snapshot().Cohorts["cohort"]; ok {
+		t.Fatal("unexpected Cohort in snapshot")
+	}
+
+	cohort := utiltesting.MakeCohort("cohort").Obj()
+	if _, err := reconciler.Reconcile(
+		ctx,
+		reconcile.Request{NamespacedName: client.ObjectKeyFromObject(cohort)},
+	); err != nil {
+		t.Fatal("unexpected error")
+	}
+
+	if _, ok := cache.Snapshot().Cohorts["cohort"]; ok {
+		t.Fatal("unexpected Cohort in snapshot")
+	}
+}
+
+func TestCohortReconcileErrorOtherThanNotFoundNotDeleted(t *testing.T) {
+	ctx := context.Background()
+	funcs := interceptor.Funcs{
+		Get: func(ctx context.Context, client client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+			return errors.New("error")
+		},
+	}
+	cl := utiltesting.NewClientBuilder().WithInterceptorFuncs(funcs).Build()
+
+	cache := cache.New(cl)
+	qManager := queue.NewManager(cl, cache)
+	reconciler := NewCohortReconciler(cl, cache, qManager)
+	cohort := utiltesting.MakeCohort("cohort").Obj()
+	cache.AddOrUpdateCohort(cohort)
+	qManager.AddOrUpdateCohort(ctx, cohort)
+	if _, ok := cache.Snapshot().Cohorts["cohort"]; !ok {
+		t.Fatal("expected Cohort in snapshot")
+	}
+
+	if _, err := reconciler.Reconcile(
+		ctx,
+		reconcile.Request{NamespacedName: client.ObjectKeyFromObject(cohort)},
+	); err == nil {
+		t.Fatal("expected error")
+	}
+
+	if _, ok := cache.Snapshot().Cohorts["cohort"]; !ok {
+		t.Fatal("expected Cohort in snapshot")
+	}
+}
+
+func TestCohortReconcileLifecycle(t *testing.T) {
+	ctx := context.Background()
+	cohort := utiltesting.MakeCohort("cohort").ResourceGroup(
+		utiltesting.MakeFlavorQuotas("red").Resource("cpu", "10").FlavorQuotas,
+	).Obj()
+	cl := utiltesting.NewClientBuilder().WithObjects(cohort).Build()
+	cache := cache.New(cl)
+	qManager := queue.NewManager(cl, cache)
+	reconciler := NewCohortReconciler(cl, cache, qManager)
+
+	// create
+	{
+		if _, err := reconciler.Reconcile(
+			ctx,
+			reconcile.Request{NamespacedName: client.ObjectKeyFromObject(cohort)},
+		); err != nil {
+			t.Fatal("unexpected error")
+		}
+
+		cohortSnap, ok := cache.Snapshot().Cohorts["cohort"]
+		if !ok {
+			t.Fatal("expected Cohort in snapshot")
+		}
+
+		wantQuotas := resources.FlavorResourceQuantities{
+			{Flavor: "red", Resource: "cpu"}: 10_000,
+		}
+		if diff := cmp.Diff(wantQuotas, cohortSnap.ResourceNode.SubtreeQuota); diff != "" {
+			t.Fatalf("unexpected quota (-want +got) %s", diff)
+		}
+	}
+
+	// update
+	{
+		if err := cl.Get(ctx, client.ObjectKeyFromObject(cohort), cohort); err != nil {
+			t.Fatal("unexpected error")
+		}
+		cohort.Spec.ResourceGroups[0] = utiltesting.ResourceGroup(
+			utiltesting.MakeFlavorQuotas("red").Resource("cpu", "5").FlavorQuotas,
+		)
+		if err := cl.Update(ctx, cohort); err != nil {
+			t.Fatal("unexpected error updating cohort", err)
+		}
+		if _, err := reconciler.Reconcile(
+			ctx,
+			reconcile.Request{NamespacedName: client.ObjectKeyFromObject(cohort)},
+		); err != nil {
+			t.Fatal("unexpected error")
+		}
+
+		cohortSnap, ok := cache.Snapshot().Cohorts["cohort"]
+		if !ok {
+			t.Fatal("expected Cohort in snapshot")
+		}
+
+		wantQuotas := resources.FlavorResourceQuantities{
+			{Flavor: "red", Resource: "cpu"}: 5_000,
+		}
+		if diff := cmp.Diff(wantQuotas, cohortSnap.ResourceNode.SubtreeQuota); diff != "" {
+			t.Fatalf("unexpected quota (-want +got) %s", diff)
+		}
+	}
+
+	// delete
+	{
+		if err := cl.Delete(ctx, cohort); err != nil {
+			t.Fatal("unexpected error during deletion")
+		}
+		if _, err := reconciler.Reconcile(
+			ctx,
+			reconcile.Request{NamespacedName: client.ObjectKeyFromObject(cohort)},
+		); err != nil {
+			t.Fatal("unexpected error")
+		}
+
+		if _, ok := cache.Snapshot().Cohorts["cohort"]; ok {
+			t.Fatal("unexpected Cohort in snapshot")
+		}
+	}
+}
+
+func TestCohortReconcilerFilters(t *testing.T) {
+	cl := utiltesting.NewClientBuilder().
+		Build()
+	cache := cache.New(cl)
+	qManager := queue.NewManager(cl, cache)
+	reconciler := NewCohortReconciler(cl, cache, qManager)
+
+	t.Run("delete returns true", func(t *testing.T) {
+		if !reconciler.Delete(event.DeleteEvent{}) {
+			t.Fatal("expected delete to return true")
+		}
+	})
+
+	t.Run("create returns true", func(t *testing.T) {
+		if !reconciler.Create(event.CreateEvent{}) {
+			t.Fatal("expected create to return true")
+		}
+	})
+
+	t.Run("generic returns true", func(t *testing.T) {
+		if !reconciler.Generic(event.GenericEvent{}) {
+			t.Fatal("expected generic to return true")
+		}
+	})
+
+	cases := map[string]struct {
+		old  client.Object
+		new  client.Object
+		want bool
+	}{
+		"old wrong type returns false": {
+			old:  utiltesting.MakeClusterQueue("cq").Obj(),
+			new:  utiltesting.MakeCohort("cohort").Obj(),
+			want: false,
+		},
+		"new wrong type returns false": {
+			old:  utiltesting.MakeCohort("cohort").Obj(),
+			new:  utiltesting.MakeClusterQueue("cq").Obj(),
+			want: false,
+		},
+		"unchanged returns false": {
+			old: utiltesting.MakeCohort("cohort").ResourceGroup(
+				utiltesting.MakeFlavorQuotas("red").Resource("cpu", "5").FlavorQuotas,
+			).Obj(),
+			new: utiltesting.MakeCohort("cohort").ResourceGroup(
+				utiltesting.MakeFlavorQuotas("red").Resource("cpu", "5").FlavorQuotas,
+			).Obj(),
+			want: false,
+		},
+		"changed resource returns true": {
+			old: utiltesting.MakeCohort("cohort").ResourceGroup(
+				utiltesting.MakeFlavorQuotas("red").Resource("cpu", "5").FlavorQuotas,
+			).Obj(),
+			new: utiltesting.MakeCohort("cohort").ResourceGroup(
+				utiltesting.MakeFlavorQuotas("red").Resource("cpu", "10").FlavorQuotas,
+			).Obj(),
+			want: true,
+		},
+		"adding parent returns true": {
+			old:  utiltesting.MakeCohort("cohort").Obj(),
+			new:  utiltesting.MakeCohort("cohort").Parent("parent").Obj(),
+			want: true,
+		},
+		"changing parent returns true": {
+			old:  utiltesting.MakeCohort("cohort").Parent("old").Obj(),
+			new:  utiltesting.MakeCohort("cohort").Parent("new").Obj(),
+			want: true,
+		},
+		"deleting parent returns true": {
+			old:  utiltesting.MakeCohort("cohort").Parent("parent").Obj(),
+			new:  utiltesting.MakeCohort("cohort").Obj(),
+			want: true,
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			event := event.UpdateEvent{
+				ObjectOld: tc.old,
+				ObjectNew: tc.new,
+			}
+			if reconciler.Update(event) != tc.want {
+				t.Fatalf("expected %v, got %v", tc.want, !tc.want)
+			}
+		})
+	}
+}
diff --git a/pkg/util/testing/client.go b/pkg/util/testing/client.go
index 1f30355b7..df84cd1d1 100644
--- a/pkg/util/testing/client.go
+++ b/pkg/util/testing/client.go
@@ -31,6 +31,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 
+	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
 )
@@ -47,6 +48,7 @@ func NewClientBuilder(addToSchemes ...func(s *runtime.Scheme) error) *fake.Clien
 	scheme := runtime.NewScheme()
 	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
 	utilruntime.Must(kueue.AddToScheme(scheme))
+	utilruntime.Must(kueuealpha.AddToScheme(scheme))
 	for i := range addToSchemes {
 		utilruntime.Must(addToSchemes[i](scheme))
 	}
diff --git a/pkg/util/testing/wrappers.go b/pkg/util/testing/wrappers.go
index 6213b3897..f7c355573 100644
--- a/pkg/util/testing/wrappers.go
+++ b/pkg/util/testing/wrappers.go
@@ -607,7 +607,7 @@ func (c *CohortWrapper) Parent(parentName string) *CohortWrapper {
 
 // ResourceGroup adds a ResourceGroup with flavors.
 func (c *CohortWrapper) ResourceGroup(flavors ...kueue.FlavorQuotas) *CohortWrapper {
-	c.Spec.ResourceGroups = append(c.Spec.ResourceGroups, createResourceGroup(flavors...))
+	c.Spec.ResourceGroups = append(c.Spec.ResourceGroups, ResourceGroup(flavors...))
 	return c
 }
 
@@ -651,7 +651,8 @@ func (c *ClusterQueueWrapper) AdmissionCheckStrategy(acs ...kueue.AdmissionCheck
 	return c
 }
 
-func createResourceGroup(flavors ...kueue.FlavorQuotas) kueue.ResourceGroup {
+// ResourceGroup creates a ResourceGroup with the given FlavorQuotas.
+func ResourceGroup(flavors ...kueue.FlavorQuotas) kueue.ResourceGroup {
 	rg := kueue.ResourceGroup{
 		Flavors: flavors,
 	}
@@ -677,7 +678,7 @@ func createResourceGroup(flavors ...kueue.FlavorQuotas) kueue.ResourceGroup {
 
 // ResourceGroup adds a ResourceGroup with flavors.
 func (c *ClusterQueueWrapper) ResourceGroup(flavors ...kueue.FlavorQuotas) *ClusterQueueWrapper {
-	c.Spec.ResourceGroups = append(c.Spec.ResourceGroups, createResourceGroup(flavors...))
+	c.Spec.ResourceGroups = append(c.Spec.ResourceGroups, ResourceGroup(flavors...))
 	return c
 }
 
diff --git a/test/integration/scheduler/scheduler_test.go b/test/integration/scheduler/scheduler_test.go
index e9e7fbf23..7c43bbc2c 100644
--- a/test/integration/scheduler/scheduler_test.go
+++ b/test/integration/scheduler/scheduler_test.go
@@ -29,6 +29,7 @@ import (
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/core"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -1407,6 +1408,98 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		})
 	})
 
+	ginkgo.When("Cohort provides resources directly", func() {
+		var (
+			cq     *kueue.ClusterQueue
+			cohort *kueuealpha.Cohort
+			wl     *kueue.Workload
+		)
+
+		ginkgo.BeforeEach(func() {
+			gomega.Expect(k8sClient.Create(ctx, onDemandFlavor)).Should(gomega.Succeed())
+		})
+
+		ginkgo.AfterEach(func() {
+			gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, cq, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, cohort, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, wl, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, onDemandFlavor, true)
+		})
+
+		ginkgo.It("Should admit workload using resources borrowed from cohort", func() {
+			cq = testing.MakeClusterQueue("clusterqueue").
+				Cohort("cohort").
+				ResourceGroup(
+					*testing.MakeFlavorQuotas("on-demand").Resource(corev1.ResourceCPU, "0").Obj(),
+				).Obj()
+			queue := testing.MakeLocalQueue("queue", ns.Name).ClusterQueue(cq.Name).Obj()
+			gomega.Expect(k8sClient.Create(ctx, cq)).Should(gomega.Succeed())
+			gomega.Expect(k8sClient.Create(ctx, queue)).Should(gomega.Succeed())
+
+			ginkgo.By("workload not admitted when Cohort doesn't provide any resources")
+			wl = testing.MakeWorkload("wl", ns.Name).Queue(queue.Name).
+				Request(corev1.ResourceCPU, "10").Obj()
+			gomega.Expect(k8sClient.Create(ctx, wl)).Should(gomega.Succeed())
+			util.ExpectWorkloadsToBePending(ctx, k8sClient, wl)
+			util.ExpectAdmittedWorkloadsTotalMetric(cq, 0)
+
+			ginkgo.By("workload not admitted when Cohort provides too few resources")
+			cohort = testing.MakeCohort("cohort").
+				ResourceGroup(
+					*testing.MakeFlavorQuotas("on-demand").Resource(corev1.ResourceCPU, "5").Obj(),
+				).Obj()
+			gomega.Expect(k8sClient.Create(ctx, cohort)).Should(gomega.Succeed())
+			util.ExpectWorkloadsToBePending(ctx, k8sClient, wl)
+			util.ExpectAdmittedWorkloadsTotalMetric(cq, 0)
+
+			ginkgo.By("workload admitted when Cohort updated to provide sufficient resources")
+			gomega.Eventually(func() error {
+				_ = k8sClient.Get(ctx, types.NamespacedName{Name: cohort.Name}, cohort)
+				cohort.Spec.ResourceGroups[0].Flavors[0].Resources[0].NominalQuota = resource.MustParse("10")
+				return k8sClient.Update(ctx, cohort)
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			expectAdmission := testing.MakeAdmission(cq.Name).Assignment(corev1.ResourceCPU, "on-demand", "10").Obj()
+			util.ExpectWorkloadToBeAdmittedAs(ctx, k8sClient, wl, expectAdmission)
+			util.ExpectPendingWorkloadsMetric(cq, 0, 0)
+			util.ExpectAdmittedWorkloadsTotalMetric(cq, 1)
+		})
+
+		ginkgo.It("Should admit workload when cq switches into cohort with capacity", func() {
+			cq = testing.MakeClusterQueue("clusterqueue").
+				Cohort("impecunious-cohort").
+				ResourceGroup(
+					*testing.MakeFlavorQuotas("on-demand").Resource(corev1.ResourceCPU, "0").Obj(),
+				).Obj()
+			queue := testing.MakeLocalQueue("queue", ns.Name).ClusterQueue(cq.Name).Obj()
+			gomega.Expect(k8sClient.Create(ctx, cq)).Should(gomega.Succeed())
+			gomega.Expect(k8sClient.Create(ctx, queue)).Should(gomega.Succeed())
+
+			ginkgo.By("workload not admitted")
+			wl = testing.MakeWorkload("wl", ns.Name).Queue(queue.Name).
+				Request(corev1.ResourceCPU, "10").Obj()
+			gomega.Expect(k8sClient.Create(ctx, wl)).Should(gomega.Succeed())
+			util.ExpectWorkloadsToBePending(ctx, k8sClient, wl)
+			util.ExpectAdmittedWorkloadsTotalMetric(cq, 0)
+
+			ginkgo.By("cohort created")
+			cohort = testing.MakeCohort("cohort").
+				ResourceGroup(*testing.MakeFlavorQuotas("on-demand").Resource(corev1.ResourceCPU, "10").Obj()).Obj()
+			gomega.Expect(k8sClient.Create(ctx, cohort)).Should(gomega.Succeed())
+
+			ginkgo.By("cq switches and workload admitted")
+			gomega.Eventually(func() error {
+				_ = k8sClient.Get(ctx, types.NamespacedName{Name: cq.Name}, cq)
+				cq.Spec.Cohort = "cohort"
+				return k8sClient.Update(ctx, cq)
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			expectAdmission := testing.MakeAdmission(cq.Name).Assignment(corev1.ResourceCPU, "on-demand", "10").Obj()
+			util.ExpectWorkloadToBeAdmittedAs(ctx, k8sClient, wl, expectAdmission)
+			util.ExpectPendingWorkloadsMetric(cq, 0, 0)
+			util.ExpectAdmittedWorkloadsTotalMetric(cq, 1)
+		})
+	})
+
 	ginkgo.When("Using cohorts for sharing with LendingLimit", func() {
 		var (
 			prodCQ *kueue.ClusterQueue