Refactor watcher initialization to use goroutines and wait for completion

Author: chenzhiguo

pkg/mutation/mutation_deploy.go

@@ -192,15 +192,18 @@ func deleteConfigMap(name, namespace string) error {
 	return nil
 }
 
-func createDeployPatch(target *appsv1.Deployment, original *appsv1.Deployment) ([]byte, error) {
-	targetPod, err := json.Marshal(target)
-	originalPod, err := json.Marshal(original)
+func createDeployPatch(target, original *appsv1.Deployment) ([]byte, error) {
+	targetJSON, err := json.Marshal(target)
 	if err != nil {
 		return nil, err
 	}
-	p, err := jsonpatch.CreatePatch(originalPod, targetPod)
+	originalJSON, err := json.Marshal(original)
 	if err != nil {
 		return nil, err
 	}
-	return json.Marshal(p)
+	patch, err := jsonpatch.CreatePatch(originalJSON, targetJSON)
+	if err != nil {
+		return nil, err
+	}
+	return json.Marshal(patch)
 }

pkg/mutation/mutation_pod.go

@@ -367,15 +367,18 @@ func addPodVolume(targetPod *corev1.Pod, deploymentName string) []corev1.Volume
 	return append(volumes, agentVolumes...)
 }
 
-func createPodPatch(target *corev1.Pod, original *corev1.Pod) ([]byte, error) {
-	targetPod, err := json.Marshal(target)
-	originalPod, err := json.Marshal(original)
+func createPodPatch(target, original *corev1.Pod) ([]byte, error) {
+	targetJSON, err := json.Marshal(target)
 	if err != nil {
 		return nil, err
 	}
-	p, err := jsonpatch.CreatePatch(originalPod, targetPod)
+	originalJSON, err := json.Marshal(original)
 	if err != nil {
 		return nil, err
 	}
-	return json.Marshal(p)
+	patch, err := jsonpatch.CreatePatch(originalJSON, targetJSON)
+	if err != nil {
+		return nil, err
+	}
+	return json.Marshal(patch)
 }

pkg/watcher/init.go

@@ -1,32 +1,59 @@
 package watcher
 
 import (
+	"sync"
+
 	"github.com/jd-opensource/joylive-injector/pkg/config"
 	"github.com/jd-opensource/joylive-injector/pkg/log"
 	"github.com/jd-opensource/joylive-injector/pkg/resource"
 	"go.uber.org/zap"
 )
 
 func init() {
-	// Start the ConfigMap listener and initialize the content
-	cmWatcher := NewConfigMapWatcher(resource.GetResource().ClientSet)
-	err := cmWatcher.Start()
-	if err != nil {
-		log.Fatal("start cmWatcher error", zap.Error(err))
-	}
-	err = cmWatcher.InitConfigMap(config.GetNamespace())
-	if err != nil {
-		log.Fatal("init cm error", zap.Error(err))
-	}
+	var wg sync.WaitGroup
+	var fatalErr error
+	var once sync.Once
 
-	// Start the AgentVersion listener and initialize the content
-	avWatcher := NewAgentVersionWatcher(resource.GetResource().RestConfig)
-	err = avWatcher.Start()
-	if err != nil {
-		log.Fatal("start avWatcher error", zap.Error(err))
-	}
-	err = avWatcher.InitAgentVersion(config.GetNamespace())
-	if err != nil {
-		log.Fatal("init agentVersion error", zap.Error(err))
+	wg.Add(2)
+
+	go func() {
+		defer wg.Done()
+		cmWatcher := NewConfigMapWatcher(resource.GetResource().ClientSet)
+		err := cmWatcher.Start()
+		if err != nil {
+			once.Do(func() {
+				fatalErr = err
+			})
+			return
+		}
+		err = cmWatcher.InitConfigMap(config.GetNamespace())
+		if err != nil {
+			once.Do(func() {
+				fatalErr = err
+			})
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		avWatcher := NewAgentVersionWatcher(resource.GetResource().RestConfig)
+		err := avWatcher.Start()
+		if err != nil {
+			once.Do(func() {
+				fatalErr = err
+			})
+			return
+		}
+		err = avWatcher.InitAgentVersion(config.GetNamespace())
+		if err != nil {
+			once.Do(func() {
+				fatalErr = err
+			})
+		}
+	}()
+
+	wg.Wait()
+	if fatalErr != nil {
+		log.Fatal("watcher init error", zap.Error(fatalErr))
 	}
 }