diff --git a/pkg/util/nodelock/nodelock.go b/pkg/util/nodelock/nodelock.go index 0ba3998fff..cfa57f8b5d 100644 --- a/pkg/util/nodelock/nodelock.go +++ b/pkg/util/nodelock/nodelock.go @@ -138,17 +138,24 @@ func SetNodeLock(nodeName string, lockname string, pods *corev1.Pod) error { return err } if _, ok := node.Annotations[NodeLockKey]; ok { - return fmt.Errorf("node %s is locked", nodeName) + return fmt.Errorf("node %s is locked: %w", nodeName, ErrNodeLockContention) } + lockOwner := NodeLockSep + GeneratePodNamespaceName(pods, NodeLockSep) err = retry.OnError(DefaultStrategy, func(err error) bool { - // Retry on any error - return true + return !IsNodeLockContention(err) }, func() error { node, err = client.GetClient().CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{}) if err != nil { klog.ErrorS(err, "Failed to get node when retry to patch", "node", nodeName) return err } + lockStr, ok := node.Annotations[NodeLockKey] + if ok && strings.Contains(lockStr, NodeLockSep) && strings.HasSuffix(lockStr, lockOwner) { + return nil + } + if ok { + return fmt.Errorf("node %s is locked: %w", nodeName, ErrNodeLockContention) + } patchData := fmt.Sprintf(`{"metadata":{"annotations":{"%s":"%s"},"resourceVersion":"%s"}}`, NodeLockKey, GenerateNodeLockKeyByPod(pods), node.ResourceVersion) _, err = client.GetClient().CoreV1().Nodes().Patch(ctx, nodeName, types.MergePatchType, []byte(patchData), metav1.PatchOptions{}) if err != nil { @@ -184,12 +191,14 @@ func ReleaseNodeLock(nodeName string, lockname string, pod *corev1.Pod, skipNode if !ok { return nil } + lockOwner := NodeLockSep + GeneratePodNamespaceName(pod, NodeLockSep) // Keep backward compatibility with the legacy format, which is simply a timestamp - if !skipNodeLockOwnerCheck && strings.Contains(lockStr, NodeLockSep) && !strings.HasSuffix(lockStr, NodeLockSep+GeneratePodNamespaceName(pod, NodeLockSep)) { + if !skipNodeLockOwnerCheck && strings.Contains(lockStr, NodeLockSep) && !strings.HasSuffix(lockStr, lockOwner) { klog.InfoS("NodeLock is not set by this pod", NodeLockKey, lockStr, "podName", pod.Name, "podNamespace", pod.Namespace) return nil } + released := false err = retry.OnError(DefaultStrategy, func(err error) bool { // Retry on any error return true @@ -199,19 +208,33 @@ func ReleaseNodeLock(nodeName string, lockname string, pod *corev1.Pod, skipNode klog.ErrorS(err, "Failed to get node when retry to patch", "node", nodeName) return err } + currentLock, ok := node.Annotations[NodeLockKey] + if !ok { + return nil + } + if skipNodeLockOwnerCheck || !strings.Contains(currentLock, NodeLockSep) { + if currentLock != lockStr { + return nil + } + } else if !strings.HasSuffix(currentLock, lockOwner) { + return nil + } patchData := fmt.Sprintf(`{"metadata":{"annotations":{"%s":null},"resourceVersion":"%s"}}`, NodeLockKey, node.ResourceVersion) _, err = client.GetClient().CoreV1().Nodes().Patch(ctx, nodeName, types.MergePatchType, []byte(patchData), metav1.PatchOptions{}) if err != nil { klog.ErrorS(err, "Failed to patch node when retry to patch", "node", nodeName) return err } + released = true return nil }) if err != nil { return fmt.Errorf("failed to release node lock (node=%s, retry strategy=%+v): %w", nodeName, DefaultStrategy, err) } - klog.InfoS("Node lock released", "node", nodeName, "podName", pod.Name) + if released { + klog.InfoS("Node lock released", "node", nodeName, "podName", pod.Name) + } return nil } diff --git a/pkg/util/nodelock/nodelock_test.go b/pkg/util/nodelock/nodelock_test.go index b4a5d56179..80a8196a95 100644 --- a/pkg/util/nodelock/nodelock_test.go +++ b/pkg/util/nodelock/nodelock_test.go @@ -17,19 +17,235 @@ limitations under the License. package nodelock import ( - "context" // Added for the new test + "context" + "errors" "runtime" "strings" "testing" "time" corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" // Added for the new test + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8sruntime "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" "github.com/Project-HAMi/HAMi/pkg/util/client" ) +func TestSetNodeLockPreservesConcurrentLockAfterConflict(t *testing.T) { + nodeLocks = newNodeLockManager() + nodeName := "node-set-conflict" + podA := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-a", Namespace: "ns"}} + podB := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-b", Namespace: "ns"}} + holderB := GenerateNodeLockKeyByPod(podB) + clientSet := fake.NewClientset(&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}}) + client.KubeClient = clientSet + + getCalls := 0 + clientSet.PrependReactor("get", "nodes", func(k8stesting.Action) (bool, k8sruntime.Object, error) { + getCalls++ + if getCalls < 3 { + return false, nil, nil + } + return true, &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: nodeName, + Annotations: map[string]string{NodeLockKey: holderB}, + }}, nil + }) + patchCalls := 0 + clientSet.PrependReactor("patch", "nodes", func(k8stesting.Action) (bool, k8sruntime.Object, error) { + patchCalls++ + return true, nil, apierrors.NewConflict(schema.GroupResource{Resource: "nodes"}, nodeName, errors.New("simulated concurrent lock")) + }) + + err := SetNodeLock(nodeName, "", podA) + if !IsNodeLockContention(err) { + t.Fatalf("SetNodeLock() error = %v, want node lock contention", err) + } + node, err := clientSet.CoreV1().Nodes().Get(context.Background(), nodeName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("get node: %v", err) + } + if got := node.Annotations[NodeLockKey]; got != holderB { + t.Fatalf("node lock = %q, want concurrent holder %q", got, holderB) + } + if patchCalls != 1 { + t.Fatalf("patch calls = %d, want 1", patchCalls) + } +} + +func TestSetNodeLockSucceedsAfterLostPatchResponse(t *testing.T) { + nodeLocks = newNodeLockManager() + nodeName := "node-set-lost-response" + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-a", Namespace: "ns"}} + lockStr := "2026-08-01T06:00:00Z,ns,pod-a" + clientSet := fake.NewClientset(&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}}) + client.KubeClient = clientSet + + getCalls := 0 + clientSet.PrependReactor("get", "nodes", func(k8stesting.Action) (bool, k8sruntime.Object, error) { + getCalls++ + if getCalls < 3 { + return false, nil, nil + } + return true, &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: nodeName, + Annotations: map[string]string{NodeLockKey: lockStr}, + }}, nil + }) + patchCalls := 0 + clientSet.PrependReactor("patch", "nodes", func(k8stesting.Action) (bool, k8sruntime.Object, error) { + patchCalls++ + return true, nil, errors.New("simulated lost patch response") + }) + + if err := SetNodeLock(nodeName, "", pod); err != nil { + t.Fatalf("SetNodeLock() error = %v, want nil", err) + } + if patchCalls != 1 { + t.Fatalf("patch calls = %d, want 1", patchCalls) + } +} + +func TestReleaseNodeLockPreservesConcurrentLockAfterConflict(t *testing.T) { + nodeLocks = newNodeLockManager() + nodeName := "node-release-conflict" + podA := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-a", Namespace: "ns"}} + podB := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-b", Namespace: "ns"}} + holderA := GenerateNodeLockKeyByPod(podA) + holderB := GenerateNodeLockKeyByPod(podB) + clientSet := fake.NewClientset(&corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: nodeName, + Annotations: map[string]string{NodeLockKey: holderA}, + }}) + client.KubeClient = clientSet + + getCalls := 0 + clientSet.PrependReactor("get", "nodes", func(k8stesting.Action) (bool, k8sruntime.Object, error) { + getCalls++ + if getCalls < 3 { + return false, nil, nil + } + return true, &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: nodeName, + Annotations: map[string]string{NodeLockKey: holderB}, + }}, nil + }) + patchCalls := 0 + clientSet.PrependReactor("patch", "nodes", func(k8stesting.Action) (bool, k8sruntime.Object, error) { + patchCalls++ + return true, nil, apierrors.NewConflict(schema.GroupResource{Resource: "nodes"}, nodeName, errors.New("simulated concurrent lock")) + }) + + if err := ReleaseNodeLock(nodeName, "", podA, false); err != nil { + t.Fatalf("ReleaseNodeLock() error = %v, want nil", err) + } + node, err := clientSet.CoreV1().Nodes().Get(context.Background(), nodeName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("get node: %v", err) + } + if got := node.Annotations[NodeLockKey]; got != holderB { + t.Fatalf("node lock = %q, want concurrent holder %q", got, holderB) + } + if patchCalls != 1 { + t.Fatalf("patch calls = %d, want 1", patchCalls) + } +} + +func TestReleaseNodeLockPreservesReplacedLegacyLockAfterConflict(t *testing.T) { + nodeLocks = newNodeLockManager() + nodeName := "node-release-legacy-conflict" + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-a", Namespace: "ns"}} + initialLock := "2026-08-01T06:00:00Z" + replacedLock := "2026-08-01T06:00:01Z" + clientSet := fake.NewClientset(&corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: nodeName, + Annotations: map[string]string{NodeLockKey: initialLock}, + }}) + client.KubeClient = clientSet + + getCalls := 0 + clientSet.PrependReactor("get", "nodes", func(k8stesting.Action) (bool, k8sruntime.Object, error) { + getCalls++ + if getCalls < 3 { + return false, nil, nil + } + return true, &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: nodeName, + Annotations: map[string]string{NodeLockKey: replacedLock}, + }}, nil + }) + patchCalls := 0 + clientSet.PrependReactor("patch", "nodes", func(k8stesting.Action) (bool, k8sruntime.Object, error) { + patchCalls++ + return true, nil, apierrors.NewConflict(schema.GroupResource{Resource: "nodes"}, nodeName, errors.New("simulated concurrent legacy lock")) + }) + + if err := ReleaseNodeLock(nodeName, "", pod, false); err != nil { + t.Fatalf("ReleaseNodeLock() error = %v, want nil", err) + } + node, err := clientSet.CoreV1().Nodes().Get(context.Background(), nodeName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("get node: %v", err) + } + if got := node.Annotations[NodeLockKey]; got != replacedLock { + t.Fatalf("node lock = %q, want concurrent legacy lock %q", got, replacedLock) + } + if patchCalls != 1 { + t.Fatalf("patch calls = %d, want 1", patchCalls) + } +} + +func TestReleaseNodeLockReleasesRestampedLockForSamePod(t *testing.T) { + nodeLocks = newNodeLockManager() + nodeName := "node-release-restamped" + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod-a", Namespace: "ns"}} + initialLock := "2026-07-29T13:00:00Z,ns,pod-a" + restampedLock := "2026-07-29T13:00:01Z,ns,pod-a" + clientSet := fake.NewClientset(&corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: nodeName, + Annotations: map[string]string{NodeLockKey: initialLock}, + }}) + client.KubeClient = clientSet + + getCalls := 0 + clientSet.PrependReactor("get", "nodes", func(k8stesting.Action) (bool, k8sruntime.Object, error) { + getCalls++ + if getCalls != 3 { + return false, nil, nil + } + return true, &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: nodeName, + Annotations: map[string]string{NodeLockKey: restampedLock}, + }}, nil + }) + patchCalls := 0 + clientSet.PrependReactor("patch", "nodes", func(k8stesting.Action) (bool, k8sruntime.Object, error) { + patchCalls++ + if patchCalls == 1 { + return true, nil, apierrors.NewConflict(schema.GroupResource{Resource: "nodes"}, nodeName, errors.New("simulated restamp")) + } + return false, nil, nil + }) + + if err := ReleaseNodeLock(nodeName, "", pod, false); err != nil { + t.Fatalf("ReleaseNodeLock() error = %v, want nil", err) + } + node, err := clientSet.CoreV1().Nodes().Get(context.Background(), nodeName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("get node: %v", err) + } + if _, ok := node.Annotations[NodeLockKey]; ok { + t.Fatalf("node lock = %q, want removed", node.Annotations[NodeLockKey]) + } + if patchCalls != 2 { + t.Fatalf("patch calls = %d, want 2", patchCalls) + } +} + func Test_LockNode(t *testing.T) { client.KubeClient = fake.NewClientset() type args struct {