Merge pull request #10 from silentred/main

setup watches and indexes to PlugableReconciler

cmd/controller/antplugins/localstorage.go

@@ -6,7 +6,6 @@ import (
  "strconv"
 
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
- "k8s.io/apimachinery/pkg/labels"
 
  v1 "code.alipay.com/dbplatform/node-disk-controller/pkg/api/volume.antstor.alipay.com/v1"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/kubeutil"
@@ -119,32 +118,35 @@ func (r *ReportLocalStoragePlugin) Reconcile(ctx *plugin.Context) (result plugin
 
  // After remote volume is scheduled, report the local storage size to Node
  if isVolume && volume != nil && volume.Spec.TargetNodeId != "" && !volume.IsLocal() {
- var localStorePct int
- var volInState *v1.AntstorVolume
- node, err := stateObj.GetNodeByNodeID(volume.Spec.TargetNodeId)
+ var node *state.Node
+ node, err = stateObj.GetNodeByNodeID(volume.Spec.TargetNodeId)
  if err != nil {
  log.Error(err, "find node failed")
  return plugin.Result{Error: err}
  }
- sp := node.Pool
-
- for _, item := range r.ReportLocalConfigs {
- selector, err := metav1.LabelSelectorAsSelector(&item.LabelSelector)
- if err != nil {
- log.Error(err, "LabelSelectorAsSelector failed", "selector", item.LabelSelector)
- continue
- }
- if selector.Matches(labels.Set(sp.Spec.NodeInfo.Labels)) && item.EnableDefault {
- localStorePct = item.DefaultLocalStoragePct
- log.Info("matched local-storage percentage", "pct", localStorePct)
+ var sp = node.Pool
+
+ /*
+ var localStorePct int
+ var volInState *v1.AntstorVolume
+ for _, item := range r.ReportLocalConfigs {
+ selector, err := metav1.LabelSelectorAsSelector(&item.LabelSelector)
+ if err != nil {
+ log.Error(err, "LabelSelectorAsSelector failed", "selector", item.LabelSelector)
+ continue
+ }
+ if selector.Matches(labels.Set(sp.Spec.NodeInfo.Labels)) && item.EnableDefault {
+ localStorePct = item.DefaultLocalStoragePct
+ log.Info("matched local-storage percentage", "pct", localStorePct)
+ }
  }
- }
 
- volInState, err = node.GetVolumeByID(volume.Spec.Uuid)
- if err == nil {
- log.Info("copy volume into state")
- *volInState = *volume
- }
+ volInState, err = node.GetVolumeByID(volume.Spec.Uuid)
+ if err == nil {
+ log.Info("copy volume into state")
+ *volInState = *volume
+ }
+ */
 
  var expectLocalSize = CalculateLocalStorageCapacity(node)
  var localSizeStr = strconv.Itoa(int(expectLocalSize))

cmd/controller/antplugins/patchpv.go

@@ -44,7 +44,6 @@ func (p *PatchPVPlugin) Reconcile(ctx *plugin.Context) (result plugin.Result) {
  return plugin.Result{}
  }
 
- log.Info("running PatchPVPlugin")
  // get pv name from label
  var pvName string
  if val, has := volume.Labels[v1.VolumePVNameLabelKey]; has {
@@ -53,11 +52,15 @@ func (p *PatchPVPlugin) Reconcile(ctx *plugin.Context) (result plugin.Result) {
  pvName = volume.Name
  }
 
- err = p.PvUtil.SetTargetNodeName(pvName, volume.Spec.TargetNodeId)
- if err != nil {
- log.Error(err, "updating PV label failed")
- return plugin.Result{
- Error: err,
+ if volume.Spec.TargetNodeId != "" {
+ log.Info("patching PV", "nodeId", volume.Spec.TargetNodeId, "pvName", pvName)
+
+ err = p.PvUtil.SetTargetNodeName(pvName, volume.Spec.TargetNodeId)
+ if err != nil {
+ log.Error(err, "updating PV label failed")
+ return plugin.Result{
+ Error: err,
+ }
  }
  }
 

pkg/controller/manager/controllers/cmd.go

@@ -9,19 +9,16 @@ import (
  "code.alipay.com/dbplatform/node-disk-controller/pkg/agent"
  v1 "code.alipay.com/dbplatform/node-disk-controller/pkg/api/volume.antstor.alipay.com/v1"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/config"
- "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/reconciler/handler"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/state"
  hostnvme "code.alipay.com/dbplatform/node-disk-controller/pkg/host-nvme"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/util"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/util/misc"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/version"
  "github.com/spf13/cobra"
  "github.com/spf13/pflag"
- corev1 "k8s.io/api/core/v1"
  "k8s.io/apimachinery/pkg/runtime"
  "k8s.io/client-go/kubernetes"
  cligoscheme "k8s.io/client-go/kubernetes/scheme"
- "k8s.io/client-go/tools/cache"
  "k8s.io/klog/v2"
  rt "sigs.k8s.io/controller-runtime"
 )
@@ -173,18 +170,21 @@ func (o *OperatorOption) Run() {
 
  ctx := rt.SetupSignalHandler()
 
- // create NodeInformer to sync nodes to cache
- nodeInformer, err := mgr.GetCache().GetInformer(ctx, &corev1.Node{})
- if err != nil {
- klog.Fatal(err)
- }
- nodeHandler := &handler.NodeEventHandler{
- Cfg: cfg,
- }
- nodeInformer.AddEventHandler(cache.FilteringResourceEventHandler{
- FilterFunc: nodeHandler.FilterObject,
- Handler: nodeHandler,
- })
+ /*
+ // create NodeInformer to sync nodes to cache
+ // moved to pool reconciler
+ nodeInformer, err := mgr.GetCache().GetInformer(ctx, &corev1.Node{})
+ if err != nil {
+ klog.Fatal(err)
+ }
+ nodeHandler := &handler.NodeEventHandler{
+ Cfg: cfg,
+ }
+ nodeInformer.AddEventHandler(cache.FilteringResourceEventHandler{
+ FilterFunc: nodeHandler.FilterObject,
+ Handler: nodeHandler,
+ })
+ */
 
  go func() {
  klog.Info("manager start working")

pkg/controller/manager/controllers/manager.go

@@ -5,25 +5,29 @@ import (
 
  // Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
  // to ensure that exec-entrypoint and run can make use of them.
+ corev1 "k8s.io/api/core/v1"
+ "k8s.io/apimachinery/pkg/runtime"
+ "k8s.io/client-go/kubernetes"
+ _ "k8s.io/client-go/plugin/pkg/client/auth"
+ "k8s.io/client-go/rest"
+ "k8s.io/klog/v2"
+ rt "sigs.k8s.io/controller-runtime"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+ "sigs.k8s.io/controller-runtime/pkg/healthz"
+ "sigs.k8s.io/controller-runtime/pkg/log/zap"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/source"
 
  v1 "code.alipay.com/dbplatform/node-disk-controller/pkg/api/volume.antstor.alipay.com/v1"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/kubeutil"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/config"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/reconciler"
+ "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/reconciler/handler"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/reconciler/plugin"
  sched "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/scheduler"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/state"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/generated/clientset/versioned"
  "code.alipay.com/dbplatform/node-disk-controller/pkg/util/misc"
- "k8s.io/apimachinery/pkg/runtime"
- "k8s.io/client-go/kubernetes"
- _ "k8s.io/client-go/plugin/pkg/client/auth"
- "k8s.io/client-go/rest"
- "k8s.io/klog/v2"
- rt "sigs.k8s.io/controller-runtime"
- "sigs.k8s.io/controller-runtime/pkg/healthz"
- "sigs.k8s.io/controller-runtime/pkg/log/zap"
- "sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
 type B64EncodedMysqlDSN string
@@ -100,7 +104,15 @@ func NewAndInitControllerManager(req NewManagerRequest) manager.Manager {
  PoolUtil: poolUtil,
  KubeCli: kubeClient,
  },
- WatchType: &v1.StoragePool{},
+ ForType: &v1.StoragePool{},
+ Watches: []reconciler.WatchObject{
+ {
+ Source: &source.Kind{Type: &corev1.Node{}},
+ EventHandler: &handler.NodeEventHandler{
+ Cfg: req.ControllerConfig,
+ },
+ },
+ },
  }
  if err = poolReconciler.SetupWithManager(mgr); err != nil {
  klog.Error(err, "unable to create controller StoragePoolReconciler")
@@ -123,7 +135,38 @@ func NewAndInitControllerManager(req NewManagerRequest) manager.Manager {
  AntstoreCli: antstorCli,
  Scheduler: scheduler,
  },
- WatchType: &v1.AntstorVolume{},
+ ForType: &v1.AntstorVolume{},
+ Watches: []reconciler.WatchObject{
+ {
+ Source: &source.Kind{Type: &v1.AntstorVolume{}},
+ EventHandler: &handler.VolumeEventHandler{
+ State: stateObj,
+ }},
+ },
+ Indexes: []reconciler.IndexObject{
+ {
+ Obj: &v1.AntstorVolume{},
+ Field: v1.IndexKeyUUID,
+ ExtractValue: func(rawObj client.Object) []string {
+ // grab the volume, extract the uuid
+ if vol, ok := rawObj.(*v1.AntstorVolume); ok {
+ return []string{vol.Spec.Uuid}
+ }
+ return nil
+ },
+ },
+ {
+ Obj: &v1.AntstorVolume{},
+ Field: v1.IndexKeyTargetNodeID,
+ ExtractValue: func(rawObj client.Object) []string {
+ // grab the volume, extract the targetNodeId
+ if vol, ok := rawObj.(*v1.AntstorVolume); ok {
+ return []string{vol.Spec.TargetNodeId}
+ }
+ return nil
+ },
+ },
+ },
  }
  if err = volReconciler.SetupWithManager(mgr); err != nil {
  klog.Error(err, "unable to create controller VolumeReconciler")
@@ -145,7 +188,7 @@ func NewAndInitControllerManager(req NewManagerRequest) manager.Manager {
  Scheduler: scheduler,
  State: stateObj,
  },
- WatchType: &v1.AntstorVolumeGroup{},
+ ForType: &v1.AntstorVolumeGroup{},
  }
  if err = volGroupReconciler.SetupWithManager(mgr); err != nil {
  klog.Error(err, "unable to create controller VolumeGroupReconciler")
@@ -165,7 +208,7 @@ func NewAndInitControllerManager(req NewManagerRequest) manager.Manager {
  MainHandler: &reconciler.AntstorDataControlReconcileHandler{
  Client: mgr.GetClient(),
  },
- WatchType: &v1.AntstorDataControl{},
+ ForType: &v1.AntstorDataControl{},
  }
  if err = dataControlReconciler.SetupWithManager(mgr); err != nil {
  klog.Error(err, "unable to create controller AntstorDataControlReconciler")

pkg/controller/manager/reconciler/handler/node_handler.go

@@ -1,44 +1,84 @@
 package handler
 
 import (
- "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/config"
- corev1 "k8s.io/api/core/v1"
  "k8s.io/apimachinery/pkg/labels"
+ "k8s.io/apimachinery/pkg/types"
  "k8s.io/client-go/tools/cache"
+ "k8s.io/client-go/util/workqueue"
  "k8s.io/klog/v2"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+ "sigs.k8s.io/controller-runtime/pkg/event"
+ "sigs.k8s.io/controller-runtime/pkg/handler"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+ v1 "code.alipay.com/dbplatform/node-disk-controller/pkg/api/volume.antstor.alipay.com/v1"
+ "code.alipay.com/dbplatform/node-disk-controller/pkg/controller/manager/config"
 )
 
 var (
  _ cache.ResourceEventHandler = &NodeEventHandler{}
+ _ handler.EventHandler = &NodeEventHandler{}
 )
 
 type NodeEventHandler struct {
  Cfg config.Config
 }
 
-func (e *NodeEventHandler) FilterObject(obj interface{}) bool {
- /*
- key, err := cache.MetaNamespaceKeyFunc(obj)
- if err != nil {
- return false
- }
-
- _, name, err := cache.SplitMetaNamespaceKey(key)
- if err != nil {
- return false
- }
- */
-
- if len(e.Cfg.Scheduler.NodeCacheSelector) == 0 {
- return true
+// Create implements EventHandler.
+func (e *NodeEventHandler) Create(evt event.CreateEvent, q workqueue.RateLimitingInterface) {
+ if !validateNode(&e.Cfg, evt.Object) {
+ return
+ }
+
+ // Node name is same with StoragePool name
+ q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+ Namespace: v1.DefaultNamespace,
+ Name: evt.Object.GetName(),
+ }})
+}
+
+// Update implements EventHandler.
+func (e *NodeEventHandler) Update(evt event.UpdateEvent, q workqueue.RateLimitingInterface) {
+ if !validateNode(&e.Cfg, evt.ObjectNew) {
+ return
  }
 
- if node, ok := obj.(*corev1.Node); ok {
- selector := labels.SelectorFromSet(labels.Set(e.Cfg.Scheduler.NodeCacheSelector))
- if selector.Matches(labels.Set(node.Labels)) {
- klog.Info("matched node to cache: ", node.Name)
- return true
- }
+ // Node name is same with StoragePool name
+ q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+ Namespace: v1.DefaultNamespace,
+ Name: evt.ObjectNew.GetName(),
+ }})
+}
+
+// Delete implements EventHandler.
+func (e *NodeEventHandler) Delete(evt event.DeleteEvent, q workqueue.RateLimitingInterface) {
+ if !validateNode(&e.Cfg, evt.Object) {
+ return
+ }
+
+ // Node name is same with StoragePool name
+ q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+ Namespace: v1.DefaultNamespace,
+ Name: evt.Object.GetName(),
+ }})
+}
+
+// Generic implements EventHandler.
+func (e *NodeEventHandler) Generic(evt event.GenericEvent, q workqueue.RateLimitingInterface) {
+ if !validateNode(&e.Cfg, evt.Object) {
+ return
+ }
+
+ // Node name is same with StoragePool name
+ q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+ Namespace: v1.DefaultNamespace,
+ Name: evt.Object.GetName(),
+ }})
+}
+
+func (e *NodeEventHandler) FilterObject(obj interface{}) bool {
+ if metaObj, ok := obj.(client.Object); ok {
+ return validateNode(&e.Cfg, metaObj)
  }
 
  return false
@@ -70,3 +110,23 @@ func (e *NodeEventHandler) OnDelete(obj interface{}) {
  }
  klog.Infof("delete Node %s", key)
 }
+
+func validateNode(cfg *config.Config, obj client.Object) bool {
+ if obj == nil {
+ klog.Error(nil, "NodeEvent received with nil object")
+ return false
+ }
+
+ // no config of NodeCacheSelector means appcet all nodes events
+ if len(cfg.Scheduler.NodeCacheSelector) == 0 {
+ return true
+ }
+
+ selector := labels.SelectorFromSet(labels.Set(cfg.Scheduler.NodeCacheSelector))
+ if selector.Matches(labels.Set(obj.GetLabels())) {
+ klog.Info("matched node to cache: ", obj.GetName())
+ return true
+ }
+
+ return false
+}