add work queue for status controller (#99)

Signed-off-by: chenxianpao <chenxianpao@huawei.com>

cmd/controller-manager/app/controllermanager.go

@@ -150,7 +150,9 @@ func setupControllers(mgr controllerruntime.Manager, stopChan <-chan struct{}) {
 		KubeClientSet:   kubeClientSet,
 		InformerManager: informermanager.NewMultiClusterInformerManager(),
 		StopChan:        stopChan,
+		WorkerNumber:    1,
 	}
+	workStatusController.RunWorkQueue()
 	if err := workStatusController.SetupWithManager(mgr); err != nil {
 		klog.Fatalf("Failed to setup work status controller: %v", err)
 	}

pkg/controllers/execution/execution_controller.go

@@ -179,7 +179,7 @@ func (c *Controller) syncToMemberClusters(memberCluster *v1alpha1.MemberCluster,
 			klog.Errorf("failed to unmarshal workload, error is: %v", err)
 			return err
 		}
-
+		c.setOwnerLabel(workload, propagationWork)
 		applied := c.isResourceApplied(&propagationWork.Status)
 		if applied {
 			err = c.updateResource(memberClusterDynamicClient, workload)
@@ -204,6 +204,16 @@ func (c *Controller) syncToMemberClusters(memberCluster *v1alpha1.MemberCluster,
 	return nil
 }
 
+// setOwnerLabel adds ownerLabel for workload that will be applied to member cluster.
+func (c *Controller) setOwnerLabel(workload *unstructured.Unstructured, propagationWork *propagationstrategy.PropagationWork) {
+	workloadLabel := workload.GetLabels()
+	if workloadLabel == nil {
+		workloadLabel = make(map[string]string, 1)
+	}
+	workloadLabel[util.OwnerLabel] = names.GenerateOwnerLabelValue(propagationWork.GetNamespace(), propagationWork.GetName())
+	workload.SetLabels(workloadLabel)
+}
+
 // deleteResource delete resource in member cluster
 func (c *Controller) deleteResource(memberClusterDynamicClient *util.DynamicClusterClient, workload *unstructured.Unstructured) error {
 	dynamicResource, err := restmapper.GetGroupVersionResource(c.RESTMapper, workload.GroupVersionKind())

pkg/controllers/status/workstatus_controller.go

@@ -3,6 +3,7 @@ package status
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -37,6 +38,8 @@ type PropagationWorkStatusController struct {
 	InformerManager informermanager.MultiClusterInformerManager
 	eventHandler    cache.ResourceEventHandler // eventHandler knows how to handle events from the member cluster.
 	StopChan        <-chan struct{}
+	WorkerNumber    int              // WorkerNumber is the number of worker goroutines
+	worker          util.AsyncWorker // worker process resources periodic from rateLimitingQueue.
 }
 
 // Reconcile performs a full reconciliation for the object referred to by the Request.
@@ -76,11 +79,53 @@ func (c *PropagationWorkStatusController) buildResourceInformers(work *v1alpha1.
 // getEventHandler return callback function that knows how to handle events from the member cluster.
 func (c *PropagationWorkStatusController) getEventHandler() cache.ResourceEventHandler {
 	if c.eventHandler == nil {
-		c.eventHandler = informermanager.NewHandlerOnAllEvents(c.syncPropagationWorkStatus)
+		c.eventHandler = informermanager.NewHandlerOnAllEvents(c.worker.EnqueueRateLimited)
 	}
 	return c.eventHandler
 }
 
+// RunWorkQueue initializes worker and run it, worker will process resource asynchronously.
+func (c *PropagationWorkStatusController) RunWorkQueue() {
+	c.worker = util.NewAsyncWorker(c.syncPropagationWorkStatus, "work-status", time.Second)
+	c.worker.Run(c.WorkerNumber, c.StopChan)
+}
+
+// syncPropagationWorkStatus will find propagationWork by label in workload, then update resource status to propagationWork status.
+// label example: "karmada.io/created-by: karmada-es-member-cluster-1.default-deployment-nginx"
+// TODO(chenxianpao): sync workload status to propagationWork status.
+func (c *PropagationWorkStatusController) syncPropagationWorkStatus(key string) error {
+	obj, err := c.getObjectFromCache(key)
+	if err != nil {
+		return err
+	}
+	klog.Infof("sync workload %s/%s/%s", obj.GetKind(), obj.GetNamespace(), obj.GetName())
+	return nil
+}
+
+// getObjectFromCache gets full object information from cache by key in worker queue.
+func (c *PropagationWorkStatusController) getObjectFromCache(key string) (*unstructured.Unstructured, error) {
+	clusterWorkload, err := util.SplitMetaKey(key)
+	if err != nil {
+		klog.Errorf("Couldn't get key for %s. Error: %v.", key, err)
+		return nil, err
+	}
+	gvr, err := restmapper.GetGroupVersionResource(c.RESTMapper, clusterWorkload.GVK)
+	if err != nil {
+		klog.Errorf("Failed to get GVR from GVK %s. Error: %v", clusterWorkload.GVK, err)
+		return nil, err
+	}
+
+	lister := c.InformerManager.GetSingleClusterManager(clusterWorkload.Cluster).Lister(gvr)
+	var obj runtime.Object
+	obj, err = lister.Get(clusterWorkload.GetListerKey())
+	if err != nil {
+		klog.Errorf("Failed to get obj %s/%s/%s from cache in cluster %s. Error: %v.", clusterWorkload.GVK.Kind,
+			clusterWorkload.Namespace, clusterWorkload.Name, clusterWorkload.Cluster, err)
+		return nil, err
+	}
+	return obj.(*unstructured.Unstructured), nil
+}
+
 // registerInformersAndStart builds informer manager for cluster if it doesn't exist, then constructs informers for gvr
 // and start it.
 func (c *PropagationWorkStatusController) registerInformersAndStart(work *v1alpha1.PropagationWork) error {
@@ -156,15 +201,6 @@ func (c *PropagationWorkStatusController) getSingleClusterManager(memberClusterN
 	return singleClusterInformerManager, nil
 }
 
-// syncPropagationWorkStatus will find propagationWork by label in workload, then update resource status to propagationWork status.
-// label example: "karmada.io/created-by: karmada-es-member-cluster-1.default-deployment-nginx"
-// TODO(chenxianpao): sync workload status to propagationWork status.
-func (c *PropagationWorkStatusController) syncPropagationWorkStatus(obj runtime.Object) error {
-	resource := obj.(*unstructured.Unstructured)
-	klog.Infof("sync obj is %s/%s/%s", resource.GetKind(), resource.GetNamespace(), resource.GetName())
-	return nil
-}
-
 // SetupWithManager creates a controller and register to controller manager.
 func (c *PropagationWorkStatusController) SetupWithManager(mgr controllerruntime.Manager) error {
 	return controllerruntime.NewControllerManagedBy(mgr).For(&v1alpha1.PropagationWork{}).Complete(c)

pkg/util/informermanager/handlers.go

@@ -9,24 +9,18 @@ import (
 )
 
 // NewHandlerOnAllEvents builds a ResourceEventHandler that the function 'fn' will be called on all events(add/update/delete).
-func NewHandlerOnAllEvents(fn func(runtime.Object) error) cache.ResourceEventHandler {
+func NewHandlerOnAllEvents(fn func(runtime.Object)) cache.ResourceEventHandler {
 	return &cache.ResourceEventHandlerFuncs{
 		AddFunc: func(cur interface{}) {
 			curObj := cur.(runtime.Object)
 			klog.V(2).Infof("Receive add event, obj is: %+v", curObj)
-			err := fn(curObj)
-			if err != nil {
-				klog.V(2).Infof("Failed to exec fn. Error: %v.", err)
-			}
+			fn(curObj)
 		},
 		UpdateFunc: func(old, cur interface{}) {
 			curObj := cur.(runtime.Object)
 			if !reflect.DeepEqual(old, cur) {
 				klog.V(2).Infof("Receive update event, obj is: %+v", curObj)
-				err := fn(curObj)
-				if err != nil {
-					klog.V(2).Infof("Failed to exec fn. Error: %v.", err)
-				}
+				fn(curObj)
 			}
 		},
 		DeleteFunc: func(old interface{}) {
@@ -39,10 +33,7 @@ func NewHandlerOnAllEvents(fn func(runtime.Object) error) cache.ResourceEventHan
 			}
 			oldObj := old.(runtime.Object)
 			klog.V(2).Infof("Receive delete event, obj is: %+v", oldObj)
-			err := fn(oldObj)
-			if err != nil {
-				klog.V(2).Infof("Failed to exec fn. Error: %v.", err)
-			}
+			fn(oldObj)
 		},
 	}
 }

pkg/util/names/names.go

@@ -25,6 +25,16 @@ func GetMemberClusterName(executionSpaceName string) (string, error) {
 	return strings.TrimPrefix(executionSpaceName, executionSpacePrefix), nil
 }
 
+// GetNamespaceAndName will get namespace and name from ownerLabel.
+// For example: "karmada-es-member-1.default-deployment-nginx"
+func GetNamespaceAndName(value string) (string, string, error) {
+	splits := strings.Split(value, ".")
+	if len(splits) != 2 {
+		return "", "", fmt.Errorf("value is not correct")
+	}
+	return splits[0], splits[1], nil
+}
+
 // GenerateBindingName will generate binding name by namespace, kind and name
 func GenerateBindingName(namespace, kind, name string) string {
 	return strings.ToLower(namespace + "-" + kind + "-" + name)

pkg/util/worker.go

@@ -0,0 +1,202 @@
+package util
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+
+	"github.com/karmada-io/karmada/pkg/util/names"
+)
+
+const (
+	// maxRetries is the number of times a resource will be retried before it is dropped out of the queue.
+	// With the current rate-limiter in use (5ms*2^(maxRetries-1)) the following numbers represent the times
+	// a resource is going to be requeued:
+	//
+	// 5ms, 10ms, 20ms, 40ms, 80ms, 160ms, 320ms, 640ms, 1.3s, 2.6s, 5.1s, 10.2s, 20.4s, 41s, 82s
+	maxRetries = 15
+)
+
+// AsyncWorker is a worker to process resources periodic with a rateLimitingQueue.
+type AsyncWorker interface {
+	EnqueueRateLimited(obj runtime.Object)
+	Run(workerNumber int, stopChan <-chan struct{})
+}
+
+// ReconcileHandler is a callback function for process resources.
+type ReconcileHandler func(key string) error
+
+type asyncWorker struct {
+	// reconcile is callback function to process object in the queue.
+	reconcile ReconcileHandler
+	// queue allowing parallel processing of resources.
+	queue workqueue.RateLimitingInterface
+	// interval is the interval for process object in the queue.
+	interval time.Duration
+}
+
+// NewAsyncWorker returns a asyncWorker which can process resource periodic.
+func NewAsyncWorker(reconcile ReconcileHandler, name string, interval time.Duration) AsyncWorker {
+	return &asyncWorker{
+		reconcile: reconcile,
+		queue:     workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), name),
+		interval:  interval,
+	}
+}
+
+// ClusterWorkload is the thumbnail of cluster workload, it contains GVK, cluster, namespace and name.
+type ClusterWorkload struct {
+	GVK       schema.GroupVersionKind
+	Cluster   string
+	Namespace string
+	Name      string
+}
+
+// GetListerKey returns the key that can be used to query full object information by GenericLister
+func (w *ClusterWorkload) GetListerKey() string {
+	if w.Namespace == "" {
+		return w.Name
+	}
+	return w.Namespace + "/" + w.Name
+}
+
+// GenerateKey generates a key from obj, the key contains cluster, GVK, namespace and name.
+func GenerateKey(obj runtime.Object) (string, error) {
+	resource := obj.(*unstructured.Unstructured)
+	gvk := schema.FromAPIVersionAndKind(resource.GetAPIVersion(), resource.GetKind())
+	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+	if err != nil {
+		klog.Errorf("Couldn't get key for object %#v: %v.", obj, err)
+		return "", err
+	}
+	cluster, err := getClusterNameFromLabel(resource)
+	if err != nil {
+		return "", err
+	}
+	if cluster == "" {
+		return "", nil
+	}
+	return cluster + "/" + gvk.Group + "/" + gvk.Version + "/" + gvk.Kind + "/" + key, nil
+}
+
+// getClusterNameFromLabel gets cluster name from ownerLabel, if label not exist, means resource is not created by karmada.
+func getClusterNameFromLabel(resource *unstructured.Unstructured) (string, error) {
+	workloadLabels := resource.GetLabels()
+	if workloadLabels == nil {
+		klog.V(2).Infof("Resource %s/%s/%s is not created by karmada.", resource.GetKind(),
+			resource.GetNamespace(), resource.GetName())
+		return "", nil
+	}
+	value, exist := workloadLabels[OwnerLabel]
+	if !exist {
+		klog.V(2).Infof("Resource %s/%s/%s is not created by karmada.", resource.GetKind(),
+			resource.GetNamespace(), resource.GetName())
+		return "", nil
+	}
+	executionNamespace, _, err := names.GetNamespaceAndName(value)
+	if err != nil {
+		klog.Errorf("Failed to get executionNamespace from label %s", value)
+		return "", err
+	}
+	cluster, err := names.GetMemberClusterName(executionNamespace)
+	if err != nil {
+		klog.Errorf("Failed to get member cluster name by %s. Error: %v.", value, err)
+		return "", err
+	}
+	return cluster, nil
+}
+
+// SplitMetaKey transforms key to struct ClusterWorkload, struct ClusterWorkload contains cluster, GVK, namespace and name.
+func SplitMetaKey(key string) (ClusterWorkload, error) {
+	var clusterWorkload ClusterWorkload
+	parts := strings.Split(key, "/")
+	switch len(parts) {
+	case 5:
+		// name only, no namespace
+		clusterWorkload.Name = parts[4]
+	case 6:
+		// namespace and name
+		clusterWorkload.Namespace = parts[4]
+		clusterWorkload.Name = parts[5]
+	default:
+		return clusterWorkload, fmt.Errorf("unexpected key format: %q", key)
+	}
+	clusterWorkload.Cluster = parts[0]
+	clusterWorkload.GVK.Group = parts[1]
+	clusterWorkload.GVK.Version = parts[2]
+	clusterWorkload.GVK.Kind = parts[3]
+	return clusterWorkload, nil
+}
+
+func (w *asyncWorker) processKey(obj runtime.Object) string {
+	key, err := GenerateKey(obj)
+	if err != nil {
+		klog.Errorf("Couldn't get key for object %#v: %v.", obj, err)
+		return ""
+	}
+	if key == "" {
+		klog.V(2).Infof("The key is empty, object is not created by karmada.")
+		return ""
+	}
+	return key
+}
+
+func (w *asyncWorker) EnqueueRateLimited(obj runtime.Object) {
+	key := w.processKey(obj)
+	if key == "" {
+		return
+	}
+	w.queue.AddRateLimited(key)
+}
+
+func (w *asyncWorker) handleError(err error, key interface{}) {
+	if err == nil || errors.HasStatusCause(err, v1.NamespaceTerminatingCause) {
+		w.queue.Forget(key)
+		return
+	}
+
+	_, keyErr := SplitMetaKey(key.(string))
+	if keyErr != nil {
+		klog.ErrorS(err, "Failed to split meta namespace cache key", "key", key)
+	}
+
+	if w.queue.NumRequeues(key) < maxRetries {
+		w.queue.AddRateLimited(key)
+		return
+	}
+
+	klog.V(2).Infof("Dropping resource %q out of the queue: %v", key, err)
+	w.queue.Forget(key)
+}
+
+func (w *asyncWorker) worker() {
+	key, quit := w.queue.Get()
+	if quit {
+		return
+	}
+	defer w.queue.Done(key)
+
+	err := w.reconcile(key.(string))
+	w.handleError(err, key)
+}
+
+func (w *asyncWorker) Run(workerNumber int, stopChan <-chan struct{}) {
+	for i := 0; i < workerNumber; i++ {
+		go wait.Until(w.worker, w.interval, stopChan)
+	}
+	// Ensure all goroutines are cleaned up when the stop channel closes
+	go func() {
+		<-stopChan
+		w.queue.ShutDown()
+	}()
+}