fix: check if target is already mounted in NodePublishVolume (juicedata#1402)

Author: zxh326

pkg/driver/fakes.go

@@ -18,6 +18,8 @@ package driver
 
 import (
 	"k8s.io/client-go/kubernetes/fake"
+	testingexec "k8s.io/utils/exec/testing"
+	"k8s.io/utils/mount"
 
 	"github.com/juicedata/juicefs-csi-driver/pkg/config"
 	"github.com/juicedata/juicefs-csi-driver/pkg/juicefs"
@@ -30,6 +32,14 @@ import (
 func NewFakeDriver(endpoint string, fakeProvider juicefs.Interface) *Driver {
 	registerer, _ := util.NewPrometheus(config.NodeName)
 	metrics := newNodeMetrics(registerer)
+	mp := make([]mount.MountPoint, 0)
+	mp = append(mp, mount.MountPoint{
+		Path: "/tmp/csi-mount/target",
+	})
+	fakeMounter := mount.SafeFormatAndMount{
+		Interface: mount.NewFakeMounter(mp),
+		Exec:      &testingexec.FakeExec{},
+	}
 	return &Driver{
 		endpoint: endpoint,
 		controllerService: controllerService{
@@ -38,11 +48,12 @@ func NewFakeDriver(endpoint string, fakeProvider juicefs.Interface) *Driver {
 			quotaPool: dispatch.NewPool(defaultQuotaPoolNum),
 		},
 		nodeService: nodeService{
-			quotaPool: dispatch.NewPool(defaultQuotaPoolNum),
-			juicefs:   fakeProvider,
-			nodeID:    "fake-node-id",
-			k8sClient: &k8sclient.K8sClient{Interface: fake.NewSimpleClientset()},
-			metrics:   metrics,
+			quotaPool:          dispatch.NewPool(defaultQuotaPoolNum),
+			juicefs:            fakeProvider,
+			nodeID:             "fake-node-id",
+			k8sClient:          &k8sclient.K8sClient{Interface: fake.NewSimpleClientset()},
+			metrics:            metrics,
+			SafeFormatAndMount: fakeMounter,
 		},
 	}
 }

pkg/driver/node.go

@@ -18,7 +18,9 @@ package driver
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"os"
 	"path"
 	"reflect"
 	"strconv"
@@ -140,6 +142,15 @@ func (d *nodeService) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 		return nil, status.Error(codes.InvalidArgument, "Volume capability not supported")
 	}
 
+	notMnt, notMntErr := d.IsLikelyNotMountPoint(target)
+	if notMntErr != nil && !errors.Is(notMntErr, os.ErrNotExist) {
+		return nil, status.Errorf(codes.Internal, "Could not check if %q is a mount point: %v", target, notMntErr)
+	}
+	if !notMnt {
+		log.Info("Volume already published at target path", "volumeId", volumeID, "target", target)
+		return &csi.NodePublishVolumeResponse{}, nil
+	}
+
 	log.Info("creating dir", "target", target)
 	if err := d.juicefs.CreateTarget(ctxWithLog, target); err != nil {
 		return nil, status.Errorf(codes.Internal, "Could not create dir %q: %v", target, err)

pkg/driver/node_test.go

@@ -53,11 +53,16 @@ var _ = Describe("nodeService", func() {
 	registerer, _ := util.NewPrometheus(config.NodeName)
 	metrics := newNodeMetrics(registerer)
 	var juicefsDriver *nodeService
+	mounter := &mount.SafeFormatAndMount{
+		Interface: mount.New(""),
+		Exec:      k8sexec.New(),
+	}
 	BeforeEach(func() {
 		juicefsDriver = &nodeService{
-			nodeID:    "fake_node_id",
-			k8sClient: &k8s.K8sClient{Interface: fake.NewSimpleClientset()},
-			metrics:   metrics,
+			nodeID:             "fake_node_id",
+			k8sClient:          &k8s.K8sClient{Interface: fake.NewSimpleClientset()},
+			metrics:            metrics,
+			SafeFormatAndMount: *mounter,
 		}
 	})