feat: reduce meta reconcile time

pkg/controller/component/reclaimer/meta_reconciler.go

@@ -17,6 +17,7 @@ limitations under the License.
 package reclaimer
 
 import (
+	"context"
 	"fmt"
 
 	corev1 "k8s.io/api/core/v1"
@@ -27,6 +28,11 @@ import (
 	"github.com/vesoft-inc/nebula-operator/apis/pkg/label"
 	"github.com/vesoft-inc/nebula-operator/pkg/controller/component"
 	"github.com/vesoft-inc/nebula-operator/pkg/kube"
+	"github.com/vesoft-inc/nebula-operator/pkg/util/async"
+)
+
+const (
+	Concurrency = 5
 )
 
 type meta struct {
@@ -49,45 +55,52 @@ func (m *meta) Reconcile(nc *v1alpha1.NebulaCluster) error {
 		return fmt.Errorf("list pods for cluster %s/%s failed: %v", namespace, clusterName, err)
 	}
 
-	// TODO: concurrent updating to reduce reconcile time
+	group := async.NewGroup(context.TODO(), Concurrency)
 	for i := range pods {
 		pod := pods[i]
 		if !label.Label(pod.Labels).IsNebulaComponent() {
 			continue
 		}
 
-		var hasDataPV bool
-		if pod.Labels[label.ComponentLabelKey] == label.GraphdLabelVal {
-			hasDataPV = false
-		}
-
-		pvcs, err := m.resolvePVCFromPod(&pod)
-		if err != nil {
-			if errors.IsNotFound(err) && !hasDataPV {
-				continue
-			}
-			return err
-		}
-		for i := range pvcs {
-			pvc := pvcs[i]
-			if err := m.clientSet.PVC().UpdateMetaInfo(pvc, &pod, nc.IsPVReclaimEnabled()); err != nil {
-				return err
-			}
-			if pvc.Spec.VolumeName == "" {
-				continue
+		worker := func() error {
+			var hasDataPV bool
+			if pod.Labels[label.ComponentLabelKey] == label.GraphdLabelVal {
+				hasDataPV = false
 			}
-			pv, err := m.clientSet.PV().GetPersistentVolume(pvc.Spec.VolumeName)
+
+			pvcs, err := m.resolvePVCFromPod(&pod)
 			if err != nil {
-				klog.Errorf("cluster [%s/%s] get PV %s failed: %v", namespace, clusterName, pvc.Spec.VolumeName, err)
+				if errors.IsNotFound(err) && !hasDataPV {
+					return nil
+				}
 				return err
 			}
-			if err := m.clientSet.PV().UpdateMetaInfo(nc, pv); err != nil {
-				return err
+			for i := range pvcs {
+				pvc := pvcs[i]
+				if err := m.clientSet.PVC().UpdateMetaInfo(pvc, &pod, nc.IsPVReclaimEnabled()); err != nil {
+					return err
+				}
+				if pvc.Spec.VolumeName == "" {
+					continue
+				}
+				pv, err := m.clientSet.PV().GetPersistentVolume(pvc.Spec.VolumeName)
+				if err != nil {
+					klog.Errorf("cluster [%s/%s] get PV %s failed: %v", namespace, clusterName, pvc.Spec.VolumeName, err)
+					return err
+				}
+				if err := m.clientSet.PV().UpdateMetaInfo(nc, pv); err != nil {
+					return err
+				}
 			}
+			return nil
 		}
+
+		group.Add(func(stopCh chan interface{}) {
+			stopCh <- worker()
+		})
 	}
 
-	return nil
+	return group.Wait()
 }
 
 func (m *meta) resolvePVCFromPod(pod *corev1.Pod) ([]*corev1.PersistentVolumeClaim, error) {

pkg/controller/nebulacluster/nebula_cluster_control.go

@@ -146,10 +146,12 @@ func (c *defaultNebulaClusterControl) updateNebulaCluster(nc *v1alpha1.NebulaClu
 		return err
 	}
 
+	klog.Infof("start reconcile pv and pvc metadata cluster")
 	if err := c.metaReconciler.Reconcile(nc); err != nil {
 		klog.Errorf("reconcile pv and pvc metadata cluster failed: %v", err)
 		return err
 	}
+	klog.Infof("finished reconcile pv and pvc metadata cluster successfully")
 
 	if err := c.pvcReclaimer.Reclaim(nc); err != nil {
 		klog.Errorf("reclaim pvc failed: %v", err)

pkg/kube/pv.go

@@ -28,6 +28,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2"
+	"reflect"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vesoft-inc/nebula-operator/apis/pkg/annotation"
@@ -128,6 +129,10 @@ func (p *pvClient) UpdatePersistentVolume(pv *corev1.PersistentVolume) error {
 
 	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
 		if updated, err := p.GetPersistentVolume(pvName); err == nil {
+			if reflect.DeepEqual(labels, updated.GetLabels()) && reflect.DeepEqual(annotations, updated.GetAnnotations()) {
+				return nil
+			}
+
 			pv = updated.DeepCopy()
 			pv.SetLabels(labels)
 			pv.SetAnnotations(annotations)

pkg/kube/pvc.go

@@ -19,6 +19,7 @@ package kube
 import (
 	"context"
 	"fmt"
+	"reflect"
 	"strconv"
 
 	corev1 "k8s.io/api/core/v1"
@@ -99,6 +100,12 @@ func (p *pvcClient) UpdatePVC(pvc *corev1.PersistentVolumeClaim) error {
 
 	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
 		if updated, err := p.GetPVC(ns, pvcName); err == nil {
+			if reflect.DeepEqual(pvcLabels, updated.GetLabels()) &&
+				reflect.DeepEqual(annotations, updated.GetAnnotations()) &&
+				reflect.DeepEqual(requests, updated.Spec.Resources.Requests) {
+				return nil
+			}
+
 			pvc = updated.DeepCopy()
 			pvc.SetLabels(pvcLabels)
 			pvc.SetAnnotations(annotations)