feat: support Kubernetes Coscheduling Plugin for PodGroup gang scheduling

- Retry LockNode with backoff on lock contention to allow concurrent
  PodGroup bind operations to serialize gracefully
- Evict pending pods with stale annotations in onAddPod to prevent
  phantom GPU occupancy when Coscheduling denies a PodGroup
- Call podManager.DelPod in Bind failure path to prevent ghost cache
  entries on failed binds
- Add --node-lock-retry-timeout flag (default: 30s)

Signed-off-by: lin121291 <4jp33f9e@gmail.com>

Author: lin121291

cmd/scheduler/main.go

@@ -80,6 +80,7 @@ func init() {
 	rootCmd.Flags().IntVar(&config.Timeout, "kube-timeout", client.DefaultTimeout, "Timeout to use while talking with kube-apiserver.")
 	rootCmd.Flags().BoolVar(&enableProfiling, "profiling", false, "Enable pprof profiling via HTTP server")
 	rootCmd.Flags().DurationVar(&config.NodeLockTimeout, "node-lock-timeout", time.Minute*5, "timeout for node locks")
+	rootCmd.Flags().DurationVar(&config.NodeLockRetryTimeout, "node-lock-retry-timeout", time.Second*30, "timeout for retrying node lock acquisition when locked by valid pod")
 	rootCmd.Flags().BoolVar(&config.ForceOverwriteDefaultScheduler, "force-overwrite-default-scheduler", true, "Overwrite schedulerName in Pod Spec when set to the const DefaultSchedulerName in https://k8s.io/api/core/v1 package")
 
 	rootCmd.Flags().BoolVar(&config.LeaderElect, "leader-elect", false, "The pod of hami-scheduler enable leader select")
@@ -115,6 +116,8 @@ func start() error {
 	// Initialize node lock timeout from config
 	nodelock.NodeLockTimeout = config.NodeLockTimeout
 	klog.InfoS("Set node lock timeout", "timeout", nodelock.NodeLockTimeout)
+	nodelock.NodeLockRetryTimeout = config.NodeLockRetryTimeout
+	klog.InfoS("Set node lock retry timeout", "timeout", nodelock.NodeLockRetryTimeout)
 	client.InitGlobalClient(
 		client.WithBurst(config.Burst),
 		client.WithQPS(config.QPS),

pkg/scheduler/config/config.go

@@ -63,6 +63,9 @@ var (
 	// NodeLockTimeout is the timeout for node locks.
 	NodeLockTimeout time.Duration
 
+	// NodeLockRetryTimeout is the timeout for retrying node lock acquisition.
+	NodeLockRetryTimeout time.Duration
+
 	// If set to false, When Pod.Spec.SchedulerName equals to the const DefaultSchedulerName in k8s.io/api/core/v1 package, webhook will not overwrite it, default value is true.
 	ForceOverwriteDefaultScheduler bool
 

pkg/scheduler/scheduler.go

@@ -157,6 +157,13 @@ func (s *Scheduler) onAddPod(obj any) {
 		s.podManager.UpdatePod(pod)
 		return
 	}
+	// When Coscheduling denies a PodGroup, pods return to Pending with stale
+	// HAMi annotations. Evict them from cache to prevent phantom GPU occupancy.
+	if pod.Status.Phase == corev1.PodPending && pod.Annotations[util.DeviceBindPhase] != util.DeviceBindSuccess {
+		klog.V(5).InfoS("Pod is pending with stale annotations, evicting from cache", "pod", pod.Name)
+		s.podManager.DelPod(pod)
+		return
+	}
 	podDev, _ := device.DecodePodDevices(device.SupportDevices, pod.Annotations)
 	if s.podManager.AddPod(pod, nodeID, podDev) {
 		s.quotaManager.AddUsage(pod, podDev)
@@ -717,6 +724,7 @@ ReleaseNodeLocks:
 	for _, val := range device.GetDevices() {
 		val.ReleaseNodeLock(node, current)
 	}
+	s.podManager.DelPod(current)
 	s.recordScheduleBindingResultEvent(current, EventReasonBindingFailed, []string{}, err)
 	return &extenderv1.ExtenderBindingResult{Error: err.Error()}, nil
 }

pkg/util/nodelock/nodelock.go

@@ -18,6 +18,7 @@ package nodelock
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"strings"
@@ -40,11 +41,19 @@ const (
 	NodeLockSep = ","
 )
 
+// errNodeLocked is a sentinel error indicating the node is locked by a valid
+// (non-expired, non-dangling) pod. LockNode uses this to distinguish retryable
+// lock contention from non-retryable errors.
+var errNodeLocked = errors.New("node is locked by a valid pod")
+
 var (
 	// nodeLocks manages per-node locks for fine-grained concurrency control.
 	nodeLocks = newNodeLockManager()
 	// NodeLockTimeout is the global timeout for node locks.
 	NodeLockTimeout time.Duration = time.Minute * 5
+	// NodeLockRetryTimeout is the timeout for retrying lock acquisition when
+	// a node is locked by a valid (non-expired, non-dangling) pod.
+	NodeLockRetryTimeout time.Duration = time.Second * 30
 
 	DefaultStrategy = wait.Backoff{
 		Steps:    5,
@@ -115,6 +124,16 @@ func setupNodeLockTimeout() {
 			klog.InfoS("Node lock expiration time set from environment variable", "duration", d)
 		}
 	}
+	retryTimeout := os.Getenv("HAMI_NODELOCK_RETRY_TIMEOUT")
+	if retryTimeout != "" {
+		d, err := time.ParseDuration(retryTimeout)
+		if err != nil {
+			klog.ErrorS(err, "Failed to parse HAMI_NODELOCK_RETRY_TIMEOUT, using default", "duration", NodeLockRetryTimeout)
+		} else {
+			NodeLockRetryTimeout = d
+			klog.InfoS("Node lock retry timeout set from environment variable", "duration", d)
+		}
+	}
 }
 
 func SetNodeLock(nodeName string, lockname string, pods *corev1.Pod) error {
@@ -205,8 +224,39 @@ func ReleaseNodeLock(nodeName string, lockname string, pod *corev1.Pod, skipNode
 	return nil
 }
 
+// LockNode attempts to acquire a node lock, retrying with backoff when the
+// node is locked by a valid (non-expired, non-dangling) pod. This allows
+// concurrent bind operations (e.g., from coscheduling PodGroups) to succeed
+// sequentially on the same node rather than failing immediately.
 func LockNode(nodeName string, lockname string, pods *corev1.Pod) error {
-	ctx := context.Background()
+	ctx, cancel := context.WithTimeout(context.Background(), NodeLockRetryTimeout)
+	defer cancel()
+
+	var lastErr error
+	err := wait.PollUntilContextCancel(ctx, time.Second, true, func(ctx context.Context) (bool, error) {
+		lastErr = tryLockNode(ctx, nodeName, lockname, pods)
+		if lastErr == nil {
+			return true, nil
+		}
+		if errors.Is(lastErr, errNodeLocked) {
+			klog.V(5).InfoS("Node locked by valid pod, retrying", "node", nodeName)
+			return false, nil
+		}
+		return false, lastErr
+	})
+	if err != nil {
+		if lastErr != nil {
+			return lastErr
+		}
+		return fmt.Errorf("failed to lock node %s: timed out after %v", nodeName, NodeLockRetryTimeout)
+	}
+	return nil
+}
+
+// tryLockNode makes a single attempt to acquire the node lock.
+// It returns nil on success, errNodeLocked when the lock is held by a valid
+// pod (retryable), or a regular error for non-retryable failures.
+func tryLockNode(ctx context.Context, nodeName string, lockname string, pods *corev1.Pod) error {
 	node, err := client.GetClient().CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
 	if err != nil {
 		return err
@@ -219,13 +269,18 @@ func LockNode(nodeName string, lockname string, pods *corev1.Pod) error {
 		return err
 	}
 
+	// Lock held by the same pod → idempotent success.
+	if ns == pods.Namespace && previousPodName == pods.Name {
+		return nil
+	}
+
 	var skipOwnerCheck = false
 	if time.Since(lockTime) > NodeLockTimeout {
 		klog.InfoS("Node lock expired", "node", nodeName, "lockTime", lockTime, "timeout", NodeLockTimeout)
 		skipOwnerCheck = true
 	} else
 	// Check dangling nodeLock
-	if ns != "" && previousPodName != "" && (ns != pods.Namespace || previousPodName != pods.Name) {
+	if ns != "" && previousPodName != "" {
 		if _, err := client.GetClient().CoreV1().Pods(ns).Get(ctx, previousPodName, metav1.GetOptions{}); err != nil {
 			if !apierrors.IsNotFound(err) {
 				klog.ErrorS(err, "Failed to get pod of NodeLock", "podName", previousPodName, "namespace", ns)
@@ -245,7 +300,7 @@ func LockNode(nodeName string, lockname string, pods *corev1.Pod) error {
 		return SetNodeLock(nodeName, lockname, pods)
 	}
 
-	return fmt.Errorf("node %s has been locked within %v", nodeName, NodeLockTimeout)
+	return errNodeLocked
 }
 
 func ParseNodeLock(value string) (lockTime time.Time, ns, name string, err error) {

pkg/util/nodelock/nodelock_test.go

@@ -32,6 +32,14 @@ import (
 
 func Test_LockNode(t *testing.T) {
 	client.KubeClient = fake.NewClientset()
+
+	// Use a short retry timeout so tests that expect lock contention complete quickly.
+	originalRetryTimeout := NodeLockRetryTimeout
+	NodeLockRetryTimeout = 500 * time.Millisecond
+	defer func() {
+		NodeLockRetryTimeout = originalRetryTimeout
+	}()
+
 	type args struct {
 		nodeName func() string
 		lockname string
@@ -58,7 +66,7 @@ func Test_LockNode(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "node has been locked",
+			name: "node has been locked by another pod",
 			args: args{
 				nodeName: func() string {
 					name := "worker-1"
@@ -67,11 +75,15 @@ func Test_LockNode(t *testing.T) {
 							Name: name,
 							Annotations: map[string]string{
 								NodeLockKey: GenerateNodeLockKeyByPod(&corev1.Pod{
-									ObjectMeta: metav1.ObjectMeta{Name: "hami", Namespace: "hami-ns"},
+									ObjectMeta: metav1.ObjectMeta{Name: "other-pod", Namespace: "other-ns"},
 								}),
 							},
 						},
 					}, metav1.CreateOptions{})
+					// The lock-holding pod must exist to avoid dangling detection.
+					client.KubeClient.CoreV1().Pods("other-ns").Create(context.TODO(), &corev1.Pod{
+						ObjectMeta: metav1.ObjectMeta{Name: "other-pod", Namespace: "other-ns"},
+					}, metav1.CreateOptions{})
 					return name
 				},
 				pods: &corev1.Pod{
@@ -139,16 +151,23 @@ func Test_LockNode(t *testing.T) {
 func TestLockNodeWithTimeout(t *testing.T) {
 	client.KubeClient = fake.NewClientset()
 
-	// Set a custom timeout for testing
+	// Set a custom timeout for testing.
 	originalTimeout := NodeLockTimeout
 	NodeLockTimeout = time.Minute * 2
 	defer func() {
 		NodeLockTimeout = originalTimeout
 	}()
 
+	// Use a short retry timeout so the test completes quickly.
+	originalRetryTimeout := NodeLockRetryTimeout
+	NodeLockRetryTimeout = 500 * time.Millisecond
+	defer func() {
+		NodeLockRetryTimeout = originalRetryTimeout
+	}()
+
 	nodeName := "test-node-timeout"
 
-	// Create a node with a fresh lock (should not be expired)
+	// Create a node with a fresh lock (should not be expired).
 	freshLockTime := time.Now().Format(time.RFC3339)
 	testNamespace := "test-ns"
 	testPodName := "test-pod"
@@ -163,7 +182,7 @@ func TestLockNodeWithTimeout(t *testing.T) {
 		},
 	}, metav1.CreateOptions{})
 
-	// Pod must exist to avoid dangling node lock
+	// Pod must exist to avoid dangling node lock.
 	client.KubeClient.CoreV1().Pods(testNamespace).Create(context.TODO(), &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      testPodName,
@@ -178,17 +197,21 @@ func TestLockNodeWithTimeout(t *testing.T) {
 		},
 	}
 
-	// Try to lock the node again - this should trigger line 130
+	start := time.Now()
 	err := LockNode(nodeName, "", pod)
+	elapsed := time.Since(start)
 
-	// Verify the error contains the NodeLockTimeout value
 	if err == nil {
 		t.Fatal("Expected error but got nil")
 	}
 
-	expectedError := "has been locked within 2m0s"
-	if !strings.Contains(err.Error(), expectedError) {
-		t.Errorf("Expected error to contain '%s', but got: %v", expectedError, err)
+	// Should have retried for roughly NodeLockRetryTimeout before giving up.
+	if elapsed < 400*time.Millisecond {
+		t.Errorf("LockNode returned too quickly (%v), expected retry for ~%v", elapsed, NodeLockRetryTimeout)
+	}
+
+	if !strings.Contains(err.Error(), "locked by a valid pod") {
+		t.Errorf("Expected errNodeLocked sentinel, got: %v", err)
 	}
 }
 
@@ -501,6 +524,92 @@ func TestGeneratePodNamespaceName(t *testing.T) {
 	}
 }
 
+// TestLockNodeRetrySuccess verifies that LockNode retries and succeeds when a
+// lock is released by the holder before the retry timeout expires.
+func TestLockNodeRetrySuccess(t *testing.T) {
+	client.KubeClient = fake.NewClientset()
+	nodeLocks = newNodeLockManager()
+
+	originalRetryTimeout := NodeLockRetryTimeout
+	NodeLockRetryTimeout = 5 * time.Second
+	defer func() {
+		NodeLockRetryTimeout = originalRetryTimeout
+	}()
+
+	nodeName := "retry-success-node"
+	holderPod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "holder-pod", Namespace: "holder-ns"},
+	}
+	waiterPod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "waiter-pod", Namespace: "waiter-ns"},
+	}
+
+	// Create node and holder pod.
+	client.KubeClient.CoreV1().Nodes().Create(context.TODO(), &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: nodeName, Annotations: map[string]string{}},
+	}, metav1.CreateOptions{})
+	client.KubeClient.CoreV1().Pods("holder-ns").Create(context.TODO(), holderPod, metav1.CreateOptions{})
+
+	// Holder acquires the lock.
+	if err := LockNode(nodeName, "", holderPod); err != nil {
+		t.Fatalf("Holder failed to acquire lock: %v", err)
+	}
+
+	// Release the lock after a short delay so the waiter can acquire it.
+	go func() {
+		time.Sleep(500 * time.Millisecond)
+		if err := ReleaseNodeLock(nodeName, "", holderPod, false); err != nil {
+			t.Errorf("Holder failed to release lock: %v", err)
+		}
+	}()
+
+	// Waiter should retry and eventually succeed.
+	start := time.Now()
+	if err := LockNode(nodeName, "", waiterPod); err != nil {
+		t.Fatalf("Waiter failed to acquire lock after retry: %v", err)
+	}
+	elapsed := time.Since(start)
+
+	if elapsed < 400*time.Millisecond {
+		t.Errorf("Waiter acquired lock too quickly (%v), expected some retry delay", elapsed)
+	}
+	if elapsed > 3*time.Second {
+		t.Errorf("Waiter took too long (%v), expected lock within ~1-2s", elapsed)
+	}
+}
+
+// TestLockNodeIdempotent verifies that LockNode succeeds immediately when the
+// lock is already held by the same pod (idempotent behavior).
+func TestLockNodeIdempotent(t *testing.T) {
+	client.KubeClient = fake.NewClientset()
+	nodeLocks = newNodeLockManager()
+
+	nodeName := "idempotent-node"
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "my-pod", Namespace: "my-ns"},
+	}
+
+	client.KubeClient.CoreV1().Nodes().Create(context.TODO(), &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: nodeName, Annotations: map[string]string{}},
+	}, metav1.CreateOptions{})
+
+	// First lock.
+	if err := LockNode(nodeName, "", pod); err != nil {
+		t.Fatalf("First LockNode failed: %v", err)
+	}
+
+	// Second lock by the same pod should succeed immediately.
+	start := time.Now()
+	if err := LockNode(nodeName, "", pod); err != nil {
+		t.Fatalf("Second LockNode (idempotent) failed: %v", err)
+	}
+	elapsed := time.Since(start)
+
+	if elapsed > 200*time.Millisecond {
+		t.Errorf("Idempotent lock took too long (%v), should be immediate", elapsed)
+	}
+}
+
 // TestSimulateRetryStorm verifies if the Backoff strategy is using exponential backoff.
 func TestSimulateRetryStorm(t *testing.T) {
 	tests := []struct {