Add support of RBD list volumes

currently all the created volumes are
stored in the metadata store, so we
can use this information to support
list volumes.

Signed-off-by: Madhu Rajanna <mrajanna@redhat.com>

pkg/rbd/controllerserver.go

@@ -19,6 +19,7 @@ package rbd
 import (
 	"fmt"
 	"os/exec"
+	"strconv"
 	"syscall"
 
 	"github.com/ceph/ceph-csi/pkg/util"
@@ -275,6 +276,45 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 	return &csi.DeleteVolumeResponse{}, nil
 }
 
+// ListVolumes returns a list of volumes stored in memory
+func (cs *ControllerServer) ListVolumes(ctx context.Context, req *csi.ListVolumesRequest) (*csi.ListVolumesResponse, error) {
+
+	if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_LIST_VOLUMES); err != nil {
+		klog.Warningf("invalid list volume req: %v", req)
+		return nil, err
+	}
+
+	//validate starting token if present
+	if len(req.GetStartingToken()) > 0 {
+		i, parseErr := strconv.ParseUint(req.StartingToken, 10, 32)
+		if parseErr != nil {
+			return nil, status.Errorf(codes.Aborted, "invalid starting token %s", parseErr.Error())
+		}
+		//check starting Token is greater than list of rbd volumes
+		if len(rbdVolumes) < int(i) {
+			return nil, status.Errorf(codes.Aborted, "invalid starting token %s", parseErr.Error())
+		}
+	}
+
+	var entries []*csi.ListVolumesResponse_Entry
+
+	for _, vol := range rbdVolumes {
+		entries = append(entries, &csi.ListVolumesResponse_Entry{
+			Volume: &csi.Volume{
+				VolumeId:      vol.VolID,
+				CapacityBytes: vol.VolSize,
+				VolumeContext: extractStoredVolOpt(vol),
+			},
+		})
+	}
+
+	resp := &csi.ListVolumesResponse{
+		Entries: entries,
+	}
+
+	return resp, nil
+}
+
 // ValidateVolumeCapabilities checks whether the volume capabilities requested
 // are supported.
 func (cs *ControllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {

pkg/rbd/rbd.go

@@ -98,6 +98,7 @@ func (r *Driver) Run(driverName, nodeID, endpoint string, containerized bool, ca
 	r.cd.AddControllerServiceCapabilities([]csi.ControllerServiceCapability_RPC_Type{
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
 		csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME,
+		csi.ControllerServiceCapability_RPC_LIST_VOLUMES,
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
 		csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
 		csi.ControllerServiceCapability_RPC_CLONE_VOLUME,

pkg/rbd/rbd_util.go

@@ -383,6 +383,37 @@ func protectSnapshot(pOpts *rbdSnapshot, adminID string, credentials map[string]
 	return nil
 }
 
+func extractStoredVolOpt(r *rbdVolume) map[string]string {
+	volOptions := make(map[string]string)
+	volOptions["pool"] = r.Pool
+
+	if len(r.Monitors) > 0 {
+		volOptions["monitors"] = r.Monitors
+	}
+
+	if len(r.MonValueFromSecret) > 0 {
+		volOptions["monValueFromSecret"] = r.MonValueFromSecret
+	}
+
+	volOptions["imageFormat"] = r.ImageFormat
+
+	if len(r.ImageFeatures) > 0 {
+		volOptions["imageFeatures"] = r.ImageFeatures
+	}
+
+	if len(r.AdminID) > 0 {
+		volOptions["adminid"] = r.AdminID
+	}
+
+	if len(r.UserID) > 0 {
+		volOptions["userid"] = r.UserID
+	}
+	if len(r.Mounter) > 0 {
+		volOptions["mounter"] = r.Mounter
+	}
+	return volOptions
+}
+
 func createSnapshot(pOpts *rbdSnapshot, adminID string, credentials map[string]string) error {
 	var output []byte