Resize RBD CSI volumes on demand of CO resize request

Signed-off-by: Humble Chirammal <hchiramm@redhat.com>

pkg/rbd/controllerserver.go

@@ -642,3 +642,56 @@ func (cs *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteS
 
 	return &csi.DeleteSnapshotResponse{}, nil
 }
+
+// ControllerExpandVolume expand RBD Volumes on demand based on resizer request
+func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
+
+	if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_EXPAND_VOLUME); err != nil {
+		klog.Warningf("invalid expand volume req: %v", protosanitizer.StripSecrets(req))
+		return nil, err
+	}
+
+	volID := req.GetVolumeId()
+	if volID == "" {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID cannot be empty")
+	}
+
+	capRange := req.GetCapacityRange()
+	if capRange == nil {
+		return nil, status.Error(codes.InvalidArgument, "CapacityRange cannot be empty")
+	}
+
+	cr, err := util.NewUserCredentials(req.GetSecrets())
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+	defer cr.DeleteCredentials()
+
+	rbdVol := &rbdVolume{}
+	err = genVolFromVolID(ctx, rbdVol, volID, cr)
+	if err != nil {
+		return nil, status.Errorf(codes.NotFound, "Source Volume ID %s cannot found", volID)
+	}
+
+	// always round up the request size in bytes to the nearest MiB/GiB
+	volSize := util.RoundOffBytes(req.GetCapacityRange().GetRequiredBytes())
+
+	klog.V(4).Infof(util.Log(ctx, "rbd volume %s/%s size is %vMiB,resizing to %vMiB"), rbdVol.Pool, rbdVol.RbdImageName, rbdVol.VolSize, util.RoundOffVolSize(volSize))
+	// resize volume if required
+	if rbdVol.VolSize < volSize {
+		rbdVol.VolSize = util.RoundOffVolSize(volSize)
+		err = resizeRBDImage(ctx, rbdVol, cr)
+		if err != nil {
+			klog.Errorf("failed to resize rbd image: %s/%s with error: %v", rbdVol.Pool, rbdVol.RbdImageName, err)
+			return nil, status.Error(codes.Internal, err.Error())
+		}
+		klog.V(4).Infof(util.Log(ctx, "successfully resized %s to %v"), volID, volSize)
+	}
+
+	nodeExpansion := true
+	return &csi.ControllerExpandVolumeResponse{
+		CapacityBytes:         volSize,
+		NodeExpansionRequired: nodeExpansion,
+	}, nil
+
+}

pkg/rbd/driver.go

@@ -119,6 +119,7 @@ func (r *Driver) Run(conf *util.Config, cachePersister util.CachePersister) {
 			csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
 			csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
 			csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
+			csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
 		})
 		// We only support the multi-writer option when using block, but it's a supported capability for the plugin in general
 		// In addition, we want to add the remaining modes like MULTI_NODE_READER_ONLY,

pkg/rbd/identityserver.go

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

pkg/rbd/nodeserver.go

@@ -30,6 +30,7 @@ import (
 	"google.golang.org/grpc/status"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/util/mount"
+	"k8s.io/kubernetes/pkg/util/resizefs"
 )
 
 // NodeServer struct of ceph rbd driver with supported methods of CSI
@@ -536,6 +537,48 @@ func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag
 	return &csi.NodeUnstageVolumeResponse{}, nil
 }
 
+func (ns *NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
+
+	volumeID := req.GetVolumeId()
+	if volumeID == "" {
+		return nil, status.Error(codes.InvalidArgument, "NodeExpandVolume volume ID must be provided")
+	}
+	volumePath := req.GetVolumePath()
+	if volumePath == "" {
+		return nil, status.Error(codes.InvalidArgument, "NodeExpandVolume volume path must be provided")
+	}
+
+	diskMounter := &mount.SafeFormatAndMount{Interface: ns.mounter, Exec: mount.NewOsExec()}
+	volumePath = volumePath + "/" + req.GetVolumeId()
+	devicePath, err := getDevicePath(ctx, diskMounter, volumePath)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+	// TODO check size and return success or error
+	resizer := resizefs.NewResizeFs(diskMounter)
+	ok, err := resizer.Resize(devicePath, volumePath)
+	if !ok {
+		return nil, fmt.Errorf("rbd: resize failed on path %s, error: %v", req.GetVolumePath(), err)
+	}
+	return &csi.NodeExpandVolumeResponse{
+		CapacityBytes: req.GetCapacityRange().GetRequiredBytes(),
+	}, nil
+}
+
+func getDevicePath(ctx context.Context, mnt *mount.SafeFormatAndMount, stagingTargetPath string) (string, error) {
+	mointPoints, err := mnt.List()
+	if err != nil {
+		return "", err
+	}
+	for _, m := range mointPoints {
+		if strings.EqualFold(stagingTargetPath, m.Path) {
+			klog.V(3).Infof(util.Log(ctx, "found device %v for %v"), m.Device, stagingTargetPath)
+			return m.Device, nil
+		}
+	}
+	return "", fmt.Errorf("failed to get device for stagingtarget path %v", stagingTargetPath)
+}
+
 // NodeGetCapabilities returns the supported capabilities of the node server
 func (ns *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetCapabilitiesRequest) (*csi.NodeGetCapabilitiesResponse, error) {
 	return &csi.NodeGetCapabilitiesResponse{
@@ -554,6 +597,13 @@ func (ns *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetC
 					},
 				},
 			},
+			{
+				Type: &csi.NodeServiceCapability_Rpc{
+					Rpc: &csi.NodeServiceCapability_RPC{
+						Type: csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
+					},
+				},
+			},
 		},
 	}, nil
 }

pkg/rbd/rbd_util.go

@@ -813,3 +813,22 @@ func cleanupRBDImageMetadataStash(path string) error {
 
 	return nil
 }
+
+// resizeRBDImage resizes the given volume to new size
+func resizeRBDImage(ctx context.Context, rbdVol *rbdVolume, cr *util.Credentials) error {
+	var output []byte
+
+	mon := rbdVol.Monitors
+	image := rbdVol.RbdImageName
+	volSzMiB := fmt.Sprintf("%dM", rbdVol.VolSize)
+
+	klog.V(4).Infof(util.Log(ctx, "rbd: resize %s size %s using mon %s, pool %s "), image, volSzMiB, mon, rbdVol.Pool)
+	args := []string{"resize", image, "--size", volSzMiB, "--pool", rbdVol.Pool, "--id", cr.ID, "-m", mon, "--keyfile=" + cr.KeyFile}
+	output, err := execCommand("rbd", args)
+
+	if err != nil {
+		return errors.Wrapf(err, "failed to resize rbd image, command output: %s", string(output))
+	}
+
+	return nil
+}