Repository: kubernetes-sigs/kueue

Issue:
TAS: Replace the "NodeToReplace" annotation with a proper status field
<!-- Please only use this template for submitting clean up requests -->

**What would you like to be cleaned**:

I would like to replace the "NodeToReplace" with the proper status API.

This was deferred for later due to [uncertainty about the API](https://github.com/kubernetes-sigs/kueue/tree/main/keps/2724-topology-aware-scheduling#failed-nodes-in-workloadstatus).

However, I think could make the API generic enough so that it can be extended, maybe:
```yaml
status:
  topologyAssignmentRecovery:
     nodesToReplace:
     - name: node1
```

**Why is this needed**:

The annotation requires a dedicated API request to update it, aside from status. 

This means it is difficult to reason about the correctness of the code, because it makes the eviction and node replacement  process "two step", examples:
- it was source of subtle bugs https://github.com/kubernetes-sigs/kueue/pull/5272 
- it requires code which is "detached" to cleanup the annotation here https://github.com/kubernetes-sigs/kueue/pull/5287/files#diff-60dd240c20adbd6a189d018d1c216c2d296730f446c341d8bf449fa6657964ffR197-R209
- it triggers an unnecessary second pass in case of https://github.com/kubernetes-sigs/kueue/issues/5511


Discussion:
@mimowo: cc @pajakd @PBundyra @tenzen-y @mwysokin @mwielgus @pajakd: /assign

Solution patch:
diff --git a/keps/2724-topology-aware-scheduling/README.md b/keps/2724-topology-aware-scheduling/README.md
index f30ed023c..7994275ef 100644
--- a/keps/2724-topology-aware-scheduling/README.md
+++ b/keps/2724-topology-aware-scheduling/README.md
@@ -854,8 +854,7 @@ Initially we plan to support node becoming not ready (as indicated in Node
 monitor nodes and will update each affected TAS workload with the information 
 about the failed nodes. This information will then be consumed by a new mechanism
 in scheduler where we will try to find a new topology assignment and replace the
-failed node(s) (by changing the assignment only on the affected pods). 
-If no replacement is possible, the workload will be evicted. Initially we plan
+failed node(s) (by changing the assignment only on the affected pods). Initially we plan
 to only replace in the case of a single node failure and if no preemption/reclamation
 is neccessary to fit the workload. Since this mechanism is dedicated
 to only replace nodes, it will only work for Topologies which specify
@@ -871,6 +870,9 @@ const (
 )
 ```
 
+Kueue tries to find a replacement for a failed node until success (or until it gets
+evicted by e.g. `waitForPodsReady.recoveryTimeout`). One can limit the number of retries
+to only one, by setting the `TASFailedNodeReplacementFailFast` feature gate to `true`.
 
 ### Implicit defaulting of TAS annotations
 
@@ -964,10 +966,10 @@ As an example let's analyze the assignment based on the size of a podset and sel
 mode. See the table below where the numbers are pods assigned to a particular node.
 The header for columns dedicated to nodes correspond node's initial capacity.
 
-| mode                   | Pods count | 6 | 5 | 4 | 3 | 2 |
-| -----------------------| ---------- | - | - | - | - | - |
-| BestFit                | 12         | 6 | . | 4 | . | 2 |
-| LeastFreeCapacity      | 10         | . | 2 | 4 | 2 | 2 |
+| mode              | Pods count | 6   | 5   | 4   | 3   | 2   |
+| ----------------- | ---------- | --- | --- | --- | --- | --- |
+| BestFit           | 12         | 6   | .   | 4   | .   | 2   |
+| LeastFreeCapacity | 10         | .   | 2   | 4   | 2   | 2   |
 
 Explanation:
 - `BestFit` - We prioritized 3rd node over 2nd node because 3rd node was a tight fit among all domains that could fit 2 slices. The last domain has been "optimized" to find the tight fit. 
@@ -1164,6 +1166,7 @@ The new validations which are for MVP, but likely will be relaxed in the future:
 - add observability metrics, some ideas are in the [discussion](https://github.com/kubernetes-sigs/kueue/pull/5078#discussion_r2060580973)
 - change how the information about the failed nodes is stored at a Workload from Annotation into a field in workload.Status
 - handle a more comprehensive set of failure scenarios (e.g., including node becoming unschedulable due to a taint)
+- re-evaluate replacing `NodeToReplace` annotation with a status field, to optimize number of requests in scheduler loop. [Discussion](https://github.com/kubernetes-sigs/kueue/issues/5560)
 
 #### Stable
 
@@ -1338,9 +1341,9 @@ the last domain.
 This algorithm was enabled by the `TASProfileMostFreeCapacity` feature flag and it
 was independent of PodSet's annotations:
 
-| featuregate/annotation     | preferred         | required          | unconstrained     |
-| -------------------------- | ----------------- | ----------------- | ----------------- |
-| TASProfileMostFreeCapacity | MostFreeCapacity  | MostFreeCapacity  | MostFreeCapacity  |
+| featuregate/annotation     | preferred        | required         | unconstrained    |
+| -------------------------- | ---------------- | ---------------- | ---------------- |
+| TASProfileMostFreeCapacity | MostFreeCapacity | MostFreeCapacity | MostFreeCapacity |
 
 #### Example
 Consider a rack with four nodes that can accommodate 3, 3, 2, and 1 pod, respectively.
diff --git a/keps/2724-topology-aware-scheduling/kep.yaml b/keps/2724-topology-aware-scheduling/kep.yaml
index a28ac72f2..9c61ac780 100644
--- a/keps/2724-topology-aware-scheduling/kep.yaml
+++ b/keps/2724-topology-aware-scheduling/kep.yaml
@@ -3,6 +3,7 @@ kep-number: 2724
 authors:
   - "@mimowo"
   - "@mwielgus"
+  - "@pbundyra"
 status: provisional
 creation-date: 2023-07-30
 reviewers:
@@ -37,6 +38,8 @@ feature-gates:
   - name: TASProfileLeastAllocated
   - name: TASProfileMixed
   - name: TASProfileLeastFreeCapacity
+  - name: TASFailedNodeReplacement
+  - name: TASFailedNodeReplacementFailFast
 disable-supported: true
 
 # The following PRR answers are required at beta release
diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go
index 905460112..a7e323689 100644
--- a/pkg/features/kube_features.go
+++ b/pkg/features/kube_features.go
@@ -169,6 +169,12 @@ const (
 	//
 	// Enable replacement of failed node in TAS.
 	TASFailedNodeReplacement featuregate.Feature = "TASFailedNodeReplacement"
+
+	// owner: @pbundyra
+	// kep: https://github.com/kubernetes-sigs/kueue/tree/main/keps/2724-topology-aware-scheduling
+	//
+	// Evict Workload if Kueue couldn't find replacement for a failed node in TAS in the first attempt.
+	TASFailedNodeReplacementFailFast featuregate.Feature = "TASFailedNodeReplacementFailFast"
 )
 
 func init() {
@@ -264,6 +270,9 @@ var defaultVersionedFeatureGates = map[featuregate.Feature]featuregate.Versioned
 	TASFailedNodeReplacement: {
 		{Version: version.MustParse("0.12"), Default: false, PreRelease: featuregate.Alpha},
 	},
+	TASFailedNodeReplacementFailFast: {
+		{Version: version.MustParse("0.13"), Default: false, PreRelease: featuregate.Alpha},
+	},
 }
 
 func SetFeatureGateDuringTest(tb testing.TB, f featuregate.Feature, value bool) {
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 8ee9d558d..6046b1424 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -37,6 +37,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"
 	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/features"
@@ -220,6 +221,17 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 		ctx := ctrl.LoggerInto(ctx, log)
 
 		mode := e.assignment.RepresentativeMode()
+
+		if features.Enabled(features.TASFailedNodeReplacementFailFast) && workload.HasTopologyAssignmentWithNodeToReplace(e.Obj) && mode != flavorassigner.Fit {
+			// evict workload we couldn't find the replacement for
+			if err := s.evictWorkloadAfterFailedTASReplacement(ctx, log, e.Obj); err != nil {
+				log.V(2).Error(err, "Failed to evict workload after failed try to find a node replacement")
+				continue
+			}
+			e.status = evicted
+			continue
+		}
+
 		if mode == flavorassigner.NoFit {
 			log.V(3).Info("Skipping workload as FlavorAssigner assigned NoFit mode")
 			continue
@@ -296,7 +308,9 @@ func (s *Scheduler) schedule(ctx context.Context) wait.SpeedSignal {
 	result := metrics.AdmissionResultInadmissible
 	for _, e := range entries {
 		logAdmissionAttemptIfVerbose(log, &e)
-		if e.status != assumed {
+		// When the workload is evicted by scheduler we skip requeueAndUpdate.
+		// The eviction process will be finalized by the workload controller.
+		if e.status != assumed && e.status != evicted {
 			s.requeueAndUpdate(ctx, e)
 		} else {
 			result = metrics.AdmissionResultSuccess
@@ -322,6 +336,8 @@ const (
 	nominated entryStatus = "nominated"
 	// indicates if the workload was skipped in this cycle.
 	skipped entryStatus = "skipped"
+	// indicates if the workload was evicted in this cycle.
+	evicted entryStatus = "evicted"
 	// indicates if the workload was assumed to have been admitted.
 	assumed entryStatus = "assumed"
 	// indicates that the workload was never nominated for admission.
@@ -486,6 +502,18 @@ func (s *Scheduler) getInitialAssignments(log logr.Logger, wl *workload.Info, sn
 	return fullAssignment, nil
 }
 
+func (s *Scheduler) evictWorkloadAfterFailedTASReplacement(ctx context.Context, log logr.Logger, wl *kueue.Workload) error {
+	log.V(3).Info("Evicting workload after failed try to find a node replacement; TASFailedNodeReplacementFailFast enabled")
+	msg := fmt.Sprintf("Workload was evicted as there was no replacement for a failed node: %s", workload.NodeToReplace(wl))
+	if err := workload.EvictWorkload(ctx, s.client, s.recorder, wl, kueue.WorkloadEvictedDueToNodeFailures, msg, s.clock); err != nil {
+		return err
+	}
+	if err := workload.RemoveAnnotation(ctx, s.client, wl, kueuealpha.NodeToReplaceAnnotation); err != nil {
+		return fmt.Errorf("failed to remove annotation for node replacement %s", kueuealpha.NodeToReplaceAnnotation)
+	}
+	return nil
+}
+
 func updateAssignmentForTAS(cq *cache.ClusterQueueSnapshot, wl *workload.Info, assignment *flavorassigner.Assignment, targets []*preemption.Target) {
 	if features.Enabled(features.TopologyAwareScheduling) && assignment.RepresentativeMode() == flavorassigner.Preempt && (wl.IsRequestingTAS() || cq.IsTASOnly()) && !workload.HasTopologyAssignmentWithNodeToReplace(wl.Obj) {
 		tasRequests := assignment.WorkloadsTopologyRequests(wl, cq)
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index 953dc2c90..ee14a54c8 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/record"
@@ -47,6 +48,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/metrics"
 	"sigs.k8s.io/kueue/pkg/resources"
 	"sigs.k8s.io/kueue/pkg/util/api"
+	clientutil "sigs.k8s.io/kueue/pkg/util/client"
 	utilptr "sigs.k8s.io/kueue/pkg/util/ptr"
 	utilslices "sigs.k8s.io/kueue/pkg/util/slices"
 )
@@ -920,6 +922,14 @@ func HasNodeToReplace(w *kueue.Workload) bool {
 	return found
 }
 
+func NodeToReplace(w *kueue.Workload) string {
+	if !HasNodeToReplace(w) {
+		return ""
+	}
+	annotations := w.GetAnnotations()
+	return annotations[kueuealpha.NodeToReplaceAnnotation]
+}
+
 func HasTopologyAssignmentWithNodeToReplace(w *kueue.Workload) bool {
 	if !HasNodeToReplace(w) || !IsAdmitted(w) {
 		return false
@@ -957,6 +967,20 @@ func RemoveFinalizer(ctx context.Context, c client.Client, wl *kueue.Workload) e
 	return nil
 }
 
+func RemoveAnnotation(ctx context.Context, cl client.Client, wl *kueue.Workload, annotation string) error {
+	wlKey := types.NamespacedName{Name: wl.Name, Namespace: wl.Namespace}
+	var wlToPatch kueue.Workload
+	if err := cl.Get(ctx, wlKey, &wlToPatch); err != nil {
+		return err
+	}
+	return clientutil.Patch(ctx, cl, &wlToPatch, false, func() (bool, error) {
+		annotations := wlToPatch.GetAnnotations()
+		delete(annotations, annotation)
+		wlToPatch.SetAnnotations(annotations)
+		return true, nil
+	})
+}
+
 // AdmissionChecksForWorkload returns AdmissionChecks that should be assigned to a specific Workload based on
 // ClusterQueue configuration and ResourceFlavors
 func AdmissionChecksForWorkload(log logr.Logger, wl *kueue.Workload, admissionChecks map[kueue.AdmissionCheckReference]sets.Set[kueue.ResourceFlavorReference]) sets.Set[kueue.AdmissionCheckReference] {
@@ -1003,6 +1027,20 @@ func AdmissionChecksForWorkload(log logr.Logger, wl *kueue.Workload, admissionCh
 	return acNames
 }
 
+func EvictWorkload(ctx context.Context, c client.Client, recorder record.EventRecorder, wl *kueue.Workload, reason, msg string, clock clock.Clock) error {
+	SetEvictedCondition(wl, reason, msg)
+	ResetChecksOnEviction(wl, clock.Now())
+	if err := ApplyAdmissionStatus(ctx, c, wl, true, clock); err != nil {
+		return err
+	}
+	reportWorkloadEvictedOnce := WorkloadEvictionStateInc(wl, reason, "")
+	ReportEvictedWorkload(recorder, wl, wl.Status.Admission.ClusterQueue, reason, msg)
+	if reportWorkloadEvictedOnce {
+		metrics.ReportEvictedWorkloadsOnce(wl.Status.Admission.ClusterQueue, reason, "")
+	}
+	return nil
+}
+
 func ReportEvictedWorkload(recorder record.EventRecorder, wl *kueue.Workload, cqName kueue.ClusterQueueReference, reason, message string) {
 	metrics.ReportEvictedWorkloads(cqName, reason)
 	if features.Enabled(features.LocalQueueMetrics) {
diff --git a/site/content/en/docs/installation/_index.md b/site/content/en/docs/installation/_index.md
index dd3750453..9fe2604d2 100644
--- a/site/content/en/docs/installation/_index.md
+++ b/site/content/en/docs/installation/_index.md
@@ -266,7 +266,7 @@ spec:
 ### Feature gates for alpha and beta features
 
 | Feature                               | Default | Stage | Since | Until |
-|---------------------------------------|---------|-------|-------|-------|
+| ------------------------------------- | ------- | ----- | ----- | ----- |
 | `FlavorFungibility`                   | `true`  | Beta  | 0.5   |       |
 | `MultiKueue`                          | `false` | Alpha | 0.6   | 0.8   |
 | `MultiKueue`                          | `true`  | Beta  | 0.9   |       |
@@ -296,11 +296,12 @@ spec:
 | `ObjectRetentionPolicies`             | `false` | Alpha | 0.12  |       |
 | `TASFailedNodeReplacement`            | `false` | Alpha | 0.12  |       |
 | `AdmissionFairSharing`                | `false` | Alpha | 0.12  |       |
+| `TASFailedNodeReplacementFailFast`    | `false` | Alpha | 0.12  |       |
 
 ### Feature gates for graduated or deprecated features
 
 | Feature                           | Default | Stage      | Since | Until |
-|-----------------------------------|---------|------------|-------|-------|
+| --------------------------------- | ------- | ---------- | ----- | ----- |
 | `QueueVisibility`                 | `false` | Alpha      | 0.4   | 0.9   |
 | `QueueVisibility`                 | `false` | Deprecated | 0.9   |       |
 | `MultiplePreemptions`             | `false` | Alpha      | 0.8   | 0.8   |



Test patch:
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index 196bbe6b6..fb855289d 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -35,6 +35,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/component-base/featuregate"
 	"k8s.io/component-base/metrics/testutil"
 	testingclock "k8s.io/utils/clock/testing"
 	"k8s.io/utils/ptr"
@@ -4818,6 +4819,8 @@ func TestScheduleForTAS(t *testing.T) {
 		wantEvents []utiltesting.EventRecord
 		// eventCmpOpts are the comparison options for the events
 		eventCmpOpts cmp.Options
+
+		featureGates []featuregate.Feature
 	}{
 		"workload in CQ with ProvisioningRequest; second pass; baseline scenario": {
 			nodes:           defaultSingleNode,
@@ -5120,6 +5123,66 @@ func TestScheduleForTAS(t *testing.T) {
 				},
 			},
 		},
+		"workload with nodeToReplace annotation; second pass; preferred; no fit; FailFast": {
+			nodes:           defaultNodes,
+			admissionChecks: []kueue.AdmissionCheck{defaultProvCheck},
+			topologies:      []kueuealpha.Topology{defaultThreeLevelTopology},
+			resourceFlavors: []kueue.ResourceFlavor{defaultTASThreeLevelFlavor},
+			clusterQueues:   []kueue.ClusterQueue{defaultClusterQueue},
+			workloads: []kueue.Workload{
+				*utiltesting.MakeWorkload("foo", "default").
+					Annotations(map[string]string{kueuealpha.NodeToReplaceAnnotation: "x0"}).
+					Queue("tas-main").
+					PodSets(*utiltesting.MakePodSet("one", 1).
+						PreferredTopologyRequest(tasRackLabel).
+						Request(corev1.ResourceCPU, "3").
+						Obj()).
+					ReserveQuota(
+						utiltesting.MakeAdmission("tas-main", "one").
+							Assignment(corev1.ResourceCPU, "tas-default", "3000m").
+							AssignmentPodCount(1).
+							TopologyAssignment(&kueue.TopologyAssignment{
+								Levels: utiltas.Levels(&defaultSingleLevelTopology),
+								Domains: []kueue.TopologyDomainAssignment{
+									{
+										Count: 1,
+										Values: []string{
+											"x0",
+										},
+									},
+								},
+							}).
+							Obj(),
+					).
+					Admitted(true).
+					Obj(),
+			},
+			wantNewAssignments: map[workload.Reference]kueue.Admission{
+				"default/foo": *utiltesting.MakeAdmission("tas-main", "one").
+					Assignment(corev1.ResourceCPU, "tas-default", "3000m").
+					AssignmentPodCount(1).
+					TopologyAssignment(&kueue.TopologyAssignment{
+						Levels: utiltas.Levels(&defaultSingleLevelTopology),
+						Domains: []kueue.TopologyDomainAssignment{
+							{
+								Count: 1,
+								Values: []string{
+									"x0",
+								},
+							},
+						},
+					}).Obj(),
+			},
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key:       types.NamespacedName{Namespace: "default", Name: "foo"},
+					EventType: corev1.EventTypeNormal,
+					Reason:    "EvictedDueToNodeFailures",
+					Message:   "Workload was evicted as there was no replacement for a failed node: x0",
+				},
+			},
+			featureGates: []featuregate.Feature{features.TASFailedNodeReplacementFailFast},
+		},
 		"workload with nodeToReplace annotation; second pass; required rack; fit": {
 			nodes:           defaultNodes,
 			admissionChecks: []kueue.AdmissionCheck{defaultProvCheck},
@@ -7083,6 +7146,9 @@ func TestScheduleForTAS(t *testing.T) {
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
 			features.SetFeatureGateDuringTest(t, features.TopologyAwareScheduling, true)
+			for _, fg := range tc.featureGates {
+				features.SetFeatureGateDuringTest(t, fg, true)
+			}
 			ctx, log := utiltesting.ContextWithLog(t)
 
 			clientBuilder := utiltesting.NewClientBuilder().
@@ -7093,9 +7159,10 @@ func TestScheduleForTAS(t *testing.T) {
 					&corev1.PodList{Items: tc.pods},
 					&corev1.NodeList{Items: tc.nodes},
 					&kueue.LocalQueueList{Items: queues}).
-				WithObjects(
-					utiltesting.MakeNamespace("default"),
-				)
+				WithObjects(utiltesting.MakeNamespace("default")).
+				WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge}).
+				WithStatusSubresource(&kueue.Workload{}, &kueue.ClusterQueue{}, &kueue.LocalQueue{})
+
 			for _, ac := range tc.admissionChecks {
 				clientBuilder = clientBuilder.WithStatusSubresource(ac.DeepCopy())
 			}
diff --git a/test/integration/singlecluster/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index 715ec804a..9c5388eb8 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -1220,6 +1220,163 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 			})
 		})
 
+		ginkgo.When("TASNodeFailedReplacementFailFast is enabled", func() {
+			var (
+				nodes []corev1.Node
+			)
+			ginkgo.BeforeEach(func() {
+				nodes = []corev1.Node{
+					*testingnode.MakeNode("x1").
+						Label("node-group", "tas").
+						Label(testing.DefaultBlockTopologyLevel, "b1").
+						Label(testing.DefaultRackTopologyLevel, "r1").
+						Label(corev1.LabelHostname, "x1").
+						StatusAllocatable(corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("1"),
+							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
+						}).
+						Ready().
+						Obj(),
+					*testingnode.MakeNode("x2").
+						Label("node-group", "tas").
+						Label(testing.DefaultBlockTopologyLevel, "b1").
+						Label(testing.DefaultRackTopologyLevel, "r1").
+						Label(corev1.LabelHostname, "x2").
+						StatusAllocatable(corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("1"),
+							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
+						}).
+						Ready().
+						Obj(),
+					*testingnode.MakeNode("x3").
+						Label("node-group", "tas").
+						Label(testing.DefaultBlockTopologyLevel, "b1").
+						Label(testing.DefaultRackTopologyLevel, "r2").
+						Label(corev1.LabelHostname, "x3").
+						StatusAllocatable(corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("1"),
+							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
+						}).
+						Ready().
+						Obj(),
+					*testingnode.MakeNode("x4").
+						Label("node-group", "tas").
+						Label(testing.DefaultBlockTopologyLevel, "b1").
+						Label(testing.DefaultRackTopologyLevel, "r2").
+						Label(corev1.LabelHostname, "x4").
+						StatusAllocatable(corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("1"),
+							corev1.ResourceMemory: resource.MustParse("1Gi"),
+							corev1.ResourcePods:   resource.MustParse("10"),
+						}).
+						Ready().
+						Obj(),
+				}
+				util.CreateNodesWithStatus(ctx, k8sClient, nodes)
+
+				topology = testing.MakeDefaultThreeLevelTopology("default")
+				util.MustCreate(ctx, k8sClient, topology)
+
+				tasFlavor = testing.MakeResourceFlavor("tas-flavor").
+					NodeLabel("node-group", "tas").
+					TopologyName("default").Obj()
+				util.MustCreate(ctx, k8sClient, tasFlavor)
+
+				clusterQueue = testing.MakeClusterQueue("cluster-queue").
+					ResourceGroup(*testing.MakeFlavorQuotas(tasFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj()).
+					Obj()
+				util.MustCreate(ctx, k8sClient, clusterQueue)
+				util.ExpectClusterQueuesToBeActive(ctx, k8sClient, clusterQueue)
+
+				localQueue = testing.MakeLocalQueue("local-queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+				util.MustCreate(ctx, k8sClient, localQueue)
+
+				_ = features.SetEnable(features.TASFailedNodeReplacementFailFast, true)
+			})
+
+			ginkgo.AfterEach(func() {
+				gomega.Expect(util.DeleteAllJobsInNamespace(ctx, k8sClient, ns)).Should(gomega.Succeed())
+				gomega.Expect(util.DeleteWorkloadsInNamespace(ctx, k8sClient, ns)).Should(gomega.Succeed())
+				gomega.Expect(util.DeleteObject(ctx, k8sClient, localQueue)).Should(gomega.Succeed())
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, tasFlavor, true)
+				util.ExpectObjectToBeDeleted(ctx, k8sClient, topology, true)
+				for _, node := range nodes {
+					util.ExpectObjectToBeDeleted(ctx, k8sClient, &node, true)
+				}
+
+				_ = features.SetEnable(features.TASFailedNodeReplacementFailFast, false)
+			})
+
+			ginkgo.It("should evict the Workload if no replacement is possible and place it on another rack", func() {
+				var wl1 *kueue.Workload
+				nodeName := nodes[0].Name
+
+				ginkgo.By("creating a workload", func() {
+					wl1 = testing.MakeWorkload("wl1", ns.Name).
+						PodSets(*testing.MakePodSet("worker", 2).
+							RequiredTopologyRequest(testing.DefaultRackTopologyLevel).
+							Obj()).
+						Queue(kueue.LocalQueueName(localQueue.Name)).Request(corev1.ResourceCPU, "1").Obj()
+					util.MustCreate(ctx, k8sClient, wl1)
+				})
+
+				ginkgo.By("verify the workload is admitted", func() {
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl1)
+					gomega.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl1), wl1)).To(gomega.Succeed())
+					gomega.Expect(wl1.Status.Admission.PodSetAssignments[0].TopologyAssignment).Should(gomega.BeComparableTo(
+						&kueue.TopologyAssignment{
+							Levels: []string{corev1.LabelHostname},
+							Domains: []kueue.TopologyDomainAssignment{
+								{Count: 1, Values: []string{"x1"}},
+								{Count: 1, Values: []string{"x2"}},
+							},
+						},
+					))
+				})
+
+				ginkgo.By("making the node NotReady 30s in the past", func() {
+					nodeToUpdate := &corev1.Node{}
+					gomega.Expect(k8sClient.Get(ctx, apitypes.NamespacedName{Name: nodeName}, nodeToUpdate)).Should(gomega.Succeed())
+
+					util.SetNodeCondition(ctx, k8sClient, nodeToUpdate, &corev1.NodeCondition{
+						Type:               corev1.NodeReady,
+						Status:             corev1.ConditionFalse,
+						LastTransitionTime: metav1.NewTime(time.Now().Add(-tas.NodeFailureDelay)),
+					})
+				})
+
+				ginkgo.By("verify the workload is evicted due to no replacement possible", func() {
+					util.FinishEvictionForWorkloads(ctx, k8sClient, wl1)
+					util.ExpectEvictedWorkloadsTotalMetric(clusterQueue.Name, kueue.WorkloadEvictedDueToNodeFailures, 1)
+					gomega.Eventually(func(g gomega.Gomega) {
+						updatedWl := &kueue.Workload{}
+						g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl1), updatedWl)).To(gomega.Succeed())
+						annotations := updatedWl.GetAnnotations()
+						_, found := annotations[kueuealpha.NodeToReplaceAnnotation]
+						g.Expect(found).To(gomega.BeFalse())
+					}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				})
+
+				ginkgo.By("verify the workload is placed on another rack", func() {
+					util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wl1)
+					gomega.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(wl1), wl1)).To(gomega.Succeed())
+					gomega.Expect(wl1.Status.Admission.PodSetAssignments[0].TopologyAssignment).Should(gomega.BeComparableTo(
+						&kueue.TopologyAssignment{
+							Levels: []string{corev1.LabelHostname},
+							Domains: []kueue.TopologyDomainAssignment{
+								{Count: 1, Values: []string{"x3"}},
+								{Count: 1, Values: []string{"x4"}},
+							},
+						},
+					))
+				})
+			})
+		})
+
 		ginkgo.When("Preemption is enabled within ClusterQueue", func() {
 			var (
 				nodes []corev1.Node