Skip to content

Commit

Permalink
refactor: start informers in one place
Browse files Browse the repository at this point in the history
  • Loading branch information
hccheng72 committed Dec 7, 2022
1 parent e8be0fe commit c9806d5
Show file tree
Hide file tree
Showing 10 changed files with 370 additions and 197 deletions.
68 changes: 37 additions & 31 deletions pkg/azuredisk/azuredisk_v2.go
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ import (
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
apiRuntime "k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/informers"
clientset "k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/cache"
Expand All @@ -46,6 +47,7 @@ import (
"k8s.io/klog/v2/klogr"

apiext "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
crdInformers "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions"
clientgoscheme "k8s.io/client-go/kubernetes/scheme"
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
clientmetrics "k8s.io/client-go/tools/metrics"
Expand All @@ -63,6 +65,7 @@ import (
"sigs.k8s.io/azuredisk-csi-driver/pkg/optimization"
"sigs.k8s.io/azuredisk-csi-driver/pkg/provisioner"
volumehelper "sigs.k8s.io/azuredisk-csi-driver/pkg/util"
"sigs.k8s.io/azuredisk-csi-driver/pkg/watcher"
"sigs.k8s.io/azuredisk-csi-driver/pkg/workflow"
azurecloudconsts "sigs.k8s.io/cloud-provider-azure/pkg/consts"
"sigs.k8s.io/controller-runtime/pkg/manager"
Expand Down Expand Up @@ -117,6 +120,7 @@ type DriverV2 struct {
kubeConfig *rest.Config
kubeClient *clientset.Clientset
azdiskClient azdisk.Interface
crdClient crdClientset.Interface
azDriverNodeInformer azdiskinformertypes.AzDriverNodeInformer
deviceChecker *deviceChecker
}
Expand Down Expand Up @@ -179,12 +183,28 @@ func (d *DriverV2) Run(endpoint, kubeconfig string, disableAVSetNodes, testingMo
klog.Fatalf("failed to get azdiskclient with kubeconfig (%s), error: %v. Exiting application...", kubeconfig, err)
}

d.crdClient, err = crdClientset.NewForConfig(d.kubeConfig)
if err != nil {
klog.Fatalf("failed to get crdclient with kubeconfig (%s), error: %v. Exiting application...", kubeconfig, err)
}

// d.crdProvisioner is set by NewFakeDriver for unit tests.
if d.crdProvisioner == nil {
d.crdProvisioner, err = provisioner.NewCrdProvisioner(d.kubeConfig, d.config.ObjectNamespace)
if err != nil {
klog.Fatalf("Failed to get crd provisioner. Error: %v", err)
}
kubeInformerFactory := informers.NewSharedInformerFactory(d.kubeClient, consts.DefaultInformerResync)
azInformerFactory := azdiskinformers.NewSharedInformerFactory(d.azdiskClient, consts.DefaultInformerResync)
crdInformerFactory := crdInformers.NewSharedInformerFactory(d.crdClient, consts.DefaultInformerResync)

nodeInformer := azureutils.NewNodeInformer(kubeInformerFactory)
azNodeInformer := azureutils.NewAzNodeInformer(azInformerFactory)
azVolumeAttachmentInformer := azureutils.NewAzVolumeAttachmentInformer(azInformerFactory)
azVolumeInformer := azureutils.NewAzVolumeInformer(azInformerFactory)
crdInformer := azureutils.NewCrdInformer(crdInformerFactory)

conditionWatcher := watcher.NewConditionWatcher(azInformerFactory, d.config.ObjectNamespace, azNodeInformer, azVolumeAttachmentInformer, azVolumeInformer)

d.crdProvisioner = provisioner.NewCrdProvisioner(d.azdiskClient, d.config.ObjectNamespace, conditionWatcher, provisioner.NewCachedReader(kubeInformerFactory, azInformerFactory, d.config.ObjectNamespace), crdInformer)

azureutils.StartInformersAndWaitForCacheSync(context.Background(), nodeInformer, azNodeInformer, azVolumeAttachmentInformer, azVolumeInformer, crdInformer)
}

// d.cloudProvisioner is set by NewFakeDriver for unit tests.
Expand Down Expand Up @@ -344,67 +364,54 @@ func (d *DriverV2) StartControllersAndDieOnExit(ctx context.Context) {
eventBroadcaster := record.NewBroadcaster()
eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: d.kubeClient.CoreV1().Events("")})
eventRecorder := eventBroadcaster.NewRecorder(clientgoscheme.Scheme, v1.EventSource{Component: consts.AzureDiskCSIDriverName})
crdClient, err := crdClientset.NewForConfig(d.kubeConfig)
if err != nil {
klog.Errorf("failed to initialize crd clientset. Error: %v. Exiting application...", err)
os.Exit(1)
}

sharedState := controller.NewSharedState(d.config, topologyKey, eventRecorder, mgr.GetClient(), d.crdProvisioner.GetDiskClientSet(), d.crdProvisioner, d.kubeClient, crdClient)
sharedState := controller.NewSharedState(d.config, topologyKey, eventRecorder, mgr.GetClient(), d.crdClient, d.kubeClient, d.crdProvisioner)

// Setup a new controller to clean-up AzDriverNodes
// objects for the nodes which get deleted
klog.V(2).Info("Initializing Node controller")
nodeReconciler, err := controller.NewNodeController(mgr, sharedState)
if err != nil {
klog.Errorf("Failed to initialize NodeController. Error: %v. Exiting application...", err)
os.Exit(1)
klog.Fatalf("Failed to initialize NodeController. Error: %v. Exiting application...", err)
}

klog.V(2).Info("Initializing AzVolumeAttachment controller")
attachReconciler, err := controller.NewAttachDetachController(mgr, d.cloudProvisioner, d.crdProvisioner, sharedState)
if err != nil {
klog.Errorf("Failed to initialize AzVolumeAttachmentController. Error: %v. Exiting application...", err)
os.Exit(1)
klog.Fatalf("Failed to initialize AzVolumeAttachmentController. Error: %v. Exiting application...", err)
}

klog.V(2).Info("Initializing Pod controller")
podReconciler, err := controller.NewPodController(mgr, sharedState)
if err != nil {
klog.Errorf("Failed to initialize PodController. Error: %v. Exiting application...", err)
os.Exit(1)
klog.Fatalf("Failed to initialize PodController. Error: %v. Exiting application...", err)
}

klog.V(2).Info("Initializing Replica controller")
_, err = controller.NewReplicaController(mgr, sharedState)
if err != nil {
klog.Errorf("Failed to initialize ReplicaController. Error: %v. Exiting application...", err)
os.Exit(1)
klog.Fatalf("Failed to initialize ReplicaController. Error: %v. Exiting application...", err)
}

klog.V(2).Info("Initializing AzVolume controller")
azvReconciler, err := controller.NewAzVolumeController(mgr, d.cloudProvisioner, sharedState)
if err != nil {
klog.Errorf("Failed to initialize AzVolumeController. Error: %v. Exiting application...", err)
os.Exit(1)
klog.Fatalf("Failed to initialize AzVolumeController. Error: %v. Exiting application...", err)
}

klog.V(2).Info("Initializing PV controller")
pvReconciler, err := controller.NewPVController(mgr, sharedState)
if err != nil {
klog.Errorf("Failed to initialize PVController. Error: %v. Exiting application...", err)
os.Exit(1)
klog.Fatalf("Failed to initialize PVController. Error: %v. Exiting application...", err)
}
klog.V(2).Info("Initializing VolumeAttachment controller")
_, err = controller.NewVolumeAttachmentController(mgr, sharedState)
if err != nil {
klog.Errorf("Failed to initialize VolumeAttachment Controller. Error: %v. Exiting application...", err)
os.Exit(1)
klog.Fatalf("Failed to initialize VolumeAttachment Controller. Error: %v. Exiting application...", err)
}
err = controller.WatchObject(mgr, source.Kind{Type: &v1.Namespace{}})
if err != nil {
klog.Errorf("Failed to watch Namespace. Error: %v. Exiting application...", err)
os.Exit(1)
klog.Fatalf("Failed to watch Namespace. Error: %v. Exiting application...", err)
}

// This goroutine is preserved for leader controller manager
Expand Down Expand Up @@ -439,8 +446,7 @@ func (d *DriverV2) StartControllersAndDieOnExit(ctx context.Context) {

klog.V(2).Info("Starting controller manager")
if err := mgr.Start(ctx); err != nil {
klog.Errorf("Controller manager exited: %v", err)
os.Exit(1)
klog.Fatalf("Controller manager exited: %v", err)
}
// If manager exits, exit the application
// as recommended for the processes doing
Expand All @@ -457,7 +463,7 @@ func (d *DriverV2) registerAzDriverNodeOrDie(ctx context.Context) {

nodeSelector := fmt.Sprintf("metadata.name=%s", d.NodeID)

azdiskInformer := azdiskinformers.NewSharedInformerFactoryWithOptions(
azdiskInformerFactory := azdiskinformers.NewSharedInformerFactoryWithOptions(
d.azdiskClient,
consts.DefaultInformerResync,
azdiskinformers.WithNamespace(d.config.ObjectNamespace),
Expand All @@ -466,7 +472,7 @@ func (d *DriverV2) registerAzDriverNodeOrDie(ctx context.Context) {
}),
)

d.azDriverNodeInformer = azdiskInformer.Disk().V1beta2().AzDriverNodes()
d.azDriverNodeInformer = azdiskInformerFactory.Disk().V1beta2().AzDriverNodes()
d.azDriverNodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
DeleteFunc: func(obj interface{}) {
if azDriverNode, ok := obj.(*azdiskv1beta2.AzDriverNode); ok && strings.EqualFold(azDriverNode.Name, d.NodeID) {
Expand All @@ -475,7 +481,7 @@ func (d *DriverV2) registerAzDriverNodeOrDie(ctx context.Context) {
},
})

go azdiskInformer.Start(ctx.Done())
azdiskInformerFactory.Start(ctx.Done())

if !cache.WaitForCacheSync(ctx.Done(), d.azDriverNodeInformer.Informer().HasSynced) {
klog.Fatal("failed to sync azdrivernode informer")
Expand Down
2 changes: 2 additions & 0 deletions pkg/azuredisk/interface_v2.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
azdiskv1beta2 "sigs.k8s.io/azuredisk-csi-driver/pkg/apis/azuredisk/v1beta2"
azdisk "sigs.k8s.io/azuredisk-csi-driver/pkg/apis/client/clientset/versioned"
consts "sigs.k8s.io/azuredisk-csi-driver/pkg/azureconstants"
"sigs.k8s.io/azuredisk-csi-driver/pkg/watcher"
)

type CrdProvisioner interface {
Expand All @@ -39,6 +40,7 @@ type CrdProvisioner interface {
GetAzVolumeAttachment(ctx context.Context, volumeID string, nodeID string) (*azdiskv1beta2.AzVolumeAttachment, error)
ExpandVolume(ctx context.Context, volumeID string, capacityRange *azdiskv1beta2.CapacityRange, secrets map[string]string) (*azdiskv1beta2.AzVolumeStatusDetail, error)
GetDiskClientSet() azdisk.Interface
GetConditionWatcher() *watcher.ConditionWatcher
IsDriverUninstall() bool
}

Expand Down
15 changes: 15 additions & 0 deletions pkg/azuredisk/mockprovisioner/interface.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

119 changes: 119 additions & 0 deletions pkg/azureutils/informer_common.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
/*
Copyright 2022 The Kubernetes 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 azureutils

import (
"context"

crdInformers "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions"
"k8s.io/client-go/informers"
"k8s.io/client-go/tools/cache"
"k8s.io/klog/v2"
azdiskinformers "sigs.k8s.io/azuredisk-csi-driver/pkg/apis/client/informers/externalversions"
)

type GenericInformer interface {
Start(stopCh <-chan struct{})
Informer() cache.SharedIndexInformer
}

type GenericAzInformerInfo struct {
factory azdiskinformers.SharedInformerFactory
informer cache.SharedIndexInformer
}

func (i *GenericAzInformerInfo) Start(stopCh <-chan struct{}) {
i.factory.Start(stopCh)
}

func (i *GenericAzInformerInfo) Informer() cache.SharedIndexInformer {
return i.informer
}

func NewAzNodeInformer(factory azdiskinformers.SharedInformerFactory) GenericInformer {
return &GenericAzInformerInfo{
factory: factory,
informer: factory.Disk().V1beta2().AzDriverNodes().Informer(),
}
}

func NewAzVolumeAttachmentInformer(factory azdiskinformers.SharedInformerFactory) GenericInformer {
return &GenericAzInformerInfo{
factory: factory,
informer: factory.Disk().V1beta2().AzVolumeAttachments().Informer(),
}
}

func NewAzVolumeInformer(factory azdiskinformers.SharedInformerFactory) GenericInformer {
return &GenericAzInformerInfo{
factory: factory,
informer: factory.Disk().V1beta2().AzVolumes().Informer(),
}
}

type GenericKubeInformer struct {
factory informers.SharedInformerFactory
informer cache.SharedIndexInformer
}

func (i *GenericKubeInformer) Start(stopCh <-chan struct{}) {
i.factory.Start(stopCh)
}

func (i *GenericKubeInformer) Informer() cache.SharedIndexInformer {
return i.informer
}

func NewNodeInformer(factory informers.SharedInformerFactory) GenericInformer {
return &GenericKubeInformer{
factory: factory,
informer: factory.Core().V1().Nodes().Informer(),
}
}

type GenericCrdInformer struct {
factory crdInformers.SharedInformerFactory
informer cache.SharedIndexInformer
}

func NewCrdInformer(factory crdInformers.SharedInformerFactory) GenericInformer {
return &GenericCrdInformer{
factory: factory,
informer: factory.Apiextensions().V1().CustomResourceDefinitions().Informer(),
}
}

func (i *GenericCrdInformer) Start(stopCh <-chan struct{}) {
i.factory.Start(stopCh)
}

func (i *GenericCrdInformer) Informer() cache.SharedIndexInformer {
return i.informer
}

func StartInformersAndWaitForCacheSync(ctx context.Context, factories ...GenericInformer) {
informerSynced := make([]cache.InformerSynced, 0)
for _, factory := range factories {
factory.Start(ctx.Done())
informerSynced = append(informerSynced, factory.Informer().HasSynced)
}

synced := cache.WaitForCacheSync(ctx.Done(), informerSynced...)
if !synced {
klog.Fatalf("Unable to sync caches")
}
}
Loading

0 comments on commit c9806d5

Please sign in to comment.