diff --git a/csi/Dockerfile b/csi/Dockerfile index d774be84..37b4f9dd 100644 --- a/csi/Dockerfile +++ b/csi/Dockerfile @@ -22,6 +22,8 @@ RUN apt-get update && \ (echo "APT::Install-Recommends false;" > /etc/apt/apt.conf) && \ apt-get update && \ apt-get install -y e2fsprogs xfsprogs kmod iproute2 \ + # NFS mount dependencies + nfs-common netbase \ # dependencies of qemu-storage-daemon libnuma1 liburing2 libglib2.0-0 libfuse3-3 libaio1 libzstd1 libnettle8 \ libgmp10 libhogweed6 libp11-kit0 libidn2-0 libunistring2 libtasn1-6 libpcre2-8-0 libffi8 && \ diff --git a/csi/deploy/009-storage-class.yaml b/csi/deploy/009-storage-class.yaml index 81d16b4a..8e31880d 100644 --- a/csi/deploy/009-storage-class.yaml +++ b/csi/deploy/009-storage-class.yaml @@ -9,8 +9,16 @@ metadata: provisioner: csi.vitastor.io volumeBindingMode: Immediate parameters: - etcdVolumePrefix: "" - poolId: "1" + # CSI driver can create block-based volumes and VitastorFS-based volumes + # only VitastorFS-based volumes and raw block volumes (without FS) support ReadWriteMany mode + # set this parameter to VitastorFS metadata volume name to use VitastorFS + # if unset, block-based volumes will be created + vitastorfs: "" + # for block-based storage classes, pool ID may be either a string (name) or a number (ID) + # for vitastorFS-based storage classes it must be a string - name of the default pool for FS data + poolId: "testpool" + # volume name prefix for block-based storage classes or NFS subdirectory (including /) for FS-based volumes + volumePrefix: "" # you can choose other configuration file if you have it in the config map # different etcd URLs and prefixes should also be put in the config #configPath: "/etc/vitastor/vitastor.conf" diff --git a/csi/src/controllerserver.go b/csi/src/controllerserver.go index cbc34f08..e03a8838 100644 --- a/csi/src/controllerserver.go +++ b/csi/src/controllerserver.go @@ -8,7 +8,6 @@ import ( "encoding/json" "fmt" "strings" - "strconv" "time" "os" "io/ioutil" @@ -68,9 +67,10 @@ func GetConnectionParams(params map[string]string) (map[string]string, error) { configPath = "/etc/vitastor/vitastor.conf" } - else + ctxVars["configPath"] = configPath + if (params["vitastorfs"] != "") { - ctxVars["configPath"] = configPath + ctxVars["vitastorfs"] = params["vitastorfs"] } config := make(map[string]interface{}) configFD, err := os.Open(configPath) @@ -140,33 +140,57 @@ func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol return nil, status.Error(codes.InvalidArgument, "volume capabilities is a required field") } - err := cs.checkCaps(volumeCapabilities) - if (err != nil) - { - return nil, err - } - - etcdVolumePrefix := req.Parameters["etcdVolumePrefix"] - poolId, _ := strconv.ParseUint(req.Parameters["poolId"], 10, 64) - if (poolId == 0) - { - return nil, status.Error(codes.InvalidArgument, "poolId is missing in storage class configuration") - } - - volName := etcdVolumePrefix + req.GetName() - volSize := 1 * GB - if capRange := req.GetCapacityRange(); capRange != nil - { - volSize = ((capRange.GetRequiredBytes() + MB - 1) / MB) * MB - } - ctxVars, err := GetConnectionParams(req.Parameters) if (err != nil) { return nil, err } - args := []string{ "create", volName, "-s", fmt.Sprintf("%v", volSize), "--pool", fmt.Sprintf("%v", poolId) } + err = cs.checkCaps(volumeCapabilities, ctxVars["vitastorfs"] != "") + if (err != nil) + { + return nil, err + } + + pool := req.Parameters["poolId"] + if (pool == "") + { + return nil, status.Error(codes.InvalidArgument, "poolId is missing in storage class configuration") + } + volumePrefix := req.Parameters["volumePrefix"] + if (volumePrefix == "") + { + // Old name + volumePrefix = req.Parameters["etcdVolumePrefix"] + } + volName := volumePrefix + req.GetName() + volSize := 1 * GB + if capRange := req.GetCapacityRange(); capRange != nil + { + volSize = ((capRange.GetRequiredBytes() + MB - 1) / MB) * MB + } + + if (ctxVars["vitastorfs"] != "") + { + // Nothing to create, subdirectories are created during mounting + // FIXME: It would be cool to support quotas some day and set it here + if (req.VolumeContentSource.GetSnapshot() != nil) + { + return nil, status.Error(codes.InvalidArgument, "VitastorFS doesn't support snapshots") + } + ctxVars["name"] = volName + ctxVars["pool"] = pool + volumeIdJson, _ := json.Marshal(ctxVars) + return &csi.CreateVolumeResponse{ + Volume: &csi.Volume{ + // Ugly, but VolumeContext isn't passed to DeleteVolume :-( + VolumeId: string(volumeIdJson), + CapacityBytes: volSize, + }, + }, nil + } + + args := []string{ "create", volName, "-s", fmt.Sprintf("%v", volSize), "--pool", pool } // Support creation from snapshot var src *csi.VolumeContentSource @@ -249,6 +273,12 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol return nil, err } + if (ctxVars["vitastorfs"] != "") + { + // FIXME: Delete FS subdirectory + return &csi.DeleteVolumeResponse{}, nil + } + _, err = invokeCLI(ctxVars, []string{ "rm", volName }) if (err != nil) { @@ -283,13 +313,25 @@ func (cs *ControllerServer) ValidateVolumeCapabilities(ctx context.Context, req { return nil, status.Error(codes.InvalidArgument, "volumeId is nil") } + volVars := make(map[string]string) + err := json.Unmarshal([]byte(volumeID), &volVars) + if (err != nil) + { + return nil, status.Error(codes.Internal, "volume ID not in JSON format") + } + ctxVars, err := GetConnectionParams(volVars) + if (err != nil) + { + return nil, err + } + volumeCapabilities := req.GetVolumeCapabilities() if (volumeCapabilities == nil) { return nil, status.Error(codes.InvalidArgument, "volumeCapabilities is nil") } - err := cs.checkCaps(volumeCapabilities) + err = cs.checkCaps(volumeCapabilities, ctxVars["vitastorfs"] != "") if (err != nil) { return nil, err @@ -302,7 +344,7 @@ func (cs *ControllerServer) ValidateVolumeCapabilities(ctx context.Context, req }, nil } -func (cs *ControllerServer) checkCaps(volumeCapabilities []*csi.VolumeCapability) error +func (cs *ControllerServer) checkCaps(volumeCapabilities []*csi.VolumeCapability, fs bool) error { var volumeCapabilityAccessModes []*csi.VolumeCapability_AccessMode for _, mode := range []csi.VolumeCapability_AccessMode_Mode{ @@ -318,6 +360,10 @@ func (cs *ControllerServer) checkCaps(volumeCapabilities []*csi.VolumeCapability { if (capability.GetBlock() != nil) { + if (fs) + { + return status.Errorf(codes.InvalidArgument, "%v not supported with FS-based volumes", capability) + } for _, mode := range []csi.VolumeCapability_AccessMode_Mode{ csi.VolumeCapability_AccessMode_MULTI_NODE_SINGLE_WRITER, csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER, @@ -328,6 +374,12 @@ func (cs *ControllerServer) checkCaps(volumeCapabilities []*csi.VolumeCapability } } + if (fs) + { + // All access modes including RWX are supported with FS-based volumes + return nil + } + capabilitySupport := false for _, capability := range volumeCapabilities { @@ -342,7 +394,7 @@ func (cs *ControllerServer) checkCaps(volumeCapabilities []*csi.VolumeCapability if (!capabilitySupport) { - return status.Errorf(codes.NotFound, "%v not supported", volumeCapabilities) + return status.Errorf(codes.InvalidArgument, "%v not supported", volumeCapabilities) } return nil @@ -434,6 +486,12 @@ func (cs *ControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS { return nil, status.Error(codes.Internal, "volume ID not in JSON format") } + + if (ctxVars["vitastorfs"] != "") + { + return nil, status.Error(codes.InvalidArgument, "VitastorFS doesn't support snapshots") + } + volName := ctxVars["name"] // Create image using vitastor-cli @@ -492,6 +550,11 @@ func (cs *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteS return nil, err } + if (ctxVars["vitastorfs"] != "") + { + return nil, status.Error(codes.InvalidArgument, "VitastorFS doesn't support snapshots") + } + _, err = invokeCLI(ctxVars, []string{ "rm", volName+"@"+snapName }) if (err != nil) { @@ -523,6 +586,11 @@ func (cs *ControllerServer) ListSnapshots(ctx context.Context, req *csi.ListSnap return nil, err } + if (ctxVars["vitastorfs"] != "") + { + return nil, status.Error(codes.InvalidArgument, "VitastorFS doesn't support snapshots") + } + inodeCfg, err := invokeList(ctxVars, volName+"@*", false) if (err != nil) { @@ -586,6 +654,16 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi return nil, err } + if (ctxVars["vitastorfs"] != "") + { + // Nothing to change + // FIXME: Support quotas and change quota here + return &csi.ControllerExpandVolumeResponse{ + CapacityBytes: req.CapacityRange.RequiredBytes, + NodeExpansionRequired: false, + }, nil + } + inodeCfg, err := invokeList(ctxVars, volName, true) if (err != nil) { diff --git a/csi/src/nodeserver.go b/csi/src/nodeserver.go index df4db6b2..aefe8d85 100644 --- a/csi/src/nodeserver.go +++ b/csi/src/nodeserver.go @@ -5,11 +5,15 @@ package vitastor import ( "context" + "crypto/sha1" + "encoding/hex" "encoding/json" "fmt" "os" "os/exec" "path/filepath" + "regexp" + "strconv" "strings" "sync" "syscall" @@ -29,13 +33,14 @@ import ( type NodeServer struct { *Driver - useVduse bool - stateDir string - mounter mount.Interface + useVduse bool + stateDir string + nfsStageDir string + mounter mount.Interface restartInterval time.Duration - mu sync.Mutex - cond *sync.Cond - volumeLocks map[string]bool + mu sync.Mutex + cond *sync.Cond + volumeLocks map[string]bool } type DeviceState struct @@ -48,6 +53,15 @@ type DeviceState struct PidFile string `json:"pidFile"` } +type NfsState struct +{ + ConfigPath string `json:"configPath"` + FsName string `json:"fsName"` + Pool string `json:"pool"` + Path string `json:"path"` + Port int `json:"port"` +} + // NewNodeServer create new instance node func NewNodeServer(driver *Driver) *NodeServer { @@ -60,11 +74,17 @@ func NewNodeServer(driver *Driver) *NodeServer { stateDir += "/" } + nfsStageDir := os.Getenv("NFS_STAGE_DIR") + if (nfsStageDir == "") + { + nfsStageDir = "/var/lib/kubelet/plugins/csi.vitastor.io/nfs" + } ns := &NodeServer{ - Driver: driver, - useVduse: checkVduseSupport(), - stateDir: stateDir, - mounter: mount.New(""), + Driver: driver, + useVduse: checkVduseSupport(), + stateDir: stateDir, + nfsStageDir: nfsStageDir, + mounter: mount.New(""), volumeLocks: make(map[string]bool), } ns.cond = sync.NewCond(&ns.mu) @@ -123,12 +143,12 @@ func (ns *NodeServer) restarter() func (ns *NodeServer) restoreVduseDaemons() { pattern := ns.stateDir+"vitastor-vduse-*.json" - matches, err := filepath.Glob(pattern) + stateFiles, err := filepath.Glob(pattern) if (err != nil) { klog.Errorf("failed to list %s: %v", pattern, err) } - if (len(matches) == 0) + if (len(stateFiles) == 0) { return } @@ -146,59 +166,162 @@ func (ns *NodeServer) restoreVduseDaemons() klog.Errorf("/sbin/vdpa -j dev list returned bad JSON (error %v): %v", err, string(devListJSON)) return } - for _, stateFile := range matches + for _, stateFile := range stateFiles { - vdpaId := filepath.Base(stateFile) - vdpaId = vdpaId[0:len(vdpaId)-5] - // Check if VDPA device is still added to the bus - if (devs[vdpaId] == nil) - { - // Unused, clean it up - unmapVduseById(ns.stateDir, vdpaId) - continue - } + ns.checkVduseState(stateFile, devs) + } +} - stateJSON, err := os.ReadFile(stateFile) +func (ns *NodeServer) checkVduseState(stateFile string, devs map[string]interface{}) +{ + // Check if VDPA device is still added to the bus + vdpaId := filepath.Base(stateFile) + vdpaId = vdpaId[0:len(vdpaId)-5] + if (devs[vdpaId] == nil) + { + // Unused, clean it up + unmapVduseById(ns.stateDir, vdpaId) + return + } + + // Read state file + stateJSON, err := os.ReadFile(stateFile) + if (err != nil) + { + klog.Warningf("error reading state file %v: %v", stateFile, err) + return + } + var state DeviceState + err = json.Unmarshal(stateJSON, &state) + if (err != nil) + { + klog.Warningf("state file %v contains invalid JSON (error %v): %v", stateFile, err, string(stateJSON)) + return + } + + // Lock volume + ns.lockVolume(state.ConfigPath+":block:"+state.Image) + defer ns.unlockVolume(state.ConfigPath+":block:"+state.Image) + + // Recheck state file after locking + _, err = os.ReadFile(stateFile) + if (err != nil) + { + klog.Warningf("state file %v disappeared, skipping volume", stateFile) + return + } + + // Check if the storage daemon is still active + pidFile := ns.stateDir + vdpaId + ".pid" + exists := false + proc, err := findByPidFile(pidFile) + if (err == nil) + { + exists = proc.Signal(syscall.Signal(0)) == nil + } + if (!exists) + { + // Restart daemon + klog.Warningf("restarting storage daemon for volume %v (VDPA ID %v)", state.Image, vdpaId) + err = startStorageDaemon(vdpaId, state.Image, pidFile, state.ConfigPath, state.Readonly) if (err != nil) { - klog.Warningf("error reading state file %v: %v", stateFile, err) - continue + klog.Warningf("failed to restart storage daemon for volume %v: %v", state.Image, err) } - var state DeviceState - err = json.Unmarshal(stateJSON, &state) + } +} + +func (ns *NodeServer) restoreNfsDaemons() +{ + pattern := ns.stateDir+"vitastor-nfs-*.json" + stateFiles, err := filepath.Glob(pattern) + if (err != nil) + { + klog.Errorf("failed to list %s: %v", pattern, err) + } + if (len(stateFiles) == 0) + { + return + } + activeNFS, err := ns.listActiveNFS() + if (err != nil) + { + return + } + // Check all state files and try to restore active mounts + for _, stateFile := range stateFiles + { + ns.checkNfsState(stateFile, activeNFS) + } +} + +func (ns *NodeServer) readNfsState(stateFile string, allowNotExists bool) (*NfsState, error) +{ + stateJSON, err := os.ReadFile(stateFile) + if (err != nil) + { + if (allowNotExists && os.IsNotExist(err)) + { + return nil, nil + } + klog.Warningf("error reading state file %v: %v", stateFile, err) + return nil, err + } + var state NfsState + err = json.Unmarshal(stateJSON, &state) + if (err != nil) + { + klog.Warningf("state file %v contains invalid JSON (error %v): %v", stateFile, err, string(stateJSON)) + return nil, err + } + return &state, nil +} + +func (ns *NodeServer) checkNfsState(stateFile string, activeNfs map[int][]string) +{ + // Read state file + state, err := ns.readNfsState(stateFile, false) + if (err != nil) + { + return + } + // Lock FS + ns.lockVolume(state.ConfigPath+":fs:"+state.FsName) + defer ns.unlockVolume(state.ConfigPath+":fs:"+state.FsName) + // Check if NFS at this port is still mounted + pidFile := ns.stateDir + filepath.Base(stateFile) + pidFile = pidFile[0:len(pidFile)-5] + ".pid" + if (len(activeNfs[state.Port]) == 0) + { + // this is a stale state file, remove it + klog.Warningf("state file %v contains stale mount at port %d, removing it", stateFile, state.Port) + ns.stopNFS(stateFile, pidFile) + return + } + // Check PID file + exists := false + proc, err := findByPidFile(pidFile) + if (err == nil) + { + exists = proc.Signal(syscall.Signal(0)) == nil + } + if (!exists) + { + // Restart vitastor-nfs server + klog.Warningf("restarting NFS server for FS %v at port %v", state.FsName, state.Port) + _, _, err := system( + "/usr/bin/vitastor-nfs", "start", + "--pidfile", pidFile, + "--bind", "127.0.0.1", + "--port", fmt.Sprintf("%d", state.Port), + "--fs", state.FsName, + "--pool", state.Pool, + "--portmap", "0", + ) if (err != nil) { - klog.Warningf("state file %v contains invalid JSON (error %v): %v", stateFile, err, string(stateJSON)) - continue + klog.Warningf("failed to restart NFS server for FS %v: %v", state.FsName, err) } - - ns.lockVolume(state.ConfigPath+":"+state.Image) - - // Recheck state file after locking - _, err = os.ReadFile(stateFile) - if (err != nil) - { - klog.Warningf("state file %v disappeared, skipping volume", stateFile) - ns.unlockVolume(state.ConfigPath+":"+state.Image) - continue - } - - // Check if the storage daemon is still active - pidFile := ns.stateDir + vdpaId + ".pid" - exists := false - proc, err := findByPidFile(pidFile) - if (err == nil) - { - exists = proc.Signal(syscall.Signal(0)) == nil - } - if (!exists) - { - // Restart daemon - klog.Warningf("restarting storage daemon for volume %v (VDPA ID %v)", state.Image, vdpaId) - _ = startStorageDaemon(vdpaId, state.Image, pidFile, state.ConfigPath, state.Readonly) - } - - ns.unlockVolume(state.ConfigPath+":"+state.Image) } } @@ -220,8 +343,13 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol } volName := ctxVars["name"] - ns.lockVolume(ctxVars["configPath"]+":"+volName) - defer ns.unlockVolume(ctxVars["configPath"]+":"+volName) + if (ctxVars["vitastorfs"] != "") + { + return &csi.NodeStageVolumeResponse{}, nil + } + + ns.lockVolume(ctxVars["configPath"]+":block:"+volName) + defer ns.unlockVolume(ctxVars["configPath"]+":block:"+volName) targetPath := req.GetStagingTargetPath() isBlock := req.GetVolumeCapability().GetBlock() != nil @@ -408,8 +536,13 @@ func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag } volName := ctxVars["name"] - ns.lockVolume(ctxVars["configPath"]+":"+volName) - defer ns.unlockVolume(ctxVars["configPath"]+":"+volName) + if (ctxVars["vitastorfs"] != "") + { + return &csi.NodeUnstageVolumeResponse{}, nil + } + + ns.lockVolume(ctxVars["configPath"]+":block:"+volName) + defer ns.unlockVolume(ctxVars["configPath"]+":block:"+volName) targetPath := req.GetStagingTargetPath() devicePath, _, err := mount.GetDeviceNameFromMount(ns.mounter, targetPath) @@ -462,6 +595,153 @@ func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag return &csi.NodeUnstageVolumeResponse{}, nil } +// Mount or check if NFS is already mounted +func (ns *NodeServer) mountNFS(ctxVars map[string]string) (string, error) +{ + sum := sha1.Sum([]byte(ctxVars["configPath"]+":fs:"+ctxVars["vitastorfs"])) + nfsHash := hex.EncodeToString(sum[:]) + stateFile := ns.stateDir+"vitastor-nfs-"+nfsHash+".json" + pidFile := ns.stateDir+"vitastor-nfs-"+nfsHash+".pid" + mountPath := ns.nfsStageDir+"/"+nfsHash + state, err := ns.readNfsState(stateFile, true) + if (state != nil) + { + return state.Path, nil + } + if (err != nil) + { + return "", err + } + err = os.MkdirAll(mountPath, 0777) + if (err != nil) + { + return "", err + } + // Create a new mount + state = &NfsState{ + ConfigPath: ctxVars["configPath"], + FsName: ctxVars["vitastorfs"], + Pool: ctxVars["pool"], + Path: mountPath, + } + klog.Infof("starting new NFS server for FS %v", state.FsName) + stdout, _, err := system( + "/usr/bin/vitastor-nfs", "start", + "--pidfile", pidFile, + "--bind", "127.0.0.1", + "--port", "auto", + "--fs", state.FsName, + "--pool", state.Pool, + "--portmap", "0", + ) + if (err != nil) + { + return "", err + } + match := regexp.MustCompile("Port: (\\d+)").FindStringSubmatch(string(stdout)) + if (match == nil) + { + klog.Errorf("failed to find port in vitastor-nfs output: %v", string(stdout)) + ns.stopNFS(stateFile, pidFile) + return "", fmt.Errorf("failed to find port in vitastor-nfs output (bad vitastor-nfs version?)") + } + port, _ := strconv.ParseUint(match[1], 0, 16) + state.Port = int(port) + // Write state file + stateJSON, _ := json.Marshal(state) + err = os.WriteFile(stateFile, stateJSON, 0600) + if (err != nil) + { + klog.Errorf("failed to write state file %v", stateFile) + ns.stopNFS(stateFile, pidFile) + return "", err + } + // Mount NFS + _, _, err = system( + "mount", "-t", "nfs", "127.0.0.1:/", state.Path, + "-o", fmt.Sprintf("port=%d,mountport=%d,nfsvers=3,soft,nolock,tcp", port, port), + ) + if (err != nil) + { + ns.stopNFS(stateFile, pidFile) + return "", err + } + return state.Path, nil +} + +// Mount or check if NFS is already mounted +func (ns *NodeServer) checkStopNFS(ctxVars map[string]string) +{ + sum := sha1.Sum([]byte(ctxVars["configPath"]+":fs:"+ctxVars["vitastorfs"])) + nfsHash := hex.EncodeToString(sum[:]) + stateFile := ns.stateDir+"vitastor-nfs-"+nfsHash+".json" + pidFile := ns.stateDir+"vitastor-nfs-"+nfsHash+".pid" + mountPath := ns.nfsStageDir+"/"+nfsHash + state, err := ns.readNfsState(stateFile, true) + if (state == nil) + { + return + } + activeNFS, err := ns.listActiveNFS() + if (err != nil) + { + return + } + if (len(activeNFS[state.Port]) > 0) + { + return + } + // All volume mounts are detached, unmount the root mount and kill the server + err = mount.CleanupMountPoint(mountPath, ns.mounter, false) + if (err != nil) + { + klog.Errorf("failed to unmount %v: %v", mountPath, err) + return + } + ns.stopNFS(stateFile, pidFile) +} + +func (ns *NodeServer) stopNFS(stateFile, pidFile string) +{ + err := killByPidFile(pidFile) + if (err != nil) + { + klog.Errorf("failed to kill process with pid from %v: %v", pidFile, err) + } + os.Remove(pidFile) + os.Remove(stateFile) +} + +func (ns *NodeServer) listActiveNFS() (map[int][]string, error) +{ + mounts, err := mount.ParseMountInfo("/proc/self/mountinfo") + if (err != nil) + { + klog.Errorf("failed to list mounts: %v", err) + return nil, err + } + activeNFS := make(map[int][]string) + for _, mount := range mounts + { + // Volume mounts always refer to subpaths + if (mount.FsType == "nfs" && mount.Root != "/") + { + for _, opt := range mount.MountOptions + { + if (strings.HasPrefix(opt, "port=")) + { + port64, err := strconv.ParseUint(opt[5:], 10, 16) + if (err == nil) + { + activeNFS[int(port64)] = append(activeNFS[int(port64)], mount.MountPoint) + } + } + } + } + } + return activeNFS, nil +} + // NodePublishVolume mounts the volume mounted to the staging path to the target path func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) { @@ -480,28 +760,39 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis } volName := ctxVars["name"] - ns.lockVolume(ctxVars["configPath"]+":"+volName) - defer ns.unlockVolume(ctxVars["configPath"]+":"+volName) + if (ctxVars["vitastorfs"] != "") + { + ns.lockVolume(ctxVars["configPath"]+":fs:"+ctxVars["vitastorfs"]) + defer ns.unlockVolume(ctxVars["configPath"]+":fs:"+ctxVars["vitastorfs"]) + } + else + { + ns.lockVolume(ctxVars["configPath"]+":block:"+volName) + defer ns.unlockVolume(ctxVars["configPath"]+":block:"+volName) + } stagingTargetPath := req.GetStagingTargetPath() targetPath := req.GetTargetPath() isBlock := req.GetVolumeCapability().GetBlock() != nil - // Check that stagingTargetPath is mounted - notmnt, err := mount.IsNotMountPoint(ns.mounter, stagingTargetPath) - if (err != nil) + if (ctxVars["vitastorfs"] == "") { - klog.Errorf("staging path %v is not mounted: %w", stagingTargetPath, err) - return nil, fmt.Errorf("staging path %v is not mounted: %w", stagingTargetPath, err) - } - else if (notmnt) - { - klog.Errorf("staging path %v is not mounted", stagingTargetPath) - return nil, fmt.Errorf("staging path %v is not mounted", stagingTargetPath) + // Check that stagingTargetPath is mounted + notmnt, err := mount.IsNotMountPoint(ns.mounter, stagingTargetPath) + if (err != nil) + { + klog.Errorf("staging path %v is not mounted: %w", stagingTargetPath, err) + return nil, fmt.Errorf("staging path %v is not mounted: %w", stagingTargetPath, err) + } + else if (notmnt) + { + klog.Errorf("staging path %v is not mounted", stagingTargetPath) + return nil, fmt.Errorf("staging path %v is not mounted", stagingTargetPath) + } } // Check that targetPath is not already mounted - notmnt, err = mount.IsNotMountPoint(ns.mounter, targetPath) + notmnt, err := mount.IsNotMountPoint(ns.mounter, targetPath) if (err != nil) { if (os.IsNotExist(err)) @@ -542,6 +833,24 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis return nil, fmt.Errorf("target path %s is already mounted", targetPath) } + if (ctxVars["vitastorfs"] != "") + { + nfspath, err := ns.mountNFS(ctxVars) + if (err != nil) + { + ns.checkStopNFS(ctxVars) + return nil, err + } + // volName should include prefix + stagingTargetPath = nfspath+"/"+volName + err = os.MkdirAll(stagingTargetPath, 0777) + if (err != nil && !os.IsExist(err)) + { + ns.checkStopNFS(ctxVars) + return nil, err + } + } + execArgs := []string{"--bind", stagingTargetPath, targetPath} if (req.GetReadonly()) { @@ -553,6 +862,10 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis out, err := cmd.Output() if (err != nil) { + if (ctxVars["vitastorfs"] != "") + { + ns.checkStopNFS(ctxVars) + } return nil, fmt.Errorf("Error running mount %v: %s", strings.Join(execArgs, " "), out) } @@ -572,8 +885,16 @@ func (ns *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu } volName := ctxVars["name"] - ns.lockVolume(ctxVars["configPath"]+":"+volName) - defer ns.unlockVolume(ctxVars["configPath"]+":"+volName) + if (ctxVars["vitastorfs"] != "") + { + ns.lockVolume(ctxVars["configPath"]+":fs:"+ctxVars["vitastorfs"]) + defer ns.unlockVolume(ctxVars["configPath"]+":fs:"+ctxVars["vitastorfs"]) + } + else + { + ns.lockVolume(ctxVars["configPath"]+":block:"+volName) + defer ns.unlockVolume(ctxVars["configPath"]+":block:"+volName) + } targetPath := req.GetTargetPath() devicePath, _, err := mount.GetDeviceNameFromMount(ns.mounter, targetPath) @@ -600,6 +921,11 @@ func (ns *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu return nil, err } + if (ctxVars["vitastorfs"] != "") + { + ns.checkStopNFS(ctxVars) + } + return &csi.NodeUnpublishVolumeResponse{}, nil }