diff --git a/pkg/rbd/controllerserver.go b/pkg/rbd/controllerserver.go index 004b6e79f..237e6b387 100644 --- a/pkg/rbd/controllerserver.go +++ b/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 + +} diff --git a/pkg/rbd/driver.go b/pkg/rbd/driver.go index 0f46a5975..1392acbf3 100644 --- a/pkg/rbd/driver.go +++ b/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, diff --git a/pkg/rbd/identityserver.go b/pkg/rbd/identityserver.go index da5b9e1da..70eaf5d0a 100644 --- a/pkg/rbd/identityserver.go +++ b/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 } diff --git a/pkg/rbd/nodeserver.go b/pkg/rbd/nodeserver.go index 3e6406f91..dc0e5966b 100644 --- a/pkg/rbd/nodeserver.go +++ b/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 } diff --git a/pkg/rbd/rbd_util.go b/pkg/rbd/rbd_util.go index af1d2c57e..de2ea9e09 100644 --- a/pkg/rbd/rbd_util.go +++ b/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 +}