Repository: kubernetes-sigs/kueue

Issue:
TAS: Filter Node Reconcile events
The node reconciler used in TAS does not filter events. QueueInadmissibleWorkloads is called each time a reconciliation occurs.
https://github.com/kubernetes-sigs/kueue/blob/3ff67d04883d1e091400acfba120a02a348d3e66/pkg/controller/tas/resource_flavor.go#L151-L157

As a result, in large clusters, InadmissibleWorkloads may return to the heap before scanning the entire heap.

How about adding a filter to node events to avoid this issue? The kube-scheduler implementation may be helpful.
https://github.com/kubernetes/kubernetes/blob/ee065dfc80b5762d53c5d9d896a8cc4b1d9c6bd2/pkg/scheduler/framework/events.go#L161-L169

// I wasn't sure which issue template would be appropriate, so I created one from blank. 

Discussion:
@mimowo: To mitigate the "too often reconciles" we implement [batching by 1s](https://github.com/kubernetes-sigs/kueue/blob/3ff67d04883d1e091400acfba120a02a348d3e66/pkg/controller/tas/resource_flavor.go#L133). 

Still, this sounds like a useful optimization. I think we can use the same list of Node properties: nodeChangeExtracters as in kube-scheduler (node labels which are primarily used by TAS are already included).

@Ladicle: Let me add something about this optimization (@utam0k and I work in the same team). 
We've tested this behavior in our environment, and most of this reconciliation is triggered by the node's `LastHeartbeatTime` condition updates.  We use the kubelet default `node-status-update-frequency`,  so at least one unnecessary reconciliation will run every 10 seconds.
@tenzen-y: This seems to be a common starvation problem due to inadmissible workload updating regardless of TAS.
We have mitigation ways to resolve that `requeuingStrategy`. Did you optimize those (https://kueue.sigs.k8s.io/docs/tasks/manage/setup_wait_for_pods_ready/#enabling-waitforpodsready)? Especially `requeuingStrategy.timestamp`@mimowo: > We've tested this behavior in our environment, and most of this reconciliation is triggered by the node's LastHeartbeatTime condition updates. 

+1, these would be helpful to skip reconciles triggered just by heartbeat.@tenzen-y: > > We've tested this behavior in our environment, and most of this reconciliation is triggered by the node's LastHeartbeatTime condition updates.
> 
> +1, these would be helpful to skip reconciles triggered just by heartbeat.

SGTM as well.@utam0k: /assign utam0k ladicle@tenzen-y: To clarify, I would like to start introducing a predicate or an event handler to filter out only `LastHeartbeatTime` updates for now.@tenzen-y: /close 
by #6570 @k8s-ci-robot: @tenzen-y: Closing this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/6551#issuecomment-3211855328):

>/close 
>by #6570 


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>

Solution patch:
diff --git a/pkg/controller/tas/resource_flavor.go b/pkg/controller/tas/resource_flavor.go
index 101894495..0fe7c4f7d 100644
--- a/pkg/controller/tas/resource_flavor.go
+++ b/pkg/controller/tas/resource_flavor.go
@@ -21,6 +21,10 @@ import (
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/conversion"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
@@ -43,6 +47,24 @@ import (
 	"sigs.k8s.io/kueue/pkg/queue"
 )
 
+var nodeSemantic = conversion.EqualitiesOrDie(
+	nodeConditionEqual,
+	// Handle metav1.Time comparison to avoid panic on unexported fields
+	func(a, b metav1.Time) bool {
+		return a.Equal(&b)
+	},
+	// Handle resource.Quantity comparison to avoid panic on unexported fields
+	func(a, b resource.Quantity) bool {
+		return a.Equal(b)
+	},
+)
+
+func nodeConditionEqual(a, b corev1.NodeCondition) bool {
+	aCopy, bCopy := a.DeepCopy(), b.DeepCopy()
+	aCopy.LastHeartbeatTime, bCopy.LastHeartbeatTime = metav1.Time{}, metav1.Time{}
+	return equality.Semantic.DeepEqual(aCopy, bCopy)
+}
+
 type rfReconciler struct {
 	log      logr.Logger
 	queues   *queue.Manager
@@ -109,6 +131,12 @@ func (h *nodeHandler) Update(ctx context.Context, e event.UpdateEvent, q workque
 	if !isOldNode || !isNewNode {
 		return
 	}
+
+	if !checkNodeSchedulingPropertiesChanged(oldNode, newNode) {
+		ctrl.LoggerFrom(ctx).V(5).Info("Skipping node update as new Node is semantically same as old Node", "node", newNode.Name)
+		return
+	}
+
 	h.queueReconcileForNode(oldNode, q)
 	h.queueReconcileForNode(newNode, q)
 }
@@ -203,3 +231,8 @@ func nodeBelongsToFlavor(node *corev1.Node, nodeLabels map[string]string, levels
 	}
 	return true
 }
+
+// checkNodeSchedulingPropertiesChanged checks if the node update affects TAS scheduling.
+func checkNodeSchedulingPropertiesChanged(oldNode, newNode *corev1.Node) bool {
+	return !nodeSemantic.DeepEqual(oldNode, newNode)
+}
diff --git a/site/content/en/docs/concepts/cohort.md b/site/content/en/docs/concepts/cohort.md
index 302755ba4..cf1b0f2e2 100644
--- a/site/content/en/docs/concepts/cohort.md
+++ b/site/content/en/docs/concepts/cohort.md
@@ -29,7 +29,7 @@ spec:
 ## Configuring Quotas
 
 Resource quotas may be defined at the Cohort level (similarly to how they are
-defined for (ClusterQueues)[/docs/concepts/cluster_queue/#flavors-and-resources]),
+defined for [ClusterQueues](/docs/concepts/cluster_queue/#flavors-and-resources)),
 and consumed by ClusterQueues within the Cohort.  Please note that
 `nominalQuota` defined at the Cohort level represents **additional resources**
 on top of those defined by ClusterQueues within the Cohort. The Cohort's



Test patch:
diff --git a/pkg/controller/tas/resource_flavor_test.go b/pkg/controller/tas/resource_flavor_test.go
new file mode 100644
index 000000000..0d2b43610
--- /dev/null
+++ b/pkg/controller/tas/resource_flavor_test.go
@@ -0,0 +1,252 @@
+/*
+Copyright 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 tas
+
+import (
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	testingnode "sigs.k8s.io/kueue/pkg/util/testingjobs/node"
+)
+
+func TestNodeHandler_Update(t *testing.T) {
+	now := metav1.Now()
+	later := metav1.NewTime(now.Add(10 * time.Second))
+
+	baseNode := testingnode.MakeNode("test-node").
+		Annotation("test-annotation", "value").
+		Label("topology.kubernetes.io/zone", "zone-a").
+		Label("node-role", "worker").
+		Taints(corev1.Taint{
+			Key:    "test-taint",
+			Value:  "value",
+			Effect: corev1.TaintEffectNoSchedule,
+		}).
+		StatusAllocatable(corev1.ResourceList{
+			corev1.ResourceCPU:    resource.MustParse("8"),
+			corev1.ResourceMemory: resource.MustParse("32Gi"),
+		})
+
+	testCases := map[string]struct {
+		oldNode     *corev1.Node
+		newNode     *corev1.Node
+		wantChanged bool
+	}{
+		"LastHeartbeatTime changed": {
+			oldNode: baseNode.Clone().
+				StatusConditions(
+					corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionTrue,
+						LastHeartbeatTime:  now,
+						LastTransitionTime: now,
+					},
+					corev1.NodeCondition{
+						Type:               corev1.NodeMemoryPressure,
+						Status:             corev1.ConditionFalse,
+						LastHeartbeatTime:  now,
+						LastTransitionTime: now,
+					},
+				).Obj(),
+			newNode: baseNode.Clone().
+				StatusConditions(
+					corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionTrue,
+						LastHeartbeatTime:  later,
+						LastTransitionTime: now,
+					},
+					corev1.NodeCondition{
+						Type:               corev1.NodeMemoryPressure,
+						Status:             corev1.ConditionFalse,
+						LastHeartbeatTime:  later,
+						LastTransitionTime: now,
+					},
+				).Obj(),
+			wantChanged: false,
+		},
+		"Annotation changed": {
+			oldNode:     baseNode.Clone().Obj(),
+			newNode:     baseNode.Clone().Annotation("new-annotation", "new-value").Obj(),
+			wantChanged: true,
+		},
+		"Label changed": {
+			oldNode:     baseNode.Clone().Obj(),
+			newNode:     baseNode.Clone().Label("new-label", "new-value").Obj(),
+			wantChanged: true,
+		},
+		"Node Ready status changed": {
+			oldNode: baseNode.Clone().
+				StatusConditions(
+					corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionTrue,
+						LastHeartbeatTime:  now,
+						LastTransitionTime: now,
+					},
+				).Obj(),
+			newNode: baseNode.Clone().
+				StatusConditions(
+					corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionFalse,
+						LastHeartbeatTime:  later,
+						LastTransitionTime: later,
+					},
+				).Obj(),
+			wantChanged: true,
+		},
+		"Allocatable resources changed": {
+			oldNode: baseNode.Clone().Obj(),
+			newNode: baseNode.Clone().StatusAllocatable(corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("16"),
+				corev1.ResourceMemory: resource.MustParse("32Gi"),
+			}).Obj(),
+			wantChanged: true,
+		},
+		"Taints changed": {
+			oldNode: baseNode.Clone().Obj(),
+			newNode: baseNode.Clone().Taints(corev1.Taint{
+				Key:    "new-taint",
+				Value:  "new-value",
+				Effect: corev1.TaintEffectNoExecute,
+			}).Obj(),
+			wantChanged: true,
+		},
+		"Taints with TimeAdded": {
+			oldNode: baseNode.Clone().Taints(corev1.Taint{
+				Key:       "test",
+				Value:     "value",
+				Effect:    corev1.TaintEffectNoExecute,
+				TimeAdded: &now,
+			}).Obj(),
+			newNode: baseNode.Clone().Taints(corev1.Taint{
+				Key:       "test",
+				Value:     "value",
+				Effect:    corev1.TaintEffectNoExecute,
+				TimeAdded: &later,
+			}).Obj(),
+			wantChanged: true,
+		},
+		"Taints TimeAdded from null to non-null": {
+			oldNode: baseNode.Clone().Taints(corev1.Taint{
+				Key:       "test",
+				Value:     "value",
+				Effect:    corev1.TaintEffectNoExecute,
+				TimeAdded: nil,
+			}).Obj(),
+			newNode: baseNode.Clone().Taints(corev1.Taint{
+				Key:       "test",
+				Value:     "value",
+				Effect:    corev1.TaintEffectNoExecute,
+				TimeAdded: &later,
+			}).Obj(),
+			wantChanged: true,
+		},
+		"Unschedulable changed": {
+			oldNode:     baseNode.Clone().Obj(),
+			newNode:     baseNode.Clone().Unschedulable().Obj(),
+			wantChanged: true,
+		},
+		"Update Multiple properties": {
+			oldNode: baseNode.Clone().
+				StatusConditions(
+					corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionTrue,
+						LastHeartbeatTime:  now,
+						LastTransitionTime: now,
+					},
+					corev1.NodeCondition{
+						Type:               corev1.NodeMemoryPressure,
+						Status:             corev1.ConditionFalse,
+						LastHeartbeatTime:  now,
+						LastTransitionTime: now,
+					},
+				).Obj(),
+			newNode: baseNode.Clone().
+				StatusConditions(
+					corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionTrue,
+						LastHeartbeatTime:  later,
+						LastTransitionTime: now,
+					},
+					corev1.NodeCondition{
+						Type:               corev1.NodeMemoryPressure,
+						Status:             corev1.ConditionFalse,
+						LastHeartbeatTime:  later,
+						LastTransitionTime: now,
+					},
+				).
+				Annotation("another-annotation", "another-value").
+				ResourceVersion("12345").
+				Obj(),
+			wantChanged: true,
+		},
+		"New condition type added": {
+			oldNode: baseNode.Clone().Obj(),
+			newNode: baseNode.Clone().StatusConditions(corev1.NodeCondition{
+				Type:               corev1.NodeDiskPressure,
+				Status:             corev1.ConditionTrue,
+				LastHeartbeatTime:  now,
+				LastTransitionTime: now,
+			}).Obj(),
+			wantChanged: true,
+		},
+		"Condition removed": {
+			oldNode: baseNode.Clone().
+				StatusConditions(
+					corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionTrue,
+						LastHeartbeatTime:  now,
+						LastTransitionTime: now,
+					},
+					corev1.NodeCondition{
+						Type:               corev1.NodeDiskPressure,
+						Status:             corev1.ConditionFalse,
+						LastHeartbeatTime:  now,
+						LastTransitionTime: now,
+					},
+				).Obj(),
+			newNode: baseNode.Clone().
+				StatusConditions(
+					corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionTrue,
+						LastHeartbeatTime:  now,
+						LastTransitionTime: now,
+					},
+				).Obj(),
+			wantChanged: true,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			got := checkNodeSchedulingPropertiesChanged(tc.oldNode, tc.newNode)
+			if got != tc.wantChanged {
+				t.Errorf("nodeSchedulingPropertiesChanged() = %v, want %v", got, tc.wantChanged)
+			}
+		})
+	}
+}
diff --git a/pkg/util/testingjobs/node/wrappers.go b/pkg/util/testingjobs/node/wrappers.go
index 80afa2281..1e48b3ea4 100644
--- a/pkg/util/testingjobs/node/wrappers.go
+++ b/pkg/util/testingjobs/node/wrappers.go
@@ -63,6 +63,15 @@ func (n *NodeWrapper) Label(k, v string) *NodeWrapper {
 	return n
 }
 
+// Annotation adds an annotation to the Node
+func (n *NodeWrapper) Annotation(k, v string) *NodeWrapper {
+	if n.Annotations == nil {
+		n.Annotations = make(map[string]string)
+	}
+	n.Annotations[k] = v
+	return n
+}
+
 // StatusConditions appends the given status conditions to the Node.
 func (n *NodeWrapper) StatusConditions(conditions ...corev1.NodeCondition) *NodeWrapper {
 	n.Status.Conditions = append(n.Status.Conditions, conditions...)
@@ -107,3 +116,20 @@ func (n *NodeWrapper) Unschedulable() *NodeWrapper {
 	n.Spec.Unschedulable = true
 	return n
 }
+
+// ConditionHeartbeat updates the LastHeartbeatTime of an existing condition.
+func (n *NodeWrapper) ConditionHeartbeat(conditionType corev1.NodeConditionType, heartbeat metav1.Time) *NodeWrapper {
+	for i := range n.Status.Conditions {
+		if n.Status.Conditions[i].Type == conditionType {
+			n.Status.Conditions[i].LastHeartbeatTime = heartbeat
+			break
+		}
+	}
+	return n
+}
+
+// ResourceVersion sets the ResourceVersion of the Node.
+func (n *NodeWrapper) ResourceVersion(version string) *NodeWrapper {
+	n.ObjectMeta.ResourceVersion = version
+	return n
+}