Merge pull request #6 from serverscom/fix-multi-volumes

Fix long mount and multi volumes mount

Author: olegy89

deploy/rbac.yaml

@@ -114,6 +114,9 @@ rules:
   - apiGroups: [""]
     resources: ["pods"]
     verbs: ["get", "list", "watch"]
+  - apiGroups: ["storage.k8s.io"]
+    resources: ["volumeattributesclasses"]
+    verbs: ["get", "list", "watch"]
 
 ---
 

pkg/csi/node.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sync"
 	"time"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
@@ -24,6 +25,7 @@ type NodeService struct {
 	nodeID       string
 	iscsiManager iscsi.ISCSIManager
 	mountManager mount.MountManager
+	stagingMu    sync.Map // map[volumeID → *sync.Mutex], serializes concurrent stage calls per volume
 }
 
 // NewNodeService creates a new node service
@@ -39,6 +41,15 @@ func NewNodeService(nodeID string) *NodeService {
 func (s *NodeService) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) {
 	klog.V(2).InfoS("NodeStageVolume called", "volume_id", req.GetVolumeId())
 
+	// Serialize concurrent staging calls for the same volume. Kubelet may retry
+	// NodeStageVolume while a previous call is still formatting the disk (mkfs can
+	// take minutes). Without this lock the second call races FormatAndMountDevice
+	// and hits "already mounted" once the first call finishes.
+	mu, _ := s.stagingMu.LoadOrStore(req.GetVolumeId(), &sync.Mutex{})
+	lock := mu.(*sync.Mutex)
+	lock.Lock()
+	defer lock.Unlock()
+
 	// Extract iSCSI connection information from publish context
 	publishContext := req.GetPublishContext()
 	klog.V(2).InfoS("PublishContext", "context", util.MaskSensitiveMap(publishContext))
@@ -142,6 +153,16 @@ func (s *NodeService) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 		}
 	}
 
+	mounted, err := s.mountManager.IsMounted(stagingPath)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to check if staging path is mounted: %v", err)
+	}
+	if mounted {
+		klog.V(1).InfoS("Staging path already mounted, volume already staged",
+			"volume_id", req.GetVolumeId(), "staging_path", stagingPath)
+		return &csi.NodeStageVolumeResponse{}, nil
+	}
+
 	// Use FormatAndMount which will check if already formatted and only format if needed
 	klog.V(2).InfoS("Formatting and mounting device if needed",
 		"device_path", devicePath,

pkg/csi/node_test.go

@@ -68,6 +68,7 @@ func TestNodeStageVolume_Success(t *testing.T) {
 	miscsi.EXPECT().Login(ctx, target).Return(nil)
 	miscsi.EXPECT().GetDevice(ctx, target).Return("/dev/sdb", nil)
 	miscsi.EXPECT().WaitForDevice(ctx, "/dev/sdb", 30*time.Second).Return(nil)
+	mmount.EXPECT().IsMounted("/tmp/staging").Return(false, nil)
 	mmount.EXPECT().FormatAndMountDevice(ctx, "/dev/sdb", "/tmp/staging", "ext4", nil).Return(nil)
 
 	resp, err := svc.NodeStageVolume(ctx, req)
@@ -112,6 +113,7 @@ func TestNodeStageVolume_AlreadyLoggedIn(t *testing.T) {
 	miscsi.EXPECT().IsLoggedIn(ctx, target).Return(true, nil)
 	miscsi.EXPECT().GetDevice(ctx, target).Return("/dev/sdb", nil)
 	miscsi.EXPECT().WaitForDevice(ctx, "/dev/sdb", 30*time.Second).Return(nil)
+	mmount.EXPECT().IsMounted("/tmp/staging").Return(false, nil)
 	mmount.EXPECT().FormatAndMountDevice(ctx, "/dev/sdb", "/tmp/staging", "xfs", nil).Return(nil)
 
 	resp, err := svc.NodeStageVolume(ctx, req)
@@ -120,6 +122,51 @@ func TestNodeStageVolume_AlreadyLoggedIn(t *testing.T) {
 	g.Expect(resp).NotTo(BeNil())
 }
 
+func TestNodeStageVolume_AlreadyStaged(t *testing.T) {
+	g := NewGomegaWithT(t)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	svc, miscsi, mmount := newTestNode(ctrl)
+	ctx := context.Background()
+
+	req := &csi.NodeStageVolumeRequest{
+		VolumeId:          "vol-1",
+		StagingTargetPath: "/tmp/staging",
+		PublishContext: map[string]string{
+			"target-iqn": "iqn.2024-01.com.example:target",
+			"ip-address": "192.168.1.100",
+			"username":   "testuser",
+			"password":   "testpass",
+		},
+		VolumeCapability: &csi.VolumeCapability{
+			AccessType: &csi.VolumeCapability_Mount{
+				Mount: &csi.VolumeCapability_MountVolume{
+					FsType: "ext4",
+				},
+			},
+		},
+	}
+
+	target := &iscsi.TargetInfo{
+		Portal:   "192.168.1.100:3260",
+		IQN:      "iqn.2024-01.com.example:target",
+		Username: "testuser",
+		Password: "testpass",
+	}
+
+	miscsi.EXPECT().IsLoggedIn(ctx, target).Return(true, nil)
+	miscsi.EXPECT().GetDevice(ctx, target).Return("/dev/sdb", nil)
+	miscsi.EXPECT().WaitForDevice(ctx, "/dev/sdb", 30*time.Second).Return(nil)
+	mmount.EXPECT().IsMounted("/tmp/staging").Return(true, nil)
+	// FormatAndMountDevice must NOT be called for an already-staged volume
+
+	resp, err := svc.NodeStageVolume(ctx, req)
+
+	g.Expect(err).To(BeNil())
+	g.Expect(resp).NotTo(BeNil())
+}
+
 func TestNodeStageVolume_MissingTargetIQN(t *testing.T) {
 	g := NewGomegaWithT(t)
 	ctrl := gomock.NewController(t)
@@ -355,6 +402,31 @@ func TestNodePublishVolume_AlreadyMounted(t *testing.T) {
 	g.Expect(resp).NotTo(BeNil())
 }
 
+func TestNodePublishVolume_BindMountFails(t *testing.T) {
+	g := NewGomegaWithT(t)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	svc, _, mmount := newTestNode(ctrl)
+	ctx := context.Background()
+
+	req := &csi.NodePublishVolumeRequest{
+		VolumeId:          "vol-1",
+		TargetPath:        "/tmp/target",
+		StagingTargetPath: "/tmp/staging",
+	}
+
+	mmount.EXPECT().IsMounted("/tmp/target").Return(false, nil)
+	mmount.EXPECT().BindMount(ctx, "/tmp/staging", "/tmp/target", []string{}).
+		Return(errors.New("already mounted"))
+
+	resp, err := svc.NodePublishVolume(ctx, req)
+
+	g.Expect(err).NotTo(BeNil())
+	g.Expect(resp).To(BeNil())
+	g.Expect(status.Code(err)).To(Equal(codes.Internal))
+}
+
 func TestNodeUnpublishVolume_Success(t *testing.T) {
 	g := NewGomegaWithT(t)
 	ctrl := gomock.NewController(t)