Implement basic VitastorFS support in CSI

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 && \

csi/deploy/009-storage-class.yaml

@@ -9,8 +9,16 @@ metadata:
 provisioner: csi.vitastor.io
 volumeBindingMode: Immediate
 parameters:
-  etcdVolumePrefix: ""
+  # CSI driver can create block-based volumes and VitastorFS-based volumes
-  poolId: "1"
+  # 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"

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)
     {

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
+    useVduse        bool
-    stateDir string
+    stateDir        string
-    mounter mount.Interface
+    nfsStageDir     string
+    mounter         mount.Interface
     restartInterval time.Duration
-    mu sync.Mutex
+    mu              sync.Mutex
-    cond *sync.Cond
+    cond            *sync.Cond
-    volumeLocks map[string]bool
+    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,
+        Driver:      driver,
-        useVduse: checkVduseSupport(),
+        useVduse:    checkVduseSupport(),
-        stateDir: stateDir,
+        stateDir:    stateDir,
-        mounter: mount.New(""),
+        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)
+        ns.checkVduseState(stateFile, devs)
-        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
-        }
 
-        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)
+            klog.Warningf("failed to restart storage daemon for volume %v: %v", state.Image, err)
-            continue
         }
-        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))
+            klog.Warningf("failed to restart NFS server for FS %v: %v", state.FsName, err)
-            continue
         }
-
-        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)
+    if (ctxVars["vitastorfs"] != "")
-    defer ns.unlockVolume(ctxVars["configPath"]+":"+volName)
+    {
+        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)
+    if (ctxVars["vitastorfs"] != "")
-    defer ns.unlockVolume(ctxVars["configPath"]+":"+volName)
+    {
+        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)
+    if (ctxVars["vitastorfs"] != "")
-    defer ns.unlockVolume(ctxVars["configPath"]+":"+volName)
+    {
+        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
+    if (ctxVars["vitastorfs"] == "")
-    notmnt, err := mount.IsNotMountPoint(ns.mounter, stagingTargetPath)
-    if (err != nil)
     {
-        klog.Errorf("staging path %v is not mounted: %w", stagingTargetPath, err)
+        // Check that stagingTargetPath is mounted
-        return nil, fmt.Errorf("staging path %v is not mounted: %w", stagingTargetPath, err)
+        notmnt, err := mount.IsNotMountPoint(ns.mounter, stagingTargetPath)
-    }
+        if (err != nil)
-    else if (notmnt)
+        {
-    {
+            klog.Errorf("staging path %v is not mounted: %w", stagingTargetPath, err)
-        klog.Errorf("staging path %v is not mounted", stagingTargetPath)
+            return nil, fmt.Errorf("staging path %v is not mounted: %w", stagingTargetPath, err)
-        return nil, fmt.Errorf("staging path %v is not mounted", stagingTargetPath)
+        }
+        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)
+    if (ctxVars["vitastorfs"] != "")
-    defer ns.unlockVolume(ctxVars["configPath"]+":"+volName)
+    {
+        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
 }