fix(nodelock): serialize expired-lock recovery

pkg/util/nodelock/nodelock.go

@@ -61,6 +61,9 @@ var (
 		Factor:   2.0,
 		Jitter:   0.5,
 	}
+
+	// beforeLockNodeMutexHook is an optional test hook called right before acquiring the per-node mutex in LockNode.
+	beforeLockNodeMutexHook func(nodeName string, pods *corev1.Pod)
 )
 
 // nodeLockManager manages locks on a per-node basis to allow concurrent
@@ -127,11 +130,18 @@ func setupNodeLockTimeout() {
 }
 
 func SetNodeLock(nodeName string, lockname string, pods *corev1.Pod) error {
+	if pods == nil {
+		return fmt.Errorf("cannot set node lock: pod is nil")
+	}
 	// Acquire per-node lock instead of global lock
 	nodeLock := nodeLocks.getLock(nodeName)
 	nodeLock.Lock()
 	defer nodeLock.Unlock()
 
+	return setNodeLockLocked(nodeName, lockname, pods)
+}
+
+func setNodeLockLocked(nodeName string, lockname string, pods *corev1.Pod) error {
 	ctx := context.Background()
 	node, err := client.GetClient().CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
 	if err != nil {
@@ -181,6 +191,10 @@ func ReleaseNodeLock(nodeName string, lockname string, pod *corev1.Pod, skipNode
 	nodeLock.Lock()
 	defer nodeLock.Unlock()
 
+	return releaseNodeLockLocked(nodeName, lockname, pod, skipNodeLockOwnerCheck)
+}
+
+func releaseNodeLockLocked(nodeName string, lockname string, pod *corev1.Pod, skipNodeLockOwnerCheck bool) error {
 	ctx := context.Background()
 	node, err := client.GetClient().CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
 	if err != nil {
@@ -239,13 +253,24 @@ func ReleaseNodeLock(nodeName string, lockname string, pod *corev1.Pod, skipNode
 }
 
 func LockNode(nodeName string, lockname string, pods *corev1.Pod) error {
+	if pods == nil {
+		return fmt.Errorf("cannot lock node: pod is nil")
+	}
+	if beforeLockNodeMutexHook != nil {
+		beforeLockNodeMutexHook(nodeName, pods)
+	}
+	// Acquire per-node lock instead of global lock
+	nodeLock := nodeLocks.getLock(nodeName)
+	nodeLock.Lock()
+	defer nodeLock.Unlock()
+
 	ctx := context.Background()
 	node, err := client.GetClient().CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
 	if err != nil {
 		return err
 	}
 	if _, ok := node.Annotations[NodeLockKey]; !ok {
-		return SetNodeLock(nodeName, lockname, pods)
+		return setNodeLockLocked(nodeName, lockname, pods)
 	}
 	lockTime, ns, previousPodName, err := ParseNodeLock(node.Annotations[NodeLockKey])
 	if err != nil {
@@ -265,9 +290,8 @@ func LockNode(nodeName string, lockname string, pods *corev1.Pod) error {
 		// become schedulable. Treat this as already acquired.
 		klog.V(4).InfoS("Node lock already held by this pod, treating as acquired", "node", nodeName, "podName", pods.Name)
 		return nil
-	} else
-	// Check dangling nodeLock
-	if ns != "" && previousPodName != "" {
+	} else if ns != "" && previousPodName != "" {
+		// Check dangling nodeLock
 		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)
@@ -279,12 +303,12 @@ func LockNode(nodeName string, lockname string, pods *corev1.Pod) error {
 	}
 
 	if skipOwnerCheck {
-		err = ReleaseNodeLock(nodeName, lockname, pods, true)
+		err = releaseNodeLockLocked(nodeName, lockname, pods, true)
 		if err != nil {
 			klog.ErrorS(err, "Failed to release node lock", "node", nodeName)
 			return err
 		}
-		return SetNodeLock(nodeName, lockname, pods)
+		return setNodeLockLocked(nodeName, lockname, pods)
 	}
 
 	return fmt.Errorf("node %s has been locked within %v: %w", nodeName, NodeLockTimeout, ErrNodeLockContention)