Prevent admitting inactive workloads (#7913)

* Prevent admitting inactive workloads

* Adress lint finding & review comment

* Yet another linter fix

Author: olekzabl

pkg/cache/queue/manager.go

@@ -445,6 +445,12 @@ func (m *Manager) AddOrUpdateWorkload(w *kueue.Workload, opts ...workload.InfoOp
 }
 
 func (m *Manager) AddOrUpdateWorkloadWithoutLock(w *kueue.Workload, opts ...workload.InfoOption) error {
+	if !workload.IsActive(w) {
+		return fmt.Errorf("workload %q is inactive and can't be added to a LocalQueue", w.Name)
+	}
+	if workload.HasQuotaReservation(w) {
+		return fmt.Errorf("workload %q already has quota reserved and can't be added to a LocalQueue", w.Name)
+	}
 	qKey := queue.KeyFromWorkload(w)
 	q := m.localQueues[qKey]
 	if q == nil {

pkg/controller/core/workload_controller.go

@@ -1149,8 +1149,10 @@ func (h *resourceUpdatesHandler) queueReconcileForPending(ctx context.Context, q
 			continue
 		}
 
-		if err = h.r.queues.AddOrUpdateWorkload(wlCopy); err != nil {
-			log.V(2).Info("ignored an error for now", "error", err)
+		if workload.IsActive(wlCopy) && !workload.HasQuotaReservation(wlCopy) {
+			if err = h.r.queues.AddOrUpdateWorkload(wlCopy); err != nil {
+				log.V(2).Info("ignored an error for now", "error", err)
+			}
 		}
 	}
 }

pkg/util/testing/core.go

@@ -60,16 +60,23 @@ func CheckEventRecordedFor(ctx context.Context, k8sClient client.Client,
 		len(events.Items), eventReason, eventType, eventMessage, ref.Namespace)
 }
 
-// HasEventAppeared returns if an event has been emitted
-func HasEventAppeared(ctx context.Context, k8sClient client.Client, event corev1.Event) (bool, error) {
+// HasMatchingEventAppeared returns if an event has been emitted
+func HasMatchingEventAppeared(ctx context.Context, k8sClient client.Client, matcher func(*corev1.Event) bool) (bool, error) {
 	events := &corev1.EventList{}
 	if err := k8sClient.List(ctx, events, &client.ListOptions{}); err != nil {
 		return false, err
 	}
 	for _, item := range events.Items {
-		if item.Reason == event.Reason && item.Type == event.Type && item.Message == event.Message {
+		if matcher(&item) {
 			return true, nil
 		}
 	}
 	return false, nil
 }
+
+// HasEventAppeared returns if an event has been emitted
+func HasEventAppeared(ctx context.Context, k8sClient client.Client, event corev1.Event) (bool, error) {
+	return HasMatchingEventAppeared(ctx, k8sClient, func(item *corev1.Event) bool {
+		return item.Reason == event.Reason && item.Type == event.Type && item.Message == event.Message
+	})
+}

test/integration/singlecluster/controller/core/suite_test.go

@@ -29,8 +29,10 @@ import (
 	config "sigs.k8s.io/kueue/apis/config/v1beta2"
 	qcache "sigs.k8s.io/kueue/pkg/cache/queue"
 	schdcache "sigs.k8s.io/kueue/pkg/cache/scheduler"
+	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/core"
 	"sigs.k8s.io/kueue/pkg/controller/core/indexer"
+	"sigs.k8s.io/kueue/pkg/scheduler"
 	"sigs.k8s.io/kueue/pkg/webhooks"
 	"sigs.k8s.io/kueue/test/integration/framework"
 	"sigs.k8s.io/kueue/test/util"
@@ -63,12 +65,31 @@ var _ = ginkgo.AfterSuite(func() {
 	fwk.Teardown()
 })
 
+type managerSetupOpts struct {
+	runScheduler bool
+}
+
+type managerSetupOption func(*managerSetupOpts)
+
+func runScheduler(opts *managerSetupOpts) {
+	opts.runScheduler = true
+}
+
 func managerSetup(ctx context.Context, mgr manager.Manager) {
 	managerAndControllerSetup(nil)(ctx, mgr)
 }
 
-func managerAndControllerSetup(controllersCfg *config.Configuration) framework.ManagerSetup {
+func managerAndSchedulerSetup(ctx context.Context, mgr manager.Manager) {
+	managerAndControllerSetup(nil, runScheduler)(ctx, mgr)
+}
+
+func managerAndControllerSetup(controllersCfg *config.Configuration, options ...managerSetupOption) framework.ManagerSetup {
 	return func(ctx context.Context, mgr manager.Manager) {
+		var opts managerSetupOpts
+		for _, opt := range options {
+			opt(&opts)
+		}
+
 		err := indexer.Setup(ctx, mgr.GetFieldIndexer())
 		gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
@@ -88,5 +109,11 @@ func managerAndControllerSetup(controllersCfg *config.Configuration) framework.M
 
 		failedCtrl, err := core.SetupControllers(mgr, queues, cCache, controllersCfg)
 		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
+
+		if opts.runScheduler {
+			sched := scheduler.New(queues, cCache, mgr.GetClient(), mgr.GetEventRecorderFor(constants.AdmissionName))
+			err = sched.Start(ctx)
+			gomega.Expect(err).NotTo(gomega.HaveOccurred())
+		}
 	}
 }

test/integration/singlecluster/controller/core/workload_controller_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
+	nodev1 "k8s.io/api/node/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
@@ -612,6 +613,95 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
 	})
 })
 
+var _ = ginkgo.Describe("Workload controller interaction with scheduler", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+	var (
+		ns           *corev1.Namespace
+		clusterQueue *kueue.ClusterQueue
+		localQueue   *kueue.LocalQueue
+		wl           *kueue.Workload
+	)
+
+	ginkgo.BeforeAll(func() {
+		fwk.StartManager(ctx, cfg, managerAndSchedulerSetup)
+	})
+
+	ginkgo.AfterAll(func() {
+		fwk.StopManager(ctx)
+	})
+
+	ginkgo.When("workload's runtime class is changed", func() {
+		var flavor *kueue.ResourceFlavor
+		var runtimeClass *nodev1.RuntimeClass
+
+		const runtimeClassName = "test-kueue-class"
+
+		ginkgo.BeforeEach(func() {
+			ns = util.CreateNamespaceFromPrefixWithLog(ctx, k8sClient, "core-workload-")
+			flavor = utiltestingapi.MakeResourceFlavor(flavorOnDemand).Obj()
+			util.MustCreate(ctx, k8sClient, flavor)
+			clusterQueue = utiltestingapi.MakeClusterQueue("cluster-queue").
+				ResourceGroup(*utiltestingapi.MakeFlavorQuotas(flavorOnDemand).Resource(corev1.ResourceCPU, "5", "5").Obj()).
+				Cohort("cohort").
+				Obj()
+			util.MustCreate(ctx, k8sClient, clusterQueue)
+			localQueue = utiltestingapi.MakeLocalQueue("queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+			util.MustCreate(ctx, k8sClient, localQueue)
+			runtimeClass = utiltesting.MakeRuntimeClass(runtimeClassName, "rc-handler-1").Obj()
+			util.MustCreate(ctx, k8sClient, runtimeClass)
+		})
+
+		ginkgo.AfterEach(func() {
+			gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, flavor, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, runtimeClass, true)
+		})
+
+		ginkgo.It("should not temporarily admit an inactive workload", func() {
+			ginkgo.By("creating an inactive workload", func() {
+				wl = utiltestingapi.MakeWorkload("wl1", ns.Name).
+					Queue(kueue.LocalQueueName(localQueue.Name)).
+					Request(corev1.ResourceCPU, "1").
+					RuntimeClass(runtimeClassName).
+					Active(false).
+					Obj()
+				util.MustCreate(ctx, k8sClient, wl)
+
+				wlKey := client.ObjectKeyFromObject(wl)
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("changing the runtime class", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					runtimeClassKey := client.ObjectKeyFromObject(runtimeClass)
+					g.Expect(k8sClient.Get(ctx, runtimeClassKey, runtimeClass)).To(gomega.Succeed())
+					if runtimeClass.ObjectMeta.Annotations == nil {
+						runtimeClass.ObjectMeta.Annotations = map[string]string{}
+					}
+					runtimeClass.ObjectMeta.Annotations["foo"] = "bar"
+					g.Expect(k8sClient.Update(ctx, runtimeClass)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("checking no 'quota reseved' event appearing for the workload", func() {
+				gomega.Consistently(func(g gomega.Gomega) {
+					found, err := utiltesting.HasMatchingEventAppeared(ctx, k8sClient, func(e *corev1.Event) bool {
+						return e.Reason == "QuotaReserved" &&
+							e.Type == corev1.EventTypeNormal &&
+							e.InvolvedObject.Kind == "Workload" &&
+							e.InvolvedObject.Name == wl.Name &&
+							e.InvolvedObject.Namespace == wl.Namespace
+					})
+					g.Expect(err).NotTo(gomega.HaveOccurred())
+					g.Expect(found).To(gomega.BeTrue())
+				}, util.ShortTimeout, util.Interval).ShouldNot(gomega.Succeed())
+			})
+		})
+	})
+})
+
 var _ = ginkgo.Describe("Workload controller with resource retention", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
 	ginkgo.When("manager is setup with tiny retention period", func() {
 		var (