diff --git a/cmd/controller-manager/app/controllers.go b/cmd/controller-manager/app/controllers.go index 0e250a023..54ffd9ffb 100644 --- a/cmd/controller-manager/app/controllers.go +++ b/cmd/controller-manager/app/controllers.go @@ -23,6 +23,8 @@ import ( "sigs.k8s.io/controller-runtime/pkg/manager" "sigs.k8s.io/kubefed/pkg/controller/util" + "kubesphere.io/kubesphere/pkg/controller/storage/snapshot" + iamv1alpha2 "kubesphere.io/api/iam/v1alpha2" authoptions "kubesphere.io/kubesphere/pkg/apiserver/authentication/options" @@ -99,7 +101,10 @@ func addControllers( client.Kubernetes().StorageV1().StorageClasses(), kubernetesInformer.Storage().V1().StorageClasses(), kubernetesInformer.Storage().V1beta1().CSIDrivers(), - capability.SnapshotSupported(client.Kubernetes().Discovery()), + ) + + volumeSnapshotController := snapshot.NewController( + kubernetesInformer.Storage().V1().StorageClasses(), client.Snapshot().SnapshotV1beta1().VolumeSnapshotClasses(), informerFactory.SnapshotSharedInformerFactory().Snapshot().V1beta1().VolumeSnapshotClasses(), ) @@ -215,6 +220,7 @@ func addControllers( "destinationrule-controller": drController, "job-controller": jobController, "storagecapability-controller": storageCapabilityController, + "volumesnapshot-controller": volumeSnapshotController, "user-controller": userController, "loginrecord-controller": loginRecordController, "cluster-controller": clusterController, diff --git a/pkg/controller/storage/capability/capability_controller.go b/pkg/controller/storage/capability/capability_controller.go index 6bbcfab7b..9acd35bdf 100644 --- a/pkg/controller/storage/capability/capability_controller.go +++ b/pkg/controller/storage/capability/capability_controller.go @@ -24,18 +24,11 @@ import ( "strconv" "time" - snapinformers "github.com/kubernetes-csi/external-snapshotter/client/v3/informers/externalversions/volumesnapshot/v1beta1" "k8s.io/apimachinery/pkg/labels" - "k8s.io/apimachinery/pkg/util/version" - "k8s.io/client-go/discovery" storageinformersv1beta1 "k8s.io/client-go/informers/storage/v1beta1" storagelistersv1beta1 "k8s.io/client-go/listers/storage/v1beta1" - snapshotv1beta1 "github.com/kubernetes-csi/external-snapshotter/client/v3/apis/volumesnapshot/v1beta1" - snapshotclient "github.com/kubernetes-csi/external-snapshotter/client/v3/clientset/versioned/typed/volumesnapshot/v1beta1" - snapshotlisters "github.com/kubernetes-csi/external-snapshotter/client/v3/listers/volumesnapshot/v1beta1" storagev1 "k8s.io/api/storage/v1" - "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" @@ -51,9 +44,8 @@ import ( ) const ( - minSnapshotSupportedVersion = "v1.17.0" - annotationSupportSnapshot = "storageclass.kubesphere.io/allow-snapshot" - annotationSupportClone = "storageclass.kubesphere.io/allow-clone" + annotationAllowSnapshot = "storageclass.kubesphere.io/allow-snapshot" + annotationAllowClone = "storageclass.kubesphere.io/allow-clone" ) type StorageCapabilityController struct { @@ -64,43 +56,26 @@ type StorageCapabilityController struct { csiDriverLister storagelistersv1beta1.CSIDriverLister csiDriverSynced cache.InformerSynced - snapshotSupported bool - snapshotClassClient snapshotclient.VolumeSnapshotClassInterface - snapshotClassLister snapshotlisters.VolumeSnapshotClassLister - snapshotClassSynced cache.InformerSynced - - workQueue workqueue.RateLimitingInterface - csiWorkQueue workqueue.RateLimitingInterface + storageClassWorkQueue workqueue.RateLimitingInterface } -// This controller is responsible to watch StorageClass/ProvisionerCapability. -// And then update StorageClassCapability CRD resource object to the newest status. +// This controller is responsible to watch StorageClass and CSIDriver. +// And then update StorageClass CRD resource object to the newest status. func NewController( storageClassClient storageclient.StorageClassInterface, storageClassInformer storageinformersv1.StorageClassInformer, csiDriverInformer storageinformersv1beta1.CSIDriverInformer, - snapshotSupported bool, - snapshotClassClient snapshotclient.VolumeSnapshotClassInterface, - snapshotClassInformer snapinformers.VolumeSnapshotClassInformer, ) *StorageCapabilityController { utilruntime.Must(crdscheme.AddToScheme(scheme.Scheme)) controller := &StorageCapabilityController{ - storageClassClient: storageClassClient, - storageClassLister: storageClassInformer.Lister(), - storageClassSynced: storageClassInformer.Informer().HasSynced, - csiDriverLister: csiDriverInformer.Lister(), - csiDriverSynced: csiDriverInformer.Informer().HasSynced, - snapshotSupported: snapshotSupported, - workQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "StorageClasses"), - csiWorkQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "csiDriver"), - } - - if snapshotSupported { - controller.snapshotClassClient = snapshotClassClient - controller.snapshotClassLister = snapshotClassInformer.Lister() - controller.snapshotClassSynced = snapshotClassInformer.Informer().HasSynced + storageClassClient: storageClassClient, + storageClassLister: storageClassInformer.Lister(), + storageClassSynced: storageClassInformer.Informer().HasSynced, + csiDriverLister: csiDriverInformer.Lister(), + csiDriverSynced: csiDriverInformer.Informer().HasSynced, + storageClassWorkQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "StorageClasses"), } storageClassInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ @@ -113,13 +88,10 @@ func NewController( } controller.enqueueStorageClass(newStorageClass) }, - DeleteFunc: controller.enqueueStorageClass, }) csiDriverInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ - AddFunc: controller.enqueueStorageClassByCSI, - UpdateFunc: nil, - DeleteFunc: controller.enqueueStorageClassByCSI, + AddFunc: controller.enqueueStorageClassByCSI, }) return controller @@ -131,7 +103,7 @@ func (c *StorageCapabilityController) Start(ctx context.Context) error { func (c *StorageCapabilityController) Run(threadCnt int, stopCh <-chan struct{}) error { defer utilruntime.HandleCrash() - defer c.workQueue.ShutDown() + defer c.storageClassWorkQueue.ShutDown() // Wait for the caches to be synced before starting workers klog.Info("Waiting for informer caches to sync") @@ -160,7 +132,7 @@ func (c *StorageCapabilityController) enqueueStorageClass(obj interface{}) { utilruntime.HandleError(err) return } - c.workQueue.Add(key) + c.storageClassWorkQueue.Add(key) } func (c *StorageCapabilityController) enqueueStorageClassByCSI(csi interface{}) { @@ -178,7 +150,7 @@ func (c *StorageCapabilityController) enqueueStorageClassByCSI(csi interface{}) } for _, obj := range objs { if obj.Provisioner == key { - c.workQueue.Add(obj.Name) + c.enqueueStorageClass(obj) } } return @@ -190,25 +162,25 @@ func (c *StorageCapabilityController) runWorker() { } func (c *StorageCapabilityController) processNextWorkItem() bool { - obj, shutdown := c.workQueue.Get() + obj, shutdown := c.storageClassWorkQueue.Get() if shutdown { return false } err := func(obj interface{}) error { - defer c.workQueue.Done(obj) + defer c.storageClassWorkQueue.Done(obj) var key string var ok bool if key, ok = obj.(string); !ok { - c.workQueue.Forget(obj) + c.storageClassWorkQueue.Forget(obj) utilruntime.HandleError(fmt.Errorf("expected string in workQueue but got %#v", obj)) return nil } if err := c.syncHandler(key); err != nil { - c.workQueue.AddRateLimited(key) + c.storageClassWorkQueue.AddRateLimited(key) return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error()) } - c.workQueue.Forget(obj) + c.storageClassWorkQueue.Forget(obj) klog.Infof("Successfully synced '%s'", key) return nil }(obj) @@ -232,55 +204,28 @@ func (c *StorageCapabilityController) syncHandler(key string) error { // Get StorageClass storageClass, err := c.storageClassLister.Get(name) - if err != nil { - // StorageClass has been deleted, delete StorageClassCapability and VolumeSnapshotClass - if errors.IsNotFound(err) && c.snapshotAllowed() { - err = c.deleteSnapshotClass(name) - if err != nil { - return err - } - } - return err - } //Cloning and volumeSnapshot support only available for CSI drivers. - withCapability := c.supportCapability(storageClass) - // Handle VolumeSnapshotClass with same name of StorageClass - // annotate "support-snapshot" of StorageClass - if c.snapshotAllowed() && withCapability { - _, err = c.snapshotClassLister.Get(name) - if err != nil { - // If VolumeSnapshotClass not exist, create it - if errors.IsNotFound(err) { - volumeSnapshotClassCreate := &snapshotv1beta1.VolumeSnapshotClass{ - ObjectMeta: metav1.ObjectMeta{Name: name}, - Driver: storageClass.Provisioner, - DeletionPolicy: snapshotv1beta1.VolumeSnapshotContentDelete, - } - _, err = c.snapshotClassClient.Create(context.Background(), volumeSnapshotClassCreate, metav1.CreateOptions{}) - if err != nil { - return err - } - } - } - } + isCSIStorage := c.hasCSIDriver(storageClass) - err = c.addStorageClassSnapshotAnnotation(storageClass, withCapability) + //Annotate storageClass + storageClassUpdated := storageClass.DeepCopy() + err = c.addStorageClassSnapshotAnnotation(storageClassUpdated, isCSIStorage) if err != nil { return err } - err = c.addCloneVolumeAnnotation(storageClass, withCapability) + err = c.addCloneVolumeAnnotation(storageClassUpdated, isCSIStorage) if err != nil { - return nil + return err } - _, err = c.storageClassClient.Update(context.Background(), storageClass, metav1.UpdateOptions{}) + _, err = c.storageClassClient.Update(context.Background(), storageClassUpdated, metav1.UpdateOptions{}) if err != nil { return err } return nil } -func (c *StorageCapabilityController) supportCapability(storageClass *storagev1.StorageClass) bool { +func (c *StorageCapabilityController) hasCSIDriver(storageClass *storagev1.StorageClass) bool { driver := storageClass.Provisioner if driver != "" { if _, err := c.csiDriverLister.Get(driver); err != nil { @@ -291,73 +236,29 @@ func (c *StorageCapabilityController) supportCapability(storageClass *storagev1. return false } -func (c *StorageCapabilityController) addStorageClassSnapshotAnnotation(storageClass *storagev1.StorageClass, snapshotSupported bool) error { - if snapshotSupported || !c.snapshotSupported { +func (c *StorageCapabilityController) addStorageClassSnapshotAnnotation(storageClass *storagev1.StorageClass, snapshotAllow bool) error { + if snapshotAllow { if storageClass.Annotations == nil { storageClass.Annotations = make(map[string]string) } - _, err := strconv.ParseBool(storageClass.Annotations[annotationSupportSnapshot]) - // err != nil means annotationSupportSnapshot is not illegal, include empty + _, err := strconv.ParseBool(storageClass.Annotations[annotationAllowSnapshot]) + // err != nil means annotationAllowSnapshot is not illegal, include empty if err != nil { - storageClass.Annotations[annotationSupportSnapshot] = strconv.FormatBool(c.snapshotSupported) - } - } else { - if storageClass.Annotations != nil && c.snapshotSupported { - if _, ok := storageClass.Annotations[annotationSupportSnapshot]; ok { - delete(storageClass.Annotations, annotationSupportSnapshot) - } + storageClass.Annotations[annotationAllowSnapshot] = strconv.FormatBool(snapshotAllow) } } return nil } -func (c *StorageCapabilityController) addCloneVolumeAnnotation(storageClass *storagev1.StorageClass, cloneSupported bool) error { - if cloneSupported { +func (c *StorageCapabilityController) addCloneVolumeAnnotation(storageClass *storagev1.StorageClass, cloneAllow bool) error { + if cloneAllow { if storageClass.Annotations == nil { storageClass.Annotations = make(map[string]string) } - _, err := strconv.ParseBool(storageClass.Annotations[annotationSupportClone]) + _, err := strconv.ParseBool(storageClass.Annotations[annotationAllowClone]) if err != nil { - storageClass.Annotations[annotationSupportClone] = strconv.FormatBool(cloneSupported) - } - } else { - if storageClass.Annotations != nil { - if _, ok := storageClass.Annotations[annotationSupportClone]; ok { - delete(storageClass.Annotations, annotationSupportClone) - } + storageClass.Annotations[annotationAllowClone] = strconv.FormatBool(cloneAllow) } } return nil } - -func (c *StorageCapabilityController) deleteSnapshotClass(name string) error { - if !c.snapshotAllowed() { - return nil - } - _, err := c.snapshotClassLister.Get(name) - if err != nil { - if errors.IsNotFound(err) { - return nil - } - return err - } - klog.Infof("Delete SnapshotClass %s", name) - return c.snapshotClassClient.Delete(context.Background(), name, metav1.DeleteOptions{}) -} - -func (c *StorageCapabilityController) snapshotAllowed() bool { - return c.snapshotSupported && c.snapshotClassClient != nil && c.snapshotClassLister != nil && c.snapshotClassSynced != nil -} - -func SnapshotSupported(discoveryInterface discovery.DiscoveryInterface) bool { - minVer := version.MustParseGeneric(minSnapshotSupportedVersion) - rawVer, err := discoveryInterface.ServerVersion() - if err != nil { - return false - } - ver, err := version.ParseSemantic(rawVer.String()) - if err != nil { - return false - } - return ver.AtLeast(minVer) -} diff --git a/pkg/controller/storage/capability/capability_controller_test.go b/pkg/controller/storage/capability/capability_controller_test.go index ef7292e59..d54ab3d66 100644 --- a/pkg/controller/storage/capability/capability_controller_test.go +++ b/pkg/controller/storage/capability/capability_controller_test.go @@ -26,9 +26,6 @@ import ( "testing" "time" - snapbeta1 "github.com/kubernetes-csi/external-snapshotter/client/v3/apis/volumesnapshot/v1beta1" - snapfake "github.com/kubernetes-csi/external-snapshotter/client/v3/clientset/versioned/fake" - snapinformers "github.com/kubernetes-csi/external-snapshotter/client/v3/informers/externalversions" storagev1 "k8s.io/api/storage/v1" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -40,7 +37,6 @@ import ( "k8s.io/client-go/tools/cache" ksfake "kubesphere.io/kubesphere/pkg/client/clientset/versioned/fake" - ksinformers "kubesphere.io/kubesphere/pkg/client/informers/externalversions" ) var ( @@ -51,17 +47,13 @@ type fixture struct { t *testing.T snapshotSupported bool // Clients - k8sClient *k8sfake.Clientset - snapshotClassClient *snapfake.Clientset - ksClient *ksfake.Clientset + k8sClient *k8sfake.Clientset + ksClient *ksfake.Clientset // Objects from here preload into NewSimpleFake. - storageObjects []runtime.Object // include StorageClass - snapshotClassObjects []runtime.Object - capabilityObjects []runtime.Object // include StorageClassCapability and ProvisionerCapability + storageObjects []runtime.Object // include StorageClass // Objects to put in the store. - storageClassLister []*storagev1.StorageClass - snapshotClassLister []*snapbeta1.VolumeSnapshotClass - csiDriverLister []*v1beta1.CSIDriver + storageClassLister []*storagev1.StorageClass + csiDriverLister []*v1beta1.CSIDriver // Actions expected to happen on the client. actions []core.Action } @@ -74,49 +66,34 @@ func newFixture(t *testing.T, snapshotSupported bool) *fixture { } func (f *fixture) newController() (*StorageCapabilityController, - k8sinformers.SharedInformerFactory, - ksinformers.SharedInformerFactory, - snapinformers.SharedInformerFactory) { + k8sinformers.SharedInformerFactory) { f.k8sClient = k8sfake.NewSimpleClientset(f.storageObjects...) - f.ksClient = ksfake.NewSimpleClientset(f.capabilityObjects...) - f.snapshotClassClient = snapfake.NewSimpleClientset(f.snapshotClassObjects...) k8sInformers := k8sinformers.NewSharedInformerFactory(f.k8sClient, noReSyncPeriodFunc()) - ksInformers := ksinformers.NewSharedInformerFactory(f.ksClient, noReSyncPeriodFunc()) - snapshotInformers := snapinformers.NewSharedInformerFactory(f.snapshotClassClient, noReSyncPeriodFunc()) - c := NewController( f.k8sClient.StorageV1().StorageClasses(), k8sInformers.Storage().V1().StorageClasses(), k8sInformers.Storage().V1beta1().CSIDrivers(), - f.snapshotSupported, - f.snapshotClassClient.SnapshotV1beta1().VolumeSnapshotClasses(), - snapshotInformers.Snapshot().V1beta1().VolumeSnapshotClasses(), ) for _, storageClass := range f.storageClassLister { _ = k8sInformers.Storage().V1().StorageClasses().Informer().GetIndexer().Add(storageClass) } - for _, snapshotClass := range f.snapshotClassLister { - _ = snapshotInformers.Snapshot().V1beta1().VolumeSnapshotClasses().Informer().GetIndexer().Add(snapshotClass) - } for _, csiDriver := range f.csiDriverLister { _ = k8sInformers.Storage().V1beta1().CSIDrivers().Informer().GetIndexer().Add(csiDriver) } - return c, k8sInformers, ksInformers, snapshotInformers + return c, k8sInformers } func (f *fixture) runController(scName string, startInformers bool, expectError bool) { - c, k8sI, crdI, snapI := f.newController() + c, k8sI := f.newController() if startInformers { stopCh := make(chan struct{}) defer close(stopCh) k8sI.Start(stopCh) - crdI.Start(stopCh) - snapI.Start(stopCh) } err := c.syncHandler(scName) @@ -127,9 +104,8 @@ func (f *fixture) runController(scName string, startInformers bool, expectError } var actions []core.Action - actions = append(actions, f.snapshotClassClient.Actions()...) actions = append(actions, f.k8sClient.Actions()...) - actions = append(actions, f.ksClient.Actions()...) + //actions = append(actions, f.ksClient.Actions()...) filerActions := filterInformerActions(actions) if len(filerActions) != len(f.actions) { f.t.Errorf("count of actions: differ (-got, +want): %s", cmp.Diff(filerActions, f.actions)) @@ -150,16 +126,6 @@ func (f *fixture) expectUpdateStorageClassAction(storageClass *storagev1.Storage schema.GroupVersionResource{Resource: "storageclasses"}, storageClass.Namespace, storageClass)) } -func (f *fixture) expectCreateSnapshotClassAction(snapshotClass *snapbeta1.VolumeSnapshotClass) { - f.actions = append(f.actions, core.NewCreateAction( - schema.GroupVersionResource{Resource: "volumesnapshotclasses"}, snapshotClass.Namespace, snapshotClass)) -} - -func (f *fixture) expectDeleteSnapshotClassAction(snapshotClass *snapbeta1.VolumeSnapshotClass) { - f.actions = append(f.actions, core.NewDeleteAction( - schema.GroupVersionResource{Resource: "volumesnapshotclasses"}, snapshotClass.Namespace, snapshotClass.Name)) -} - // filterInformerActions filters list and watch actions for testing resources. // Since list and watch don't change resource state we can filter it to lower // nose level in our tests. @@ -240,16 +206,6 @@ func newCSIDriver(name string) *v1beta1.CSIDriver { } } -func newSnapshotClass(storageClass *storagev1.StorageClass) *snapbeta1.VolumeSnapshotClass { - return &snapbeta1.VolumeSnapshotClass{ - ObjectMeta: v1.ObjectMeta{ - Name: storageClass.Name, - }, - Driver: storageClass.Provisioner, - DeletionPolicy: snapbeta1.VolumeSnapshotContentDelete, - } -} - func getKey(sc *storagev1.StorageClass, t *testing.T) string { key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(sc) if err != nil { @@ -263,46 +219,7 @@ func TestCreateStorageClass(t *testing.T) { fixture := newFixture(t, true) storageClass := newStorageClass("csi-example", "csi.example.com") storageClassUpdate := storageClass.DeepCopy() - storageClassUpdate.Annotations = map[string]string{annotationSupportSnapshot: "true", annotationSupportClone: "true"} - snapshotClass := newSnapshotClass(storageClass) - csiDriver := newCSIDriver("csi.example.com") - - // Objects exist - fixture.storageObjects = append(fixture.storageObjects, storageClass) - fixture.storageClassLister = append(fixture.storageClassLister, storageClass) - fixture.csiDriverLister = append(fixture.csiDriverLister, csiDriver) - - // Action expected - fixture.expectCreateSnapshotClassAction(snapshotClass) - fixture.expectUpdateStorageClassAction(storageClassUpdate) - - // Run test - fixture.run(getKey(storageClass, t)) -} - -func TestDeleteStorageClass(t *testing.T) { - storageClass := newStorageClass("csi-example", "csi.example.com") - snapshotClass := newSnapshotClass(storageClass) - - fixture := newFixture(t, true) - // Object exist - fixture.storageObjects = append(fixture.storageObjects, storageClass) - fixture.snapshotClassObjects = append(fixture.snapshotClassObjects, snapshotClass) - fixture.snapshotClassLister = append(fixture.snapshotClassLister, snapshotClass) - - // Action expected - fixture.expectDeleteSnapshotClassAction(snapshotClass) - - // Run test - fixture.run(getKey(storageClass, t)) -} - -func TestCreateStorageClassNotSupportSnapshot(t *testing.T) { - // K8S version < 1.17.0 - fixture := newFixture(t, false) - storageClass := newStorageClass("csi-example", "csi.example.com") - storageClassUpdate := storageClass.DeepCopy() - storageClassUpdate.Annotations = map[string]string{annotationSupportSnapshot: "false", annotationSupportClone: "true"} + storageClassUpdate.Annotations = map[string]string{annotationAllowSnapshot: "true", annotationAllowClone: "true"} csiDriver := newCSIDriver("csi.example.com") // Objects exist @@ -320,10 +237,9 @@ func TestCreateStorageClassNotSupportSnapshot(t *testing.T) { func TestStorageClassHadAnnotation(t *testing.T) { fixture := newFixture(t, true) storageClass := newStorageClass("csi-example", "csi.example.com") - storageClass.Annotations = map[string]string{annotationSupportSnapshot: "false", annotationSupportClone: "false"} + storageClass.Annotations = map[string]string{annotationAllowSnapshot: "false", annotationAllowClone: "false"} storageClassUpdate := storageClass.DeepCopy() csiDriver := newCSIDriver("csi.example.com") - snapshotClass := newSnapshotClass(storageClass) //Object exist fixture.storageObjects = append(fixture.storageObjects, storageClass) @@ -331,7 +247,6 @@ func TestStorageClassHadAnnotation(t *testing.T) { fixture.csiDriverLister = append(fixture.csiDriverLister, csiDriver) //Action expected - fixture.expectCreateSnapshotClassAction(snapshotClass) fixture.expectUpdateStorageClassAction(storageClassUpdate) //Run test @@ -341,36 +256,16 @@ func TestStorageClassHadAnnotation(t *testing.T) { func TestStorageClassHadOneAnnotation(t *testing.T) { fixture := newFixture(t, true) storageClass := newStorageClass("csi-example", "csi.example.com") - storageClass.Annotations = map[string]string{annotationSupportSnapshot: "false"} + storageClass.Annotations = map[string]string{annotationAllowSnapshot: "false"} storageClassUpdate := storageClass.DeepCopy() - storageClassUpdate.Annotations[annotationSupportClone] = "true" + storageClassUpdate.Annotations[annotationAllowClone] = "true" csiDriver := newCSIDriver("csi.example.com") - snapshotClass := newSnapshotClass(storageClass) - //object exist + //Object exist fixture.storageObjects = append(fixture.storageObjects, storageClass) fixture.storageClassLister = append(fixture.storageClassLister, storageClass) fixture.csiDriverLister = append(fixture.csiDriverLister, csiDriver) - //Action expected - fixture.expectCreateSnapshotClassAction(snapshotClass) - fixture.expectUpdateStorageClassAction(storageClassUpdate) - - //Run test - fixture.run(getKey(storageClass, t)) -} - -func TestDeleteCSIDriver(t *testing.T) { - fixture := newFixture(t, true) - storageClass := newStorageClass("csi-example", "csi.example.com") - storageClass.Annotations = map[string]string{annotationSupportSnapshot: "false", annotationSupportClone: "false"} - storageClassUpdate := storageClass.DeepCopy() - storageClassUpdate.Annotations = map[string]string{} - - //object exist - fixture.storageObjects = append(fixture.storageObjects, storageClass) - fixture.storageClassLister = append(fixture.storageClassLister, storageClass) - //Action expected fixture.expectUpdateStorageClassAction(storageClassUpdate) diff --git a/pkg/controller/storage/snapshot/snapshot_controller.go b/pkg/controller/storage/snapshot/snapshot_controller.go new file mode 100644 index 000000000..90808e28c --- /dev/null +++ b/pkg/controller/storage/snapshot/snapshot_controller.go @@ -0,0 +1,195 @@ +/* + + Copyright 2021 The KubeSphere Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + +*/ + +package snapshot + +import ( + "context" + "fmt" + "time" + + snapshotv1beta1 "github.com/kubernetes-csi/external-snapshotter/client/v3/apis/volumesnapshot/v1beta1" + snapshotclient "github.com/kubernetes-csi/external-snapshotter/client/v3/clientset/versioned/typed/volumesnapshot/v1beta1" + snapinformers "github.com/kubernetes-csi/external-snapshotter/client/v3/informers/externalversions/volumesnapshot/v1beta1" + snapshotlisters "github.com/kubernetes-csi/external-snapshotter/client/v3/listers/volumesnapshot/v1beta1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/wait" + storageinformersv1 "k8s.io/client-go/informers/storage/v1" + "k8s.io/client-go/kubernetes/scheme" + storagelistersv1 "k8s.io/client-go/listers/storage/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" + "k8s.io/klog" + + crdscheme "kubesphere.io/kubesphere/pkg/client/clientset/versioned/scheme" +) + +type VolumeSnapshotClassController struct { + storageClassLister storagelistersv1.StorageClassLister + storageClassSynced cache.InformerSynced + snapshotClassClient snapshotclient.VolumeSnapshotClassInterface + snapshotClassLister snapshotlisters.VolumeSnapshotClassLister + snapshotClassSynced cache.InformerSynced + + snapshotClassWorkQueue workqueue.RateLimitingInterface +} + +//This controller is responseible to watch StorageClass +//When storageClass has created ,create snapshot class +func NewController( + storageClassInformer storageinformersv1.StorageClassInformer, + snapshotClassClient snapshotclient.VolumeSnapshotClassInterface, + snapshotClassInformer snapinformers.VolumeSnapshotClassInformer, +) *VolumeSnapshotClassController { + + utilruntime.Must(crdscheme.AddToScheme(scheme.Scheme)) + + controller := &VolumeSnapshotClassController{ + storageClassLister: storageClassInformer.Lister(), + storageClassSynced: storageClassInformer.Informer().HasSynced, + snapshotClassClient: snapshotClassClient, + snapshotClassLister: snapshotClassInformer.Lister(), + snapshotClassSynced: snapshotClassInformer.Informer().HasSynced, + snapshotClassWorkQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "SnapshotClass"), + } + + storageClassInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: controller.enqueueStorageClass, + DeleteFunc: controller.enqueueStorageClass, + }) + + return controller +} + +func (c *VolumeSnapshotClassController) Start(ctx context.Context) error { + return c.Run(5, ctx.Done()) +} + +func (c *VolumeSnapshotClassController) Run(threadCnt int, stopCh <-chan struct{}) error { + defer utilruntime.HandleCrash() + defer c.snapshotClassWorkQueue.ShutDown() + + klog.Info("Waiting for informer cache to sync.") + cacheSyncs := []cache.InformerSynced{ + c.storageClassSynced, + c.snapshotClassSynced, + } + + if ok := cache.WaitForCacheSync(stopCh, cacheSyncs...); !ok { + return fmt.Errorf("failed to wait for caches to syne") + } + + for i := 0; i < threadCnt; i++ { + go wait.Until(c.runWorker, time.Second, stopCh) + } + klog.Info("Started workers") + <-stopCh + klog.Info("Shutting down workers") + return nil +} + +func (c *VolumeSnapshotClassController) enqueueStorageClass(obj interface{}) { + var key string + var err error + if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil { + utilruntime.HandleError(err) + return + } + c.snapshotClassWorkQueue.Add(key) +} + +func (c *VolumeSnapshotClassController) runWorker() { + for c.processNextWorkItem() { + } +} + +func (c *VolumeSnapshotClassController) processNextWorkItem() bool { + obj, shutdown := c.snapshotClassWorkQueue.Get() + if shutdown { + return false + } + + err := func(obj interface{}) error { + defer c.snapshotClassWorkQueue.Done(obj) + var key string + var ok bool + if key, ok = obj.(string); !ok { + c.snapshotClassWorkQueue.Forget(obj) + utilruntime.HandleError(fmt.Errorf("expected string in workQueue but got %#v", obj)) + return nil + } + if err := c.syncHandler(key); err != nil { + c.snapshotClassWorkQueue.AddRateLimited(key) + return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error()) + } + c.snapshotClassWorkQueue.Forget(obj) + klog.Infof("Successfully synced '%s'", key) + return nil + }(obj) + + if err != nil { + utilruntime.HandleError(err) + return true + } + return true +} + +func (c *VolumeSnapshotClassController) syncHandler(key string) error { + _, name, err := cache.SplitMetaNamespaceKey(key) + if err != nil { + utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key)) + return nil + } + + storageClass, err := c.storageClassLister.Get(name) + if err != nil { + // StorageClass has been deleted, delete VolumeSnapshotClass + if errors.IsNotFound(err) { + err = c.deleteSnapshotClass(name) + } + return err + } + + // If VolumeSnapshotClass not exist, create it + _, err = c.snapshotClassLister.Get(name) + if err != nil { + if errors.IsNotFound(err) { + volumeSnapshotClassCreate := &snapshotv1beta1.VolumeSnapshotClass{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Driver: storageClass.Provisioner, + DeletionPolicy: snapshotv1beta1.VolumeSnapshotContentDelete, + } + _, err = c.snapshotClassClient.Create(context.Background(), volumeSnapshotClassCreate, metav1.CreateOptions{}) + } + } + return err +} + +func (c *VolumeSnapshotClassController) deleteSnapshotClass(name string) error { + _, err := c.snapshotClassLister.Get(name) + if err != nil { + if errors.IsNotFound(err) { + return nil + } + return err + } + klog.Infof("Delete SnapshotClass %s", name) + return c.snapshotClassClient.Delete(context.Background(), name, metav1.DeleteOptions{}) +} diff --git a/pkg/controller/storage/snapshot/snapshot_controller_test.go b/pkg/controller/storage/snapshot/snapshot_controller_test.go new file mode 100644 index 000000000..82b4bd660 --- /dev/null +++ b/pkg/controller/storage/snapshot/snapshot_controller_test.go @@ -0,0 +1,258 @@ +/* + + Copyright 2021 The KubeSphere Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + +*/ + +package snapshot + +import ( + "reflect" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + snapbeta1 "github.com/kubernetes-csi/external-snapshotter/client/v3/apis/volumesnapshot/v1beta1" + snapfake "github.com/kubernetes-csi/external-snapshotter/client/v3/clientset/versioned/fake" + snapinformers "github.com/kubernetes-csi/external-snapshotter/client/v3/informers/externalversions" + storagev1 "k8s.io/api/storage/v1" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/diff" + k8sinformers "k8s.io/client-go/informers" + k8sfake "k8s.io/client-go/kubernetes/fake" + core "k8s.io/client-go/testing" + "k8s.io/client-go/tools/cache" + + ksfake "kubesphere.io/kubesphere/pkg/client/clientset/versioned/fake" +) + +var ( + noReSyncPeriodFunc = func() time.Duration { return 0 } +) + +type fixture struct { + t *testing.T + snapshotSupported bool + // Clients + k8sClient *k8sfake.Clientset + snapshotClassClient *snapfake.Clientset + ksClient *ksfake.Clientset + // Objects from here preload into NewSimpleFake. + storageObjects []runtime.Object // include StorageClass + snapshotClassObjects []runtime.Object + // Objects to put in the store. + storageClassLister []*storagev1.StorageClass + snapshotClassLister []*snapbeta1.VolumeSnapshotClass + // Actions expected to happen on the client. + actions []core.Action +} + +func newFixture(t *testing.T) *fixture { + return &fixture{t: t} +} + +func (f *fixture) newController() (*VolumeSnapshotClassController, k8sinformers.SharedInformerFactory, snapinformers.SharedInformerFactory) { + f.k8sClient = k8sfake.NewSimpleClientset(f.storageObjects...) + f.snapshotClassClient = snapfake.NewSimpleClientset(f.snapshotClassObjects...) + + k8sInformers := k8sinformers.NewSharedInformerFactory(f.k8sClient, noReSyncPeriodFunc()) + snapshotInformers := snapinformers.NewSharedInformerFactory(f.snapshotClassClient, noReSyncPeriodFunc()) + + c := NewController( + k8sInformers.Storage().V1().StorageClasses(), + f.snapshotClassClient.SnapshotV1beta1().VolumeSnapshotClasses(), + snapshotInformers.Snapshot().V1beta1().VolumeSnapshotClasses(), + ) + + for _, storageClass := range f.storageClassLister { + _ = k8sInformers.Storage().V1().StorageClasses().Informer().GetIndexer().Add(storageClass) + } + for _, snapshotClass := range f.snapshotClassLister { + _ = snapshotInformers.Snapshot().V1beta1().VolumeSnapshotClasses().Informer().GetIndexer().Add(snapshotClass) + } + + return c, k8sInformers, snapshotInformers +} + +func (f *fixture) runController(scName string, startInformers bool, expectError bool) { + c, k8sI, snapI := f.newController() + + if startInformers { + stopCh := make(chan struct{}) + defer close(stopCh) + k8sI.Start(stopCh) + snapI.Start(stopCh) + } + + err := c.syncHandler(scName) + if !expectError && err != nil { + f.t.Errorf("error syncing: %v", err) + } else if expectError && err == nil { + f.t.Error("expected error syncing, got nil") + } + + var actions []core.Action + actions = append(actions, f.snapshotClassClient.Actions()...) + actions = append(actions, f.k8sClient.Actions()...) + //actions = append(actions, f.ksClient.Actions()...) + filerActions := filterInformerActions(actions) + if len(filerActions) != len(f.actions) { + f.t.Errorf("count of actions: differ (-got, +want): %s", cmp.Diff(filerActions, f.actions)) + return + } + for i, action := range filerActions { + expectedAction := f.actions[i] + checkAction(expectedAction, action, f.t) + } +} + +func (f *fixture) run(scName string) { + f.runController(scName, true, false) +} + +func (f *fixture) expectCreateSnapshotClassAction(snapshotClass *snapbeta1.VolumeSnapshotClass) { + f.actions = append(f.actions, core.NewCreateAction( + schema.GroupVersionResource{Resource: "volumesnapshotclasses"}, snapshotClass.Namespace, snapshotClass)) +} + +func (f *fixture) expectDeleteSnapshotClassAction(snapshotClass *snapbeta1.VolumeSnapshotClass) { + f.actions = append(f.actions, core.NewDeleteAction( + schema.GroupVersionResource{Resource: "volumesnapshotclasses"}, snapshotClass.Namespace, snapshotClass.Name)) +} + +// filterInformerActions filters list and watch actions for testing resources. +// Since list and watch don't change resource state we can filter it to lower +// nose level in our tests. +func filterInformerActions(actions []core.Action) []core.Action { + var ret []core.Action + for _, action := range actions { + if action.GetVerb() == "list" || action.GetVerb() == "watch" { + continue + } + ret = append(ret, action) + } + return ret +} + +// checkAction verifies that expected and actual actions are equal and both have +// same attached resources +func checkAction(expected, actual core.Action, t *testing.T) { + if !(expected.Matches(actual.GetVerb(), actual.GetResource().Resource) && actual.GetSubresource() == expected.GetSubresource()) { + t.Errorf("\nExpected\n\t%#v\ngot\n\t%#v", expected, actual) + return + } + + if reflect.TypeOf(actual) != reflect.TypeOf(expected) { + t.Errorf("Action has wrong type. Expected: %t. Got: %t", expected, actual) + return + } + + switch a := actual.(type) { + case core.CreateActionImpl: + e, _ := expected.(core.CreateActionImpl) + expObject := e.GetObject() + object := a.GetObject() + if difference := cmp.Diff(object, expObject); len(difference) > 0 { + t.Errorf("[CreateAction] %T differ (-got, +want): %s", expObject, difference) + } + case core.UpdateActionImpl: + e, _ := expected.(core.UpdateActionImpl) + expObject := e.GetObject() + object := a.GetObject() + if difference := cmp.Diff(object, expObject); len(difference) > 0 { + t.Errorf("[UpdateAction] %T differ (-got, +want): %s", expObject, difference) + } + case core.PatchActionImpl: + e, _ := expected.(core.PatchActionImpl) + expPatch := e.GetPatch() + patch := a.GetPatch() + if !reflect.DeepEqual(expPatch, patch) { + t.Errorf("Action %s %s has wrong patch\nDiff:\n %s", + a.GetVerb(), a.GetResource().Resource, diff.ObjectGoPrintSideBySide(expPatch, patch)) + } + case core.DeleteActionImpl: + e, _ := expected.(core.DeleteActionImpl) + if difference := cmp.Diff(e.Name, a.Name); len(difference) > 0 { + t.Errorf("[UpdateAction] %T differ (-got, +want): %s", e.Name, difference) + } + default: + t.Errorf("Uncaptured Action %s %s, you should explicitly add a case to capture it", + actual.GetVerb(), actual.GetResource().Resource) + } +} + +func newStorageClass(name string) *storagev1.StorageClass { + isExpansion := true + return &storagev1.StorageClass{ + ObjectMeta: v1.ObjectMeta{ + Name: name, + }, + AllowVolumeExpansion: &isExpansion, + } +} + +func newSnapshotClass(storageClass *storagev1.StorageClass) *snapbeta1.VolumeSnapshotClass { + return &snapbeta1.VolumeSnapshotClass{ + ObjectMeta: v1.ObjectMeta{ + Name: storageClass.Name, + }, + Driver: storageClass.Provisioner, + DeletionPolicy: snapbeta1.VolumeSnapshotContentDelete, + } +} + +func getKey(sc *storagev1.StorageClass, t *testing.T) string { + key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(sc) + if err != nil { + t.Errorf("Unexpected error getting key for %v: %v", sc.Name, err) + return "" + } + return key +} + +func TestCreateStorageClass(t *testing.T) { + fixture := newFixture(t) + storageClass := newStorageClass("csi-example") + snapshotClass := newSnapshotClass(storageClass) + + // Objects exist + fixture.storageObjects = append(fixture.storageObjects, storageClass) + fixture.storageClassLister = append(fixture.storageClassLister, storageClass) + + // Action expected + fixture.expectCreateSnapshotClassAction(snapshotClass) + + // Run test + fixture.run(getKey(storageClass, t)) +} + +func TestDeleteStorageClass(t *testing.T) { + storageClass := newStorageClass("csi-example") + snapshotClass := newSnapshotClass(storageClass) + + fixture := newFixture(t) + // Object exist + fixture.storageObjects = append(fixture.storageObjects, storageClass) + fixture.snapshotClassObjects = append(fixture.snapshotClassObjects, snapshotClass) + fixture.snapshotClassLister = append(fixture.snapshotClassLister, snapshotClass) + + // Action expected + fixture.expectDeleteSnapshotClassAction(snapshotClass) + + // Run test + fixture.run(getKey(storageClass, t)) +}