From a16a1458d5bcc0ed5e136e73458cb601e116b617 Mon Sep 17 00:00:00 2001 From: Madhan-SWE Date: Mon, 7 Nov 2022 15:33:50 +0530 Subject: [PATCH] Volume clone feature added --- pkg/cloud/cloud_interface.go | 3 + pkg/cloud/mocks/mock_cloud.go | 44 +++++++++++++++ pkg/cloud/powervs.go | 70 +++++++++++++++++++++++ pkg/driver/controller.go | 50 +++++++++++++++-- pkg/driver/controller_test.go | 101 +++++++++++++++++++++++++++++++++- pkg/driver/sanity_test.go | 12 ++++ 6 files changed, 274 insertions(+), 6 deletions(-) diff --git a/pkg/cloud/cloud_interface.go b/pkg/cloud/cloud_interface.go index c3ec3bdcf..9a887e8fc 100644 --- a/pkg/cloud/cloud_interface.go +++ b/pkg/cloud/cloud_interface.go @@ -26,8 +26,11 @@ type Cloud interface { AttachDisk(volumeID string, nodeID string) (err error) DetachDisk(volumeID string, nodeID string) (err error) ResizeDisk(volumeID string, reqSize int64) (newSize int64, err error) + CloneDisk(sourceVolumeName string, cloneVolumeName string) (disk *Disk, err error) WaitForVolumeState(volumeID, state string) error + WaitForCloneStatus(taskId string) error GetDiskByName(name string) (disk *Disk, err error) + GetDiskByNamePrefix(namePrefix string) (disk *Disk, err error) GetDiskByID(volumeID string) (disk *Disk, err error) GetPVMInstanceByName(instanceName string) (instance *PVMInstance, err error) GetPVMInstanceByID(instanceID string) (instance *PVMInstance, err error) diff --git a/pkg/cloud/mocks/mock_cloud.go b/pkg/cloud/mocks/mock_cloud.go index f34367824..cf74fe976 100644 --- a/pkg/cloud/mocks/mock_cloud.go +++ b/pkg/cloud/mocks/mock_cloud.go @@ -49,6 +49,21 @@ func (mr *MockCloudMockRecorder) AttachDisk(volumeID, nodeID interface{}) *gomoc return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AttachDisk", reflect.TypeOf((*MockCloud)(nil).AttachDisk), volumeID, nodeID) } +// CloneDisk mocks base method. +func (m *MockCloud) CloneDisk(sourceVolumeName, cloneVolumeName string) (*cloud.Disk, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "CloneDisk", sourceVolumeName, cloneVolumeName) + ret0, _ := ret[0].(*cloud.Disk) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// CloneDisk indicates an expected call of CloneDisk. +func (mr *MockCloudMockRecorder) CloneDisk(sourceVolumeName, cloneVolumeName interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloneDisk", reflect.TypeOf((*MockCloud)(nil).CloneDisk), sourceVolumeName, cloneVolumeName) +} + // CreateDisk mocks base method. func (m *MockCloud) CreateDisk(volumeName string, diskOptions *cloud.DiskOptions) (*cloud.Disk, error) { m.ctrl.T.Helper() @@ -123,6 +138,21 @@ func (mr *MockCloudMockRecorder) GetDiskByName(name interface{}) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetDiskByName", reflect.TypeOf((*MockCloud)(nil).GetDiskByName), name) } +// GetDiskByNamePrefix mocks base method. +func (m *MockCloud) GetDiskByNamePrefix(namePrefix string) (*cloud.Disk, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetDiskByNamePrefix", namePrefix) + ret0, _ := ret[0].(*cloud.Disk) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetDiskByNamePrefix indicates an expected call of GetDiskByNamePrefix. +func (mr *MockCloudMockRecorder) GetDiskByNamePrefix(namePrefix interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetDiskByNamePrefix", reflect.TypeOf((*MockCloud)(nil).GetDiskByNamePrefix), namePrefix) +} + // GetImageByID mocks base method. func (m *MockCloud) GetImageByID(imageID string) (*cloud.PVMImage, error) { m.ctrl.T.Helper() @@ -227,6 +257,20 @@ func (mr *MockCloudMockRecorder) UpdateStoragePoolAffinity(instanceID interface{ return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateStoragePoolAffinity", reflect.TypeOf((*MockCloud)(nil).UpdateStoragePoolAffinity), instanceID) } +// WaitForCloneStatus mocks base method. +func (m *MockCloud) WaitForCloneStatus(taskId string) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "WaitForCloneStatus", taskId) + ret0, _ := ret[0].(error) + return ret0 +} + +// WaitForCloneStatus indicates an expected call of WaitForCloneStatus. +func (mr *MockCloudMockRecorder) WaitForCloneStatus(taskId interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WaitForCloneStatus", reflect.TypeOf((*MockCloud)(nil).WaitForCloneStatus), taskId) +} + // WaitForVolumeState mocks base method. func (m *MockCloud) WaitForVolumeState(volumeID, state string) error { m.ctrl.T.Helper() diff --git a/pkg/cloud/powervs.go b/pkg/cloud/powervs.go index 0e92f8990..8b0404a8c 100644 --- a/pkg/cloud/powervs.go +++ b/pkg/cloud/powervs.go @@ -71,6 +71,7 @@ type powerVSCloud struct { imageClient *instance.IBMPIImageClient pvmInstancesClient *instance.IBMPIInstanceClient volClient *instance.IBMPIVolumeClient + cloneVolumeClient *instance.IBMPICloneVolumeClient } type PVMInstance struct { @@ -123,6 +124,7 @@ func newPowerVSCloud(cloudInstanceID, zone string, debug bool) (Cloud, error) { volClient := instance.NewIBMPIVolumeClient(backgroundContext, piSession, cloudInstanceID) pvmInstancesClient := instance.NewIBMPIInstanceClient(backgroundContext, piSession, cloudInstanceID) imageClient := instance.NewIBMPIImageClient(backgroundContext, piSession, cloudInstanceID) + cloneVolumeClient := instance.NewIBMPICloneVolumeClient(backgroundContext, piSession, cloudInstanceID) return &powerVSCloud{ piSession: piSession, @@ -130,6 +132,7 @@ func newPowerVSCloud(cloudInstanceID, zone string, debug bool) (Cloud, error) { imageClient: imageClient, pvmInstancesClient: pvmInstancesClient, volClient: volClient, + cloneVolumeClient: cloneVolumeClient, }, nil } @@ -271,6 +274,36 @@ func (p *powerVSCloud) ResizeDisk(volumeID string, reqSize int64) (newSize int64 return int64(*v.Size), nil } +func (p *powerVSCloud) CloneDisk(sourceVolumeID string, cloneVolumeName string) (disk *Disk, err error) { + _, err = p.GetDiskByID(sourceVolumeID) + if err != nil { + return nil, err + } + cloneVolumeReq := &models.VolumesCloneAsyncRequest{ + Name: &cloneVolumeName, + VolumeIDs: []string{sourceVolumeID}, + } + cloneTaskRef, err := p.cloneVolumeClient.Create(cloneVolumeReq) + if err != nil { + return nil, err + } + cloneTaskId := cloneTaskRef.CloneTaskID + err = p.WaitForCloneStatus(*cloneTaskId) + if err != nil { + return nil, err + } + clonedVolumeDetails, err := p.cloneVolumeClient.Get(*cloneTaskId) + if err != nil { + return nil, err + } + clonedVolumeID := clonedVolumeDetails.ClonedVolumes[0].ClonedVolumeID + err = p.WaitForVolumeState(clonedVolumeID, VolumeAvailableState) + if err != nil { + return nil, err + } + return p.GetDiskByID(clonedVolumeID) +} + func (p *powerVSCloud) WaitForVolumeState(volumeID, state string) error { err := wait.PollImmediate(PollInterval, PollTimeout, func() (bool, error) { v, err := p.volClient.Get(volumeID) @@ -286,6 +319,21 @@ func (p *powerVSCloud) WaitForVolumeState(volumeID, state string) error { return nil } +func (p *powerVSCloud) WaitForCloneStatus(cloneTaskId string) error { + err := wait.PollImmediate(PollInterval, PollTimeout, func() (bool, error) { + c, err := p.cloneVolumeClient.Get(cloneTaskId) + if err != nil { + return false, err + } + spew.Dump(*c) + return *c.Status == "completed", nil + }) + if err != nil { + return err + } + return nil +} + func (p *powerVSCloud) GetDiskByName(name string) (disk *Disk, err error) { //TODO: remove capacityBytes params := p_cloud_volumes.NewPcloudCloudinstancesVolumesGetallParamsWithTimeout(TIMEOUT).WithCloudInstanceID(p.cloudInstanceID) @@ -309,6 +357,28 @@ func (p *powerVSCloud) GetDiskByName(name string) (disk *Disk, err error) { return nil, ErrNotFound } +func (p *powerVSCloud) GetDiskByNamePrefix(namePrefix string) (disk *Disk, err error) { + params := p_cloud_volumes.NewPcloudCloudinstancesVolumesGetallParamsWithTimeout(TIMEOUT).WithCloudInstanceID(p.cloudInstanceID) + resp, err := p.piSession.Power.PCloudVolumes.PcloudCloudinstancesVolumesGetall(params, p.piSession.AuthInfo(p.cloudInstanceID)) + if err != nil { + return nil, errors.ToError(err) + } + for _, v := range resp.Payload.Volumes { + if strings.HasPrefix(*v.Name, namePrefix) { + return &Disk{ + Name: *v.Name, + DiskType: *v.DiskType, + VolumeID: *v.VolumeID, + WWN: strings.ToLower(*v.Wwn), + Shareable: *v.Shareable, + CapacityGiB: int64(*v.Size), + }, nil + } + } + + return nil, ErrNotFound +} + func (p *powerVSCloud) GetDiskByID(volumeID string) (disk *Disk, err error) { v, err := p.volClient.Get(volumeID) if err != nil { diff --git a/pkg/driver/controller.go b/pkg/driver/controller.go index 5b9080ee0..0fa6273da 100644 --- a/pkg/driver/controller.go +++ b/pkg/driver/controller.go @@ -58,6 +58,7 @@ var ( csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME, csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME, csi.ControllerServiceCapability_RPC_EXPAND_VOLUME, + csi.ControllerServiceCapability_RPC_CLONE_VOLUME, } ) @@ -140,6 +141,47 @@ func (d *controllerService) CreateVolume(ctx context.Context, req *csi.CreateVol VolumeType: volumeType, } + if req.GetVolumeContentSource() != nil { + volumeSource := req.VolumeContentSource + switch volumeSource.Type.(type) { + case *csi.VolumeContentSource_Volume: + diskDetails, _ := d.cloud.GetDiskByNamePrefix("clone-" + volName) + if diskDetails != nil { + err := verifyVolumeDetails(opts, diskDetails) + if err != nil { + return nil, err + } + return newCreateVolumeResponse(diskDetails, req.VolumeContentSource), nil + } + if srcVolume := volumeSource.GetVolume(); srcVolume != nil { + srcVolumeID := srcVolume.GetVolumeId() + diskDetails, err := d.cloud.GetDiskByID(srcVolumeID) + if err != nil { + return nil, status.Errorf(codes.Internal, "Could not get the source volume %q: %v", srcVolumeID, err) + } + if util.GiBToBytes(diskDetails.CapacityGiB) != volSizeBytes { + return nil, status.Errorf(codes.Internal, "Cannot clone volume %v, source volume size is not equal to the clone volume", srcVolumeID) + } + err = verifyVolumeDetails(opts, diskDetails) + if err != nil { + return nil, err + } + diskFromSourceVolume, err := d.cloud.CloneDisk(srcVolumeID, volName) + if err != nil { + return nil, status.Errorf(codes.Internal, "Could not create volume %q: %v", volName, err) + } + + cloneDiskDetails, err := d.cloud.GetDiskByID(diskFromSourceVolume.VolumeID) + if err != nil { + return nil, status.Errorf(codes.Internal, "Could not create volume %q: %v", volName, err) + } + return newCreateVolumeResponse(cloneDiskDetails, req.VolumeContentSource), nil + } + default: + return nil, status.Errorf(codes.InvalidArgument, "%v not a proper volume source", volumeSource) + } + } + // check if disk exists // disk exists only if previous createVolume request fails due to any network/tcp error diskDetails, _ := d.cloud.GetDiskByName(volName) @@ -153,14 +195,14 @@ func (d *controllerService) CreateVolume(ctx context.Context, req *csi.CreateVol if err != nil { return nil, status.Errorf(codes.Internal, "Disk already exists and not in expected state") } - return newCreateVolumeResponse(diskDetails), nil + return newCreateVolumeResponse(diskDetails, req.VolumeContentSource), nil } disk, err := d.cloud.CreateDisk(volName, opts) if err != nil { return nil, status.Errorf(codes.Internal, "Could not create volume %q: %v", volName, err) } - return newCreateVolumeResponse(disk), nil + return newCreateVolumeResponse(disk, req.VolumeContentSource), nil } func (d *controllerService) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) { @@ -435,9 +477,7 @@ func (d *controllerService) ListSnapshots(ctx context.Context, req *csi.ListSnap return nil, status.Error(codes.Unimplemented, "") } -func newCreateVolumeResponse(disk *cloud.Disk) *csi.CreateVolumeResponse { - var src *csi.VolumeContentSource - +func newCreateVolumeResponse(disk *cloud.Disk, src *csi.VolumeContentSource) *csi.CreateVolumeResponse { return &csi.CreateVolumeResponse{ Volume: &csi.Volume{ VolumeId: disk.VolumeID, diff --git a/pkg/driver/controller_test.go b/pkg/driver/controller_test.go index a5bd479f5..4d26ebd2c 100644 --- a/pkg/driver/controller_test.go +++ b/pkg/driver/controller_test.go @@ -19,7 +19,7 @@ import ( "reflect" "testing" - csi "github.com/container-storage-interface/spec/lib/go/csi" + "github.com/container-storage-interface/spec/lib/go/csi" "github.com/golang/mock/gomock" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -95,7 +95,106 @@ func TestCreateVolume(t *testing.T) { } }, }, + { + name: "success normal with datasource PVC", + testFunc: func(t *testing.T) { + req := &csi.CreateVolumeRequest{ + Name: "clone-volume-name", + CapacityRange: stdCapRange, + VolumeCapabilities: stdVolCap, + Parameters: stdParams, + VolumeContentSource: &csi.VolumeContentSource{ + Type: &csi.VolumeContentSource_Volume{ + Volume: &csi.VolumeContentSource_VolumeSource{ + VolumeId: "test-volume-src-100", + }, + }, + }, + } + + ctx := context.Background() + + mockDisk := &cloud.Disk{ + VolumeID: req.Name, + CapacityGiB: util.BytesToGiB(stdVolSize), + DiskType: cloud.DefaultVolumeType, + } + mockSrcDisk := &cloud.Disk{ + VolumeID: "test-volume-src-100", + CapacityGiB: util.BytesToGiB(stdVolSize), + DiskType: cloud.DefaultVolumeType, + } + + mockCtl := gomock.NewController(t) + defer mockCtl.Finish() + + mockCloud := mocks.NewMockCloud(mockCtl) + mockCloud.EXPECT().GetDiskByNamePrefix(gomock.Eq("clone-"+req.Name)).Return(nil, nil) + mockCloud.EXPECT().GetDiskByID(gomock.Eq(mockSrcDisk.VolumeID)).Return(mockSrcDisk, nil) + mockCloud.EXPECT().CloneDisk(gomock.Eq(mockSrcDisk.VolumeID), gomock.Eq(req.Name)).Return(mockDisk, nil) + mockCloud.EXPECT().GetDiskByID(gomock.Eq(mockDisk.VolumeID)).Return(mockDisk, nil) + + powervsDriver := controllerService{ + cloud: mockCloud, + driverOptions: &Options{}, + volumeLocks: util.NewVolumeLocks(), + } + if _, err := powervsDriver.CreateVolume(ctx, req); err != nil { + srvErr, ok := status.FromError(err) + if !ok { + t.Fatalf("Could not get error status code from error: %v", srvErr) + } + t.Fatalf("Unexpected error: %v", srvErr.Code()) + } + }, + }, + { + name: "Create PVC with Data source - volume already exists", + testFunc: func(t *testing.T) { + req := &csi.CreateVolumeRequest{ + Name: "clone-volume-name", + CapacityRange: &csi.CapacityRange{RequiredBytes: stdVolSize}, + VolumeCapabilities: stdVolCap, + Parameters: stdParams, + VolumeContentSource: &csi.VolumeContentSource{ + Type: &csi.VolumeContentSource_Volume{ + Volume: &csi.VolumeContentSource_VolumeSource{ + VolumeId: "test-volume-src-100", + }, + }, + }, + } + + ctx := context.Background() + + mockDisk := &cloud.Disk{ + VolumeID: req.Name, + CapacityGiB: util.BytesToGiB(stdVolSize), + DiskType: cloud.DefaultVolumeType, + } + + mockCtl := gomock.NewController(t) + defer mockCtl.Finish() + + mockCloud := mocks.NewMockCloud(mockCtl) + mockCloud.EXPECT().GetDiskByNamePrefix(gomock.Eq("clone-"+req.Name)).Return(mockDisk, nil) + + powervsDriver := controllerService{ + cloud: mockCloud, + driverOptions: &Options{}, + volumeLocks: util.NewVolumeLocks(), + } + + if _, err := powervsDriver.CreateVolume(ctx, req); err != nil { + srvErr, ok := status.FromError(err) + if !ok { + t.Fatalf("Could not get error status code from error: %v", srvErr) + } + t.Fatalf("Unexpected error: %v", srvErr.Code()) + } + }, + }, { name: "csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER", testFunc: func(t *testing.T) { diff --git a/pkg/driver/sanity_test.go b/pkg/driver/sanity_test.go index dd84b07fc..34a98694a 100644 --- a/pkg/driver/sanity_test.go +++ b/pkg/driver/sanity_test.go @@ -218,6 +218,10 @@ func (c *fakeCloudProvider) DetachDisk(volumeID, nodeID string) error { return nil } +func (c *fakeCloudProvider) CloneDisk(sourceVolumeName string, cloneVolumeName string) (disk *cloud.Disk, err error) { + return nil, nil +} + func (c *fakeCloudProvider) IsAttached(volumeID string, nodeID string) (attached bool, err error) { return true, nil } @@ -226,6 +230,10 @@ func (c *fakeCloudProvider) WaitForVolumeState(volumeID, expectedState string) e return nil } +func (c *fakeCloudProvider) WaitForCloneStatus(cloneTaskId string) error { + return nil +} + func (c *fakeCloudProvider) GetDiskByName(name string) (*cloud.Disk, error) { var disks []*fakeDisk for _, d := range c.disks { @@ -240,6 +248,10 @@ func (c *fakeCloudProvider) GetDiskByName(name string) (*cloud.Disk, error) { return nil, nil } +func (c *fakeCloudProvider) GetDiskByNamePrefix(namePrefix string) (*cloud.Disk, error) { + return nil, nil +} + func (c *fakeCloudProvider) GetDiskByID(volumeID string) (*cloud.Disk, error) { for _, f := range c.disks { if f.Disk.VolumeID == volumeID {