add PlugableReconciler wrap the main reconcilers

cmd/controller/antplugins/localstorage.go

@@ -84,8 +84,8 @@ func (r *ReportLocalStoragePlugin) Reconcile(ctx *plugin.Context) (result plugin
 		r.PoolUtil = kubeutil.NewStoragePoolUtil(cli)
 	}
 
-	volume, isVolume = ctx.Object.(*v1.AntstorVolume)
-	pool, isPool = ctx.Object.(*v1.StoragePool)
+	volume, isVolume = ctx.ReqCtx.Object.(*v1.AntstorVolume)
+	pool, isPool = ctx.ReqCtx.Object.(*v1.StoragePool)
 
 	if !isVolume && !isPool {
 		err = fmt.Errorf("obj is not *v1.AntstorVolume or *v1.StoragePool")

cmd/controller/antplugins/patchpv.go

@@ -32,7 +32,7 @@ func (p *PatchPVPlugin) Reconcile(ctx *plugin.Context) (result plugin.Result) {
 
 	var (
 		log    = ctx.Log
-		obj    = ctx.Object
+		obj    = ctx.ReqCtx.Object
 		volume *v1.AntstorVolume
 		ok     bool
 		err    error

cmd/controller/main.go

@@ -6,27 +6,23 @@ import (
 	antplugin "code.alipay.com/dbplatform/node-disk-controller/cmd/controller/antplugins"
 	antfilter "code.alipay.com/dbplatform/node-disk-controller/cmd/controller/antplugins/filter"
 	"code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/controllers"
-	"code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/scheduler"
 	"code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/scheduler/filter"
 	"code.alipay.com/dbplatform/node-disk-controller/pkg/csi"
 )
 
 func main() {
 	// add plugins
-	controllers.RegisterPlugins([]controllers.PluginFactoryFunc{
+	controllers.RegisterPluginsInPoolReconciler([]controllers.PluginFactoryFunc{
 		antplugin.NewReportLocalStoragePlugin,
-	}, []controllers.PluginFactoryFunc{
+	})
+	controllers.RegisterPluginsInVolumeReconciler([]controllers.PluginFactoryFunc{
 		antplugin.NewReportLocalStoragePlugin,
 		antplugin.NewPatchPVPlugin,
 	})
 
 	// add filters
-	filter.RegisterFilter("MinLocalStorage", antfilter.MinLocalStorageFilterFunc)
 	filter.RegisterFilter("ObReplica", antfilter.ObReplicaFilterFunc)
 
-	// for volumegroup
-	scheduler.RegisterVolumeGroupPickSizeFn("MinLocalStorage", antfilter.GetAllocatableRemoveVolumeSize)
-
 	cmd := controllers.NewApplicationCmd()
 	// add CSI command
 	cmd.AddCommand(csi.NewCSICommand())

hack/deploy/aio-lvs/050-configmap.yaml

@@ -7,6 +7,7 @@ data:
   config.yaml: |
     scheduler:
       maxRemoteVolumeCount: 3
+      minLocalStoragePct: 20
       filters:
       - Basic
       - Affinity

hack/deploy/lvm/050-configmap.yaml

@@ -7,6 +7,7 @@ data:
   config.yaml: |
     scheduler:
       maxRemoteVolumeCount: 3
+      minLocalStoragePct: 20
       filters:
       - Basic
       - Affinity

pkg/controller/manager/config/config.go

@@ -28,6 +28,8 @@ type SchedulerConfig struct {
 	// NodeCacheSelector specify which nodes are cached to Node Informer.
 	// Empty selector means all nodes are allowd to be cached.
 	NodeCacheSelector map[string]string `json:"nodeCacheSelector" yaml:"nodeCacheSelector"`
+	// MinLocalStoragePct defines the minimun percentage of local storage to be reserved on one node.
+	MinLocalStoragePct int `json:"minLocalStoragePct" yaml:"minLocalStoragePct"`
 }
 
 type NoScheduleConfig struct {

pkg/controller/manager/controllers/manager.go

@@ -85,44 +85,87 @@ func NewAndInitControllerManager(req NewManagerRequest) manager.Manager {
 		scheduler  = sched.NewScheduler(req.ControllerConfig)
 	)
 
-	// setup StoragePoolReconciler
-	poolReconciler := &reconciler.StoragePoolReconciler{
-		Plugable: plugin.NewPluginList(),
+	poolReconciler := reconciler.PlugableReconciler{
 		Client:   mgr.GetClient(),
-		Log:      rt.Log.WithName("controllers").WithName("StoragePool"),
-		State:    stateObj,
-		PoolUtil: poolUtil,
-		KubeCli:  kubeClient,
-		Lock:     misc.NewResourceLocks(),
+		Plugable: plugin.NewPluginList(),
+
+		Log:     rt.Log.WithName("Controller:StoragePool"),
+		KubeCli: kubeClient,
+		State:   stateObj,
+
+		Concurrency: 4,
+		MainHandler: &reconciler.StoragePoolReconcileHandler{
+			Client:   mgr.GetClient(),
+			State:    stateObj,
+			PoolUtil: poolUtil,
+			KubeCli:  kubeClient,
+		},
+		WatchType: &v1.StoragePool{},
+	}
+	if err = poolReconciler.SetupWithManager(mgr); err != nil {
+		klog.Error(err, "unable to create controller StoragePoolReconciler")
+		os.Exit(1)
 	}
 
 	// setup AntstorVolumeReconciler
-	volReconciler := &reconciler.AntstorVolumeReconciler{
-		Plugable:    plugin.NewPluginList(),
-		Client:      mgr.GetClient(),
-		Log:         rt.Log.WithName("controllers").WithName("AntstorVolume"),
-		State:       stateObj,
-		AntstoreCli: antstorCli,
-		Scheduler:   scheduler,
-		// EventRecorder for AntstorVolume
-		EventRecorder: mgr.GetEventRecorderFor("AntstorVolume"),
+	volReconciler := reconciler.PlugableReconciler{
+		Client:   mgr.GetClient(),
+		Plugable: plugin.NewPluginList(),
+
+		Log:     rt.Log.WithName("Controller:AntstorVolume"),
+		KubeCli: kubeClient,
+		State:   stateObj,
+
+		Concurrency: 1,
+		MainHandler: &reconciler.AntstorVolumeReconcileHandler{
+			Client:      mgr.GetClient(),
+			State:       stateObj,
+			AntstoreCli: antstorCli,
+			Scheduler:   scheduler,
+		},
+		WatchType: &v1.AntstorVolume{},
 	}
+	if err = volReconciler.SetupWithManager(mgr); err != nil {
+		klog.Error(err, "unable to create controller VolumeReconciler")
+		os.Exit(1)
+	}
+
+	// setup AntstorVolumeGroupReconciler
+	volGroupReconciler := reconciler.PlugableReconciler{
+		Client:   mgr.GetClient(),
+		Plugable: plugin.NewPluginList(),
 
-	volGroupReconciler := &reconciler.AntstorVolumeGroupReconciler{
-		Client:    mgr.GetClient(),
-		Plugable:  plugin.NewPluginList(),
-		Log:       rt.Log.WithName("controllers").WithName("AntstorVolumeGroup"),
-		Scheduler: scheduler,
-		State:     stateObj,
+		Log:     rt.Log.WithName("Controller:AntstorVolumeGroup"),
+		KubeCli: kubeClient,
+		State:   stateObj,
+
+		Concurrency: 1,
+		MainHandler: &reconciler.AntstorVolumeGroupReconcileHandler{
+			Client:    mgr.GetClient(),
+			Scheduler: scheduler,
+			State:     stateObj,
+		},
+		WatchType: &v1.AntstorVolumeGroup{},
+	}
+	if err = volGroupReconciler.SetupWithManager(mgr); err != nil {
+		klog.Error(err, "unable to create controller VolumeGroupReconciler")
+		os.Exit(1)
 	}
 
 	// setup AntstorDataControlReconciler
-	dataControlReconciler := &reconciler.AntstorDataControlReconciler{
+	dataControlReconciler := reconciler.PlugableReconciler{
 		Client:   mgr.GetClient(),
 		Plugable: plugin.NewPluginList(),
 
-		Log:   rt.Log.WithName("controllers").WithName("DataControl"),
-		State: stateObj,
+		Log:     rt.Log.WithName("Controller:AntstorDataControl"),
+		KubeCli: kubeClient,
+		State:   stateObj,
+
+		Concurrency: 1,
+		MainHandler: &reconciler.AntstorDataControlReconcileHandler{
+			Client: mgr.GetClient(),
+		},
+		WatchType: &v1.AntstorDataControl{},
 	}
 	if err = dataControlReconciler.SetupWithManager(mgr); err != nil {
 		klog.Error(err, "unable to create controller AntstorDataControlReconciler")
@@ -169,20 +212,6 @@ func NewAndInitControllerManager(req NewManagerRequest) manager.Manager {
 		dataControlReconciler.RegisterPlugin(p)
 	}
 
-	// setup StoragePool/Volume Reconciler
-	if err = poolReconciler.SetupWithManager(mgr); err != nil {
-		klog.Error(err, "unable to create controller StoragePoolReconciler")
-		os.Exit(1)
-	}
-	if err = volReconciler.SetupWithManager(mgr); err != nil {
-		klog.Error(err, "unable to create controller VolumeReconciler")
-		os.Exit(1)
-	}
-	if err = volGroupReconciler.SetupWithManager(mgr); err != nil {
-		klog.Error(err, "unable to create controller VolumeGroupReconciler")
-		os.Exit(1)
-	}
-
 	// setup SnapshotReconsiler
 	snapshotReconciler := &reconciler.SnapshotReconciler{
 		// kube client

pkg/controller/manager/controllers/plugin.go

@@ -26,8 +26,11 @@ func init() {
 	DataControlReconcilerPluginCreaters = append(DataControlReconcilerPluginCreaters, NewMetaSyncerPlugin)
 }
 
-func RegisterPlugins(poolPlugins, volumePlugins []PluginFactoryFunc) {
+func RegisterPluginsInPoolReconciler(poolPlugins []PluginFactoryFunc) {
 	PoolReconcilerPluginCreaters = append(PoolReconcilerPluginCreaters, poolPlugins...)
+}
+
+func RegisterPluginsInVolumeReconciler(volumePlugins []PluginFactoryFunc) {
 	VolumeReconcilerPluginCreaters = append(VolumeReconcilerPluginCreaters, volumePlugins...)
 }