Implement CSI volume staging (one device per volume on each host)

csi/go.mod

@@ -9,6 +9,7 @@ require (
 	golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/grpc v1.33.1
+	google.golang.org/protobuf v1.24.0
 	k8s.io/klog v1.0.0
 	k8s.io/utils v0.0.0-20210305010621-2afb4311ab10
 )

csi/src/controllerserver.go

@@ -20,6 +20,7 @@ import (
 
     "google.golang.org/grpc/codes"
     "google.golang.org/grpc/status"
+    "google.golang.org/protobuf/types/known/timestamppb"
 
     "github.com/container-storage-interface/spec/lib/go/csi"
 )
@@ -45,6 +46,7 @@ type InodeConfig struct
     ParentPool uint64 `json:"parent_pool,omitempty"`
     ParentId uint64 `json:"parent_id,omitempty"`
     Readonly bool `json:"readonly,omitempty"`
+    CreateTs uint64 `json:"create_ts,omitempty"`
 }
 
 type ControllerServer struct
@@ -184,21 +186,11 @@ func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
     {
         if (strings.Index(err.Error(), "already exists") > 0)
         {
-            stat, err := invokeCLI(ctxVars, []string{ "ls", "--json", volName })
+            inodeCfg, err := invokeList(ctxVars, volName, true)
             if (err != nil)
             {
                 return nil, err
             }
-            var inodeCfg []InodeConfig
-            err = json.Unmarshal(stat, &inodeCfg)
-            if (err != nil)
-            {
-                return nil, status.Error(codes.Internal, "Invalid JSON in vitastor-cli ls: "+err.Error())
-            }
-            if (len(inodeCfg) == 0)
-            {
-                return nil, status.Error(codes.Internal, "vitastor-cli create said that image already exists, but ls can't find it")
-            }
             if (inodeCfg[0].Size < uint64(volSize))
             {
                 return nil, status.Error(codes.Internal, "image "+volName+" is already created, but size is less than expected")
@@ -230,15 +222,15 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
         return nil, status.Error(codes.InvalidArgument, "request cannot be empty")
     }
 
-    ctxVars := make(map[string]string)
-    err := json.Unmarshal([]byte(req.VolumeId), &ctxVars)
+    volVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.VolumeId), &volVars)
     if (err != nil)
     {
         return nil, status.Error(codes.Internal, "volume ID not in JSON format")
     }
-    volName := ctxVars["name"]
+    volName := volVars["name"]
 
-    ctxVars, _, _ = GetConnectionParams(ctxVars)
+    ctxVars, _, _ := GetConnectionParams(volVars)
 
     _, err = invokeCLI(ctxVars, []string{ "rm", volName })
     if (err != nil)
@@ -344,6 +336,8 @@ func (cs *ControllerServer) ControllerGetCapabilities(ctx context.Context, req *
         csi.ControllerServiceCapability_RPC_LIST_VOLUMES,
         csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
         csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
+        csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
+        // TODO: csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
     } {
         controllerServerCapabilities = append(controllerServerCapabilities, functionControllerServerCapabilities(capability))
     }
@@ -353,28 +347,219 @@ func (cs *ControllerServer) ControllerGetCapabilities(ctx context.Context, req *
     }, nil
 }
 
+func invokeList(ctxVars map[string]string, pattern string, expectExist bool) ([]InodeConfig, error)
+{
+    stat, err := invokeCLI(ctxVars, []string{ "ls", "--json", pattern })
+    if (err != nil)
+    {
+        return nil, err
+    }
+    var inodeCfg []InodeConfig
+    err = json.Unmarshal(stat, &inodeCfg)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "Invalid JSON in vitastor-cli ls: "+err.Error())
+    }
+    if (expectExist && len(inodeCfg) == 0)
+    {
+        return nil, status.Error(codes.Internal, "Can't find expected image "+pattern+" via vitastor-cli ls")
+    }
+    return inodeCfg, nil
+}
+
 // CreateSnapshot create snapshot of an existing PV
 func (cs *ControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error)
 {
-    return nil, status.Error(codes.Unimplemented, "")
+    klog.Infof("received controller create snapshot request %+v", protosanitizer.StripSecrets(req))
+    if (req == nil)
+    {
+        return nil, status.Errorf(codes.InvalidArgument, "request cannot be empty")
+    }
+    if (req.SourceVolumeId == "" || req.Name == "")
+    {
+        return nil, status.Error(codes.InvalidArgument, "source volume ID and snapshot name are required fields")
+    }
+
+    // snapshot name
+    snapName := req.Name
+
+    // req.VolumeId is an ugly json string in our case :)
+    ctxVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.SourceVolumeId), &ctxVars)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "volume ID not in JSON format")
+    }
+    volName := ctxVars["name"]
+
+    // Create image using vitastor-cli
+    _, err = invokeCLI(ctxVars, []string{ "create", "--snapshot", snapName, volName })
+    if (err != nil && strings.Index(err.Error(), "already exists") <= 0)
+    {
+        return nil, err
+    }
+
+    // Check created snapshot
+    inodeCfg, err := invokeList(ctxVars, volName+"@"+snapName, true)
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    // Use ugly JSON snapshot ID again, DeleteSnapshot doesn't have context :-(
+    ctxVars["snapshot"] = snapName
+    snapIdJson, _ := json.Marshal(ctxVars)
+    return &csi.CreateSnapshotResponse{
+        Snapshot: &csi.Snapshot{
+            SizeBytes: int64(inodeCfg[0].Size),
+            SnapshotId: string(snapIdJson),
+            SourceVolumeId: req.SourceVolumeId,
+            CreationTime: &timestamppb.Timestamp{ Seconds: int64(inodeCfg[0].CreateTs) },
+            ReadyToUse: true,
+        },
+    }, nil
 }
 
 // DeleteSnapshot delete provided snapshot of a PV
 func (cs *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error)
 {
-    return nil, status.Error(codes.Unimplemented, "")
+    klog.Infof("received controller delete snapshot request %+v", protosanitizer.StripSecrets(req))
+    if (req == nil)
+    {
+        return nil, status.Errorf(codes.InvalidArgument, "request cannot be empty")
+    }
+    if (req.SnapshotId == "")
+    {
+        return nil, status.Error(codes.InvalidArgument, "snapshot ID is a required field")
+    }
+
+    volVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.SnapshotId), &volVars)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "snapshot ID not in JSON format")
+    }
+    volName := volVars["name"]
+    snapName := volVars["snapshot"]
+
+    ctxVars, _, _ := GetConnectionParams(volVars)
+
+    _, err = invokeCLI(ctxVars, []string{ "rm", volName+"@"+snapName })
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    return &csi.DeleteSnapshotResponse{}, nil
 }
 
 // ListSnapshots list the snapshots of a PV
 func (cs *ControllerServer) ListSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error)
 {
-    return nil, status.Error(codes.Unimplemented, "")
+    klog.Infof("received controller list snapshots request %+v", protosanitizer.StripSecrets(req))
+    if (req == nil)
+    {
+        return nil, status.Error(codes.InvalidArgument, "request cannot be empty")
+    }
+
+    volVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.SourceVolumeId), &volVars)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "volume ID not in JSON format")
+    }
+    volName := volVars["name"]
+    ctxVars, _, _ := GetConnectionParams(volVars)
+
+    inodeCfg, err := invokeList(ctxVars, volName+"@*", false)
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    resp := &csi.ListSnapshotsResponse{}
+    for _, ino := range inodeCfg
+    {
+        snapName := ino.Name[len(volName)+1:]
+        if (len(req.StartingToken) > 0 && snapName < req.StartingToken)
+        {
+        }
+        else if (req.MaxEntries == 0 || len(resp.Entries) < int(req.MaxEntries))
+        {
+            volVars["snapshot"] = snapName
+            snapIdJson, _ := json.Marshal(volVars)
+            resp.Entries = append(resp.Entries, &csi.ListSnapshotsResponse_Entry{
+                Snapshot: &csi.Snapshot{
+                    SizeBytes: int64(ino.Size),
+                    SnapshotId: string(snapIdJson),
+                    SourceVolumeId: req.SourceVolumeId,
+                    CreationTime: &timestamppb.Timestamp{ Seconds: int64(ino.CreateTs) },
+                    ReadyToUse: true,
+                },
+            })
+        }
+        else
+        {
+            resp.NextToken = snapName
+            break
+        }
+    }
+
+    return resp, nil
 }
 
-// ControllerExpandVolume resizes a volume
+// ControllerExpandVolume increases the size of a volume
 func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error)
 {
-    return nil, status.Error(codes.Unimplemented, "")
+    klog.Infof("received controller expand volume request %+v", protosanitizer.StripSecrets(req))
+    if (req == nil)
+    {
+        return nil, status.Error(codes.InvalidArgument, "request cannot be empty")
+    }
+    if (req.VolumeId == "" || req.CapacityRange == nil || req.CapacityRange.RequiredBytes == 0)
+    {
+        return nil, status.Error(codes.InvalidArgument, "VolumeId, CapacityRange and RequiredBytes are required fields")
+    }
+
+    volVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.VolumeId), &volVars)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "volume ID not in JSON format")
+    }
+    volName := volVars["name"]
+    ctxVars, _, _ := GetConnectionParams(volVars)
+
+    inodeCfg, err := invokeList(ctxVars, volName, true)
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    if (req.CapacityRange.RequiredBytes > 0 && inodeCfg[0].Size < uint64(req.CapacityRange.RequiredBytes))
+    {
+        sz := ((req.CapacityRange.RequiredBytes+4095)/4096)*4096
+        _, err := invokeCLI(ctxVars, []string{ "modify", "--inc_size", "1", "--resize", fmt.Sprintf("%d", sz), volName })
+        if (err != nil)
+        {
+            return nil, err
+        }
+        inodeCfg, err := invokeList(ctxVars, volName, true)
+        if (err != nil)
+        {
+            return nil, err
+        }
+        return &csi.ControllerExpandVolumeResponse{
+            CapacityBytes: int64(inodeCfg[0].Size),
+            // Node expansion is required for FS
+            NodeExpansionRequired: req.VolumeCapability == nil || req.VolumeCapability.GetBlock() == nil,
+        }, nil
+    }
+
+    return &csi.ControllerExpandVolumeResponse{
+        CapacityBytes: int64(inodeCfg[0].Size),
+        NodeExpansionRequired: false,
+    }, nil
 }
 
 // ControllerGetVolume get volume info

csi/src/identityserver.go

@@ -49,6 +49,13 @@ func (is *IdentityServer) GetPluginCapabilities(ctx context.Context, req *csi.Ge
                     },
                 },
             },
+            {
+                Type: &csi.PluginCapability_VolumeExpansion_{
+                    VolumeExpansion: &csi.PluginCapability_VolumeExpansion{
+                        Type: csi.PluginCapability_VolumeExpansion_OFFLINE,
+                    },
+                },
+            },
         },
     }, nil
 }

csi/src/nodeserver.go

@@ -37,43 +37,13 @@ func NewNodeServer(driver *Driver) *NodeServer
     }
 }
 
-// NodeStageVolume mounts the volume to a staging path on the node.
-func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error)
+func (ns *NodeServer) checkMountPoint(targetPath string, isBlock bool)
 {
-    return &csi.NodeStageVolumeResponse{}, nil
-}
-
-// NodeUnstageVolume unstages the volume from the staging path
-func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error)
-{
-    return &csi.NodeUnstageVolumeResponse{}, nil
-}
-
-func Contains(list []string, s string) bool
-{
-    for i := 0; i < len(list); i++
-    {
-        if (list[i] == s)
-        {
-            return true
-        }
-    }
-    return false
-}
-
-// 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)
-{
-    klog.Infof("received node publish volume request %+v", protosanitizer.StripSecrets(req))
-
-    targetPath := req.GetTargetPath()
-    isBlock := req.GetVolumeCapability().GetBlock() != nil
-
     // Check that it's not already mounted
-    _, error := mount.IsNotMountPoint(ns.mounter, targetPath)
-    if (error != nil)
+    _, err := mount.IsNotMountPoint(ns.mounter, targetPath)
+    if (err != nil)
     {
-        if (os.IsNotExist(error))
+        if (os.IsNotExist(err))
         {
             if (isBlock)
             {
@@ -81,13 +51,13 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
                 if (err != nil)
                 {
                     klog.Errorf("failed to create block device mount target %s with error: %v", targetPath, err)
-                    return nil, status.Error(codes.Internal, err.Error())
+                    return status.Error(codes.Internal, err.Error("failed to create block device mount target"))
                 }
                 err = pathFile.Close()
                 if (err != nil)
                 {
                     klog.Errorf("failed to close %s with error: %v", targetPath, err)
-                    return nil, status.Error(codes.Internal, err.Error())
+                    return status.Error(codes.Internal, err.Error("failed to create block device mount target"))
                 }
             }
             else
@@ -96,15 +66,32 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
                 if (err != nil)
                 {
                     klog.Errorf("failed to create fs mount target %s with error: %v", targetPath, err)
-                    return nil, status.Error(codes.Internal, err.Error())
+                    return status.Error(codes.Internal, err.Error("failed to create fs mount target"))
                 }
             }
         }
         else
         {
-            return nil, status.Error(codes.Internal, error.Error())
+            return status.Error(codes.Internal, err)
         }
     }
+    return nil
+}
+
+// NodeStageVolume mounts the volume to a staging path on the node.
+func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error)
+{
+    klog.Infof("received node stage volume request %+v", protosanitizer.StripSecrets(req))
+
+    targetPath := req.GetStagingTargetPath()
+    isBlock := req.GetVolumeCapability().GetBlock() != nil
+
+    // Check that it's not already mounted
+    err := ns.checkMountPoint(targetPath, isBlock)
+    if (err != nil)
+    {
+        return nil, err
+    }
 
     ctxVars := make(map[string]string)
     err := json.Unmarshal([]byte(req.VolumeId), &ctxVars)
@@ -121,7 +108,6 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
     }
 
     // Map NBD device
-    // FIXME: Check if already mapped
     args := []string{
         "map", "--etcd_address", strings.Join(etcdUrl, ","),
         "--etcd_prefix", etcdPrefix,
@@ -226,6 +212,60 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
         }
         return nil, status.Error(codes.Internal, err.Error())
     }
+
+    return &csi.NodeStageVolumeResponse{}, nil
+}
+
+// NodeUnstageVolume unstages the volume from the staging path
+func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error)
+{
+    klog.Infof("received node unstage volume request %+v", protosanitizer.StripSecrets(req))
+    err := ns.unmount(req.GetStagingTargetPath())
+    if (err != nil)
+    {
+        return nil, err
+    }
+    return &csi.NodeUnstageVolumeResponse{}, nil
+}
+
+func Contains(list []string, s string) bool
+{
+    for i := 0; i < len(list); i++
+    {
+        if (list[i] == s)
+        {
+            return true
+        }
+    }
+    return false
+}
+
+// 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)
+{
+    klog.Infof("received node publish volume request %+v", protosanitizer.StripSecrets(req))
+
+    stagingTargetPath := req.GetStagingTargetPath()
+    targetPath := req.GetTargetPath()
+    isBlock := req.GetVolumeCapability().GetBlock() != nil
+
+    err := ns.checkMountPoint(targetPath, isBlock)
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    diskMounter := &mount.SafeFormatAndMount{Interface: ns.mounter, Exec: utilexec.New()}
+    err = diskMounter.Mount(stagingTargetPath, targetPath, "", []string{"bind"})
+    if (err != nil)
+    {
+        klog.Errorf(
+            "failed to bind staging path (%s) to path (%s) for volume (%s) error: %s",
+            stagingPath, targetPath, volName, err,
+        )
+        return nil, status.Error(codes.Internal, err.Error("failed to bind staging path"))
+    }
+
     return &csi.NodePublishVolumeResponse{}, nil
 }
 
@@ -233,25 +273,34 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 func (ns *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error)
 {
     klog.Infof("received node unpublish volume request %+v", protosanitizer.StripSecrets(req))
-    targetPath := req.GetTargetPath()
+    err := ns.unmount(req.GetTargetPath())
+    if (err != nil)
+    {
+        return nil, err
+    }
+    return &csi.NodeUnpublishVolumeResponse{}, nil
+}
+
+func (ns *NodeServer) unmount(targetPath string) error
+{
     devicePath, refCount, err := mount.GetDeviceNameFromMount(ns.mounter, targetPath)
     if (err != nil)
     {
         if (os.IsNotExist(err))
         {
-            return nil, status.Error(codes.NotFound, "Target path not found")
+            return status.Error(codes.NotFound, "Target path not found")
         }
-        return nil, status.Error(codes.Internal, err.Error())
+        return status.Error(codes.Internal, err.Error())
     }
     if (devicePath == "")
     {
-        return nil, status.Error(codes.NotFound, "Volume not mounted")
+        return status.Error(codes.NotFound, "Volume not mounted")
     }
     // unmount
     err = mount.CleanupMountPoint(targetPath, ns.mounter, false)
     if (err != nil)
     {
-        return nil, status.Error(codes.Internal, err.Error())
+        return status.Error(codes.Internal, err.Error())
     }
     // unmap NBD device
     if (refCount == 1)
@@ -262,7 +311,7 @@ func (ns *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu
             klog.Errorf("failed to unmap NBD device %s: %s, error: %v", devicePath, unmapOut, unmapErr)
         }
     }
-    return &csi.NodeUnpublishVolumeResponse{}, nil
+    return nil
 }
 
 // NodeGetVolumeStats returns volume capacity statistics available for the volume