admin/kubesphere
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

split snapshot controller and update capability.

Browse Source 
Signed-off-by: f10atin9 <f10atin9@kubesphere.io>
This commit is contained in:
f10atin9
2021-08-23 10:47:27 +08:00
parent 574eb221ab
commit 5e9679941b
5 changed files with 511 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
cmd/controller-manager/app/controllers.go
View File

@@ -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,

173
pkg/controller/storage/capability/capability_controller.go
View File

@@ -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)
}

133
pkg/controller/storage/capability/capability_controller_test.go
View File

@@ -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)

195
pkg/controller/storage/snapshot/snapshot_controller.go Normal file
View File

@@ -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{})
}

258
pkg/controller/storage/snapshot/snapshot_controller_test.go Normal file
View File

@@ -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))
}