fix: solve preemption issue (#555)

Author: Mishaoyue

charts/tensor-fusion/values.yaml

@@ -201,8 +201,6 @@ schedulerConfig:
         maxWorkerPerNode: 256
         vramWeight: 0.7
         tflopsWeight: 0.3
-        # preemptClusterWide: true = preempt pods cluster-wide; false = only accept victims in preemptor's namespace
-        preemptClusterWide: true
     - name: GPUNetworkTopologyAware
       args:
         # Avoid the remote TFWorker RX/TX to avoid single node consume too much bandwidth
@@ -238,7 +236,7 @@ dynamicConfig:
   # extra pod labels to be added to metrics, 
   # you can map label keys to other measure tags
   metricsExtraPodLabels: {}
-
+  preemptClusterWideFromEnv: true
   # alert rules
   alertRules:    
     # Worker TFlops throttled alert

cmd/main.go

@@ -21,6 +21,7 @@ import (
 	"crypto/tls"
 	"flag"
 	"os"
+	"strconv"
 	"time"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
@@ -597,7 +598,14 @@ func setupTimeSeriesAndWatchGlobalConfigChanges(ctx context.Context, mgr manager
 		os.Exit(1)
 	}
 	config.SetGlobalConfig(globalConfig)
-
+	// set env so scheduler default preemption reads effective preempt cluster-wide
+	preemptVal := strconv.FormatBool(globalConfig.PreemptClusterWideFromEnv)
+	if err := os.Setenv(constants.SchedulerPreemptClusterWideEnv, preemptVal); err != nil {
+		setupLog.Error(err, "failed to set preempt cluster-wide env")
+	}
+	// get env PreemptClusterWideFromEnv
+	preemptClusterWideFromEnv := os.Getenv(constants.SchedulerPreemptClusterWideEnv)
+	setupLog.Info("preempt cluster-wide from env", "preemptClusterWideFromEnv", preemptClusterWideFromEnv)
 	// only init TSDB and evaluator in leader
 	needTSDB := enableAlert || enableAutoScale
 
@@ -636,8 +644,16 @@ func watchAndHandleConfigChanges(ctx context.Context, mgr manager.Manager, needT
 				"configPath", dynamicConfigPath)
 			continue
 		}
+		// check if preemptClusterWideFromEnv is changed
+		if globalConfig.PreemptClusterWideFromEnv != config.GetGlobalConfig().PreemptClusterWideFromEnv {
+			preemptVal := strconv.FormatBool(globalConfig.PreemptClusterWideFromEnv)
+			if err := os.Setenv(constants.SchedulerPreemptClusterWideEnv, preemptVal); err != nil {
+				ctrl.Log.Error(err, "failed to set preempt cluster-wide env")
+			}
+			preemptClusterWideFromEnv := os.Getenv(constants.SchedulerPreemptClusterWideEnv)
+			ctrl.Log.Info("preempt cluster-wide from env", "preemptClusterWideFromEnv", preemptClusterWideFromEnv)
+		}
 		config.SetGlobalConfig(globalConfig)
-
 		// handle alert rules update
 		go func() {
 			<-alertEvaluatorReady

config/samples/dynamic-config.yaml

@@ -4,7 +4,7 @@ metricsTTL: 30d
 metricsFormat: influx
 
 autoScalingInterval: 10s
-
+preemptClusterWideFromEnv: true
 alertRules:    
   # Worker TFlops throttled alert
   - name: WorkerTFlopsThrottled

config/samples/scheduler-config.yaml

@@ -33,8 +33,6 @@ profiles:
       maxWorkerPerNode: 256
       vramWeight: 0.7
       tflopsWeight: 0.3
-      # preemptClusterWide: true = preempt pods cluster-wide; false = only accept victims in preemptor's namespace
-      preemptClusterWide: true
   - name: GPUNetworkTopologyAware
     args:
       # Avoid the remote TFWorker RX/TX to avoid single node consume too much bandwidth

internal/config/global_config.go

@@ -7,15 +7,15 @@ import (
 )
 
 type GlobalConfig struct {
-	MetricsTTL            string            `yaml:"metricsTTL"`
-	MetricsFormat         string            `yaml:"metricsFormat"`
-	MetricsExtraPodLabels map[string]string `yaml:"metricsExtraPodLabels"`
-
-	AlertRules    []AlertRule          `yaml:"alertRules"`
-	AutoMigration *AutoMigrationConfig `yaml:"autoMigration"`
-
-	AutoScalingInterval  string `yaml:"autoScalingInterval"`
-	GPUOperatorNamespace string `yaml:"gpuOperatorNamespace"`
+	MetricsTTL            string               `yaml:"metricsTTL"`
+	MetricsFormat         string               `yaml:"metricsFormat"`
+	MetricsExtraPodLabels map[string]string    `yaml:"metricsExtraPodLabels"`
+	AlertRules            []AlertRule          `yaml:"alertRules"`
+	AutoMigration         *AutoMigrationConfig `yaml:"autoMigration"`
+
+	AutoScalingInterval       string `yaml:"autoScalingInterval"`
+	GPUOperatorNamespace      string `yaml:"gpuOperatorNamespace"`
+	PreemptClusterWideFromEnv bool   `yaml:"preemptClusterWideFromEnv"`
 }
 
 type AutoMigrationConfig struct {

internal/config/scheduler_config.go

@@ -5,10 +5,6 @@ type GPUFitConfig struct {
 
 	VramWeight   float64 `json:"vramWeight"`
 	TflopsWeight float64 `json:"tflopsWeight"`
-
-	// PreemptClusterWide when true or unset allows preempting pods cluster-wide; when false only victims
-	// in the preemptor pod's namespace are accepted (validated in Filter phase after DefaultPreemption selects victims).
-	PreemptClusterWide *bool `json:"preemptClusterWide,omitempty"`
 }
 
 type GPUNetworkTopologyAwareConfig struct {

internal/scheduler/gpuresources/gpuresources.go

@@ -1062,30 +1062,13 @@ func (s *GPUFit) validatePreemption(state fwk.CycleState, pod *v1.Pod, nodeInfo
 		return fwk.NewStatus(fwk.Error, "invalid type for preempt pods")
 	}
 
-	// When PreemptClusterWide is explicitly false, only use victims in the preemptor's namespace for subsequent checks.
-	victimsToUse := victims
-	if s.cfg != nil && s.cfg.PreemptClusterWide != nil && !*s.cfg.PreemptClusterWide {
-		victimsToUse = sets.New[types.NamespacedName]()
-		for v := range victims {
-			if v.Namespace == pod.Namespace {
-				victimsToUse.Insert(v)
-			}
-		}
-		if victimsToUse.Len() == 0 {
-			s.logger.Info("GPU preemption: no same-namespace victims on node",
-				"pod", klog.KObj(pod),
-				"node", nodeName)
-			return fwk.NewStatus(fwk.Unschedulable, "no victims in preemptor namespace on this node")
-		}
-	}
-
 	victimList := victims.UnsortedList()
 	//  Verify that preemption will release sufficient GPU resources
 	// This calls CheckQuotaAndFilterSingleNodePreempt which:
 	// - Simulates releasing victim GPU resources
 	// - Applies GPU filters (resource, model, vendor, affinity, same-node)
 	// - Checks quota constraints
-	err = s.allocator.CheckQuotaAndFilterSingleNodePreempt(nodeName, allocReq, victimsToUse)
+	err = s.allocator.CheckQuotaAndFilterSingleNodePreempt(nodeName, allocReq, victims)
 	if err != nil {
 		s.logger.Info("GPU preemption validation failed",
 			"pod", pod.Name,

patches/scheduler-pdb-1.patch

@@ -1,15 +1,23 @@
 --- ../vendor/k8s.io/kubernetes/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go	2026-01-26 00:00:00
 +++ ../vendor/k8s.io/kubernetes/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go	2026-01-26 00:00:00
-@@ -20,6 +20,7 @@
- 	"context"
- 	"fmt"
+@@ -22,4 +22,7 @@
  	"math/rand"
  	"sort"
 
++	"os"
++	"strings"
 +	"time"
  	v1 "k8s.io/api/core/v1"
-@@ -108,3 +109,17 @@
+@@ -108,6 +111,26 @@
+ 	// Default behavior: No additional filtering, beyond the internal requirement that the victim pod
+ 	// have lower priority than the preemptor pod.
  	pl.IsEligiblePod = func(nodeInfo fwk.NodeInfo, victim fwk.PodInfo, preemptor *v1.Pod) bool {
++		// Cluster-wide when env is unset or "true"; "false" = same-tenant (same namespace) only.
++		v := strings.ToLower(strings.TrimSpace(os.Getenv("TF_SCHEDULER_PREEMPT_CLUSTER_WIDE")))
++		preemptClusterWide := v == "" || v == "true"
++		if !preemptClusterWide && victim.GetPod().Namespace != preemptor.Namespace {
++			return false
++		}
 +		victimAnnotation := victim.GetPod().Annotations
 +		if victimAnnotation == nil {
 +			return true
@@ -26,3 +34,4 @@
 +		}
  		return true
  	}
+ 

pkg/constants/env.go

@@ -3,6 +3,10 @@ package constants
 // Controller itself envs
 const NamespaceEnv = "OPERATOR_NAMESPACE"
 
+// Scheduler preemption: when PreemptClusterWideFromEnv is enabled, effective value is read from this env.
+// "true" = cluster-wide preemption; "false" = same-tenant (same namespace) only.
+const SchedulerPreemptClusterWideEnv = "TF_SCHEDULER_PREEMPT_CLUSTER_WIDE"
+
 // System feature toggles
 const (
 	EnableWebhookEnv                  = "ENABLE_WEBHOOKS"