Fix deadlock while reattaching volume

pkg/kubevirt/client.go

@@ -67,7 +67,9 @@ type Client interface {
 	RemoveVolumeFromVM(ctx context.Context, namespace string, vmName string, hotPlugRequest *kubevirtv1.RemoveVolumeOptions) error
 	RemoveVolumeFromVMI(ctx context.Context, namespace string, vmName string, hotPlugRequest *kubevirtv1.RemoveVolumeOptions) error
 	EnsureVolumeAvailable(ctx context.Context, namespace, vmName, volumeName string, timeout time.Duration) error
+	EnsureVolumeAvailableVM(ctx context.Context, namespace, name, volumeName string) (bool, error)
 	EnsureVolumeRemoved(ctx context.Context, namespace, vmName, volumeName string, timeout time.Duration) error
+	EnsureVolumeRemovedVM(ctx context.Context, namespace, name, volumeName string) (bool, error)
 	EnsureSnapshotReady(ctx context.Context, namespace, name string, timeout time.Duration) error
 	EnsureControllerResize(ctx context.Context, namespace, claimName string, timeout time.Duration) error
 	CreateVolumeSnapshot(ctx context.Context, namespace, name, claimName, snapshotClassName string) (*snapshotv1.VolumeSnapshot, error)
@@ -221,7 +223,7 @@ func (c *client) EnsureVolumeAvailable(ctx context.Context, namespace, vmName, v
 		}
 		for _, volume := range vmi.Status.VolumeStatus {
 			if volume.Name == volumeName && volume.Phase == kubevirtv1.VolumeReady {
-				return ensureVolumeAvailableVM(ctx, c, namespace, vmName, volumeName)
+				return c.EnsureVolumeAvailableVM(ctx, namespace, vmName, volumeName)
 			}
 		}
 
@@ -238,15 +240,15 @@ func (c *client) EnsureVolumeRemoved(ctx context.Context, namespace, vmName, vol
 				return false, err
 			}
 			// No VMI, volume considered removed if it's not on the VM
-			return ensureVolumeRemovedVM(ctx, c, namespace, vmName, volumeName)
+			return c.EnsureVolumeRemovedVM(ctx, namespace, vmName, volumeName)
 		}
 		for _, volume := range vmi.Status.VolumeStatus {
 			if volume.Name == volumeName {
 				return false, nil
 			}
 		}
 
-		return ensureVolumeRemovedVM(ctx, c, namespace, vmName, volumeName)
+		return c.EnsureVolumeRemovedVM(ctx, namespace, vmName, volumeName)
 	})
 }
 
@@ -581,7 +583,7 @@ func appendVolumeSnapshotInfraTenantMapping(mapping *InfraTenantStorageSnapshotM
 	return mapping
 }
 
-func ensureVolumeRemovedVM(ctx context.Context, c *client, namespace, name, volumeName string) (bool, error) {
+func (c *client) EnsureVolumeRemovedVM(ctx context.Context, namespace, name, volumeName string) (bool, error) {
 	vm, err := c.virtClient.KubevirtV1().VirtualMachines(namespace).Get(ctx, name, metav1.GetOptions{})
 	if err != nil {
 		if !errors.IsNotFound(err) {
@@ -603,7 +605,7 @@ func ensureVolumeRemovedVM(ctx context.Context, c *client, namespace, name, volu
 	return true, nil
 }
 
-func ensureVolumeAvailableVM(ctx context.Context, c *client, namespace, name, volumeName string) (bool, error) {
+func (c *client) EnsureVolumeAvailableVM(ctx context.Context, namespace, name, volumeName string) (bool, error) {
 	vm, err := c.virtClient.KubevirtV1().VirtualMachines(namespace).Get(ctx, name, metav1.GetOptions{})
 	if err != nil {
 		if !errors.IsNotFound(err) {

pkg/service/controller.go

@@ -355,6 +355,16 @@ func (c *ControllerService) ControllerPublishVolume(
 	// Determine BUS type
 	bus := req.VolumeContext[busParameter]
 
+	// Fast-path: nothing to do if the volume is already attached
+	attached, err := c.virtClient.EnsureVolumeAvailableVM(ctx, c.infraClusterNamespace, vmName, dvName)
+	if err != nil {
+		return nil, err
+	}
+	if attached {
+		klog.V(3).Infof("Volume %s already attached to VM %s - skipping hot-plug", dvName, vmName)
+		return &csi.ControllerPublishVolumeResponse{}, nil
+	}
+
 	// hotplug DataVolume to VM
 	klog.V(3).Infof("Start attaching DataVolume %s to VM %s. Volume name: %s. Serial: %s. Bus: %s", dvName, vmName, dvName, serial, bus)
 
@@ -473,6 +483,16 @@ func (c *ControllerService) ControllerUnpublishVolume(ctx context.Context, req *
 		return nil, nil
 	}
 
+	// Fast-path: if the disk is no longer on the VM, succeed immediately
+	notPresent, err := c.virtClient.EnsureVolumeRemovedVM(ctx, c.infraClusterNamespace, vmName, dvName)
+	if err != nil {
+		return nil, err
+	}
+	if notPresent {
+		klog.V(3).Infof("Volume %s already detached from VM %s – skipping hot-unplug", dvName, vmName)
+		return &csi.ControllerUnpublishVolumeResponse{}, nil
+	}
+
 	if err := wait.ExponentialBackoff(wait.Backoff{
 		Duration: time.Second,
 		Steps:    5,

pkg/service/controller_test.go

@@ -1006,6 +1006,14 @@ func (c *ControllerClientMock) EnsureControllerResize(_ context.Context, namespa
 	return nil
 }
 
+func (c *ControllerClientMock) EnsureVolumeAvailableVM(_ context.Context, namespace, vmName, volName string) (bool, error) {
+	return false, nil
+}
+
+func (c *ControllerClientMock) EnsureVolumeRemovedVM(_ context.Context, namespace, vmName, volName string) (bool, error) {
+	return false, nil
+}
+
 func (c *ControllerClientMock) CreateVolumeSnapshot(ctx context.Context, namespace, name, claimName, snapshotClassName string) (*snapshotv1.VolumeSnapshot, error) {
 	if c.FailCreateSnapshot {
 		return nil, errors.New("CreateVolumeSnapshot failed")
@@ -1084,3 +1092,96 @@ func (c *vmiUnplugCapturingClient) RemoveVolumeFromVMI(_ context.Context, namesp
 func getKey(namespace, name string) string {
 	return fmt.Sprintf("%s/%s", namespace, name)
 }
+
+var _ = Describe("Fast-path publish / unpublish", func() {
+	It("should skip hot-plug when the disk is already attached", func() {
+		cli := &attachSkipClient{ControllerClientMock: &ControllerClientMock{}}
+
+		// fake DataVolume so GetDataVolume succeeds
+		cli.datavolumes = map[string]*cdiv1.DataVolume{
+			getKey(testInfraNamespace, testVolumeName): {
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      testVolumeName,
+					Namespace: testInfraNamespace,
+				},
+			},
+		}
+
+		ctrl := &ControllerService{
+			virtClient:              cli,
+			infraClusterNamespace:   testInfraNamespace,
+			infraClusterLabels:      testInfraLabels,
+			storageClassEnforcement: storageClassEnforcement,
+		}
+
+		_, err := ctrl.ControllerPublishVolume(
+			context.TODO(), getPublishVolumeRequest())
+		Expect(err).ToNot(HaveOccurred())
+
+		Expect(cli.addCnt).To(Equal(0))    // hot-plug was skipped
+		Expect(cli.ensureCnt).To(Equal(1)) // fast-path check executed
+	})
+
+	It("should skip hot-unplug when the disk is already detached", func() {
+		cli := &detachSkipClient{ControllerClientMock: &ControllerClientMock{}}
+
+		ctrl := &ControllerService{
+			virtClient:              cli,
+			infraClusterNamespace:   testInfraNamespace,
+			infraClusterLabels:      testInfraLabels,
+			storageClassEnforcement: storageClassEnforcement,
+		}
+
+		_, err := ctrl.ControllerUnpublishVolume(
+			context.TODO(), getUnpublishVolumeRequest())
+		Expect(err).ToNot(HaveOccurred())
+
+		Expect(cli.removeCnt).To(Equal(0)) // unplug skipped
+		Expect(cli.ensureCnt).To(Equal(1)) // fast-path check executed
+	})
+})
+
+// returns "already attached", records calls
+type attachSkipClient struct {
+	*ControllerClientMock
+	addCnt    int // AddVolumeToVM called
+	ensureCnt int // EnsureVolumeAvailable(VM) called
+}
+
+// the driver never calls AddVolumeToVM when it detects an
+// already-attached disk, but we still implement it for safety.
+func (c *attachSkipClient) AddVolumeToVM(_ context.Context,
+	ns, vm string, opts *kubevirtv1.AddVolumeOptions) error {
+
+	c.addCnt++
+	return nil
+}
+
+// EnsureVolumeAvailableVM is the test hook for fast-path detection.
+// Return (true,nil) to say "already attached".
+func (c *attachSkipClient) EnsureVolumeAvailableVM(_ context.Context,
+	ns, vm, dv string) (bool, error) {
+
+	c.ensureCnt++
+	return true, nil
+}
+
+type detachSkipClient struct {
+	*ControllerClientMock
+	removeCnt int // RemoveVolumeFromVM called
+	ensureCnt int // EnsureVolumeRemoved(VM) called
+}
+
+func (c *detachSkipClient) RemoveVolumeFromVM(_ context.Context,
+	ns, vm string, opts *kubevirtv1.RemoveVolumeOptions) error {
+
+	c.removeCnt++
+	return nil
+}
+
+func (c *detachSkipClient) EnsureVolumeRemovedVM(_ context.Context,
+	ns, vm, dv string) (bool, error) {
+
+	c.ensureCnt++
+	return true, nil // volume absent
+}

sanity/sanity_test.go

@@ -209,6 +209,14 @@ func (k *fakeKubeVirtClient) EnsureVolumeAvailable(_ context.Context, namespace,
 	return nil
 }
 
+func (c *fakeKubeVirtClient) EnsureVolumeAvailableVM(_ context.Context, namespace, vmName, volName string) (bool, error) {
+	return false, nil
+}
+
+func (c *fakeKubeVirtClient) EnsureVolumeRemovedVM(_ context.Context, namespace, vmName, volName string) (bool, error) {
+	return false, nil
+}
+
 func (k *fakeKubeVirtClient) EnsureVolumeRemoved(_ context.Context, namespace, vmName, volumeName string, timeout time.Duration) error {
 	return nil
 }