Refactor waitCRDs function

Move the logic to wait for CRDs to pkg/manager package to keep main.go
slightly cleaner.

Signed-off-by: Michi Mutsuzaki <[email protected]>

Author: michi-covalent

cmd/tetragon/main.go

@@ -66,12 +66,8 @@ import (
 	"github.com/spf13/viper"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/durationpb"
-	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
-	apiextensionsclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
-	apiextensionsinformer "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions/apiextensions/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/cache"
 )
 
 var (
@@ -401,8 +397,22 @@ func tetragonExecuteCtx(ctx context.Context, cancel context.CancelFunc, ready fu
 		// Start controller-runtime manager.
 		controllerManager = manager.Get()
 		controllerManager.Start(ctx)
+		crds := make(map[string]struct{})
+		if option.Config.EnableTracingPolicyCRD {
+			crds[v1alpha1.TPName] = struct{}{}
+			crds[v1alpha1.TPNamespacedName] = struct{}{}
+		}
+		if option.Config.EnablePodInfo {
+			crds[v1alpha1.PIName] = struct{}{}
+		}
+		if len(crds) > 0 {
+			err = controllerManager.WaitCRDs(ctx, crds)
+			if err != nil {
+				return err
+			}
+		}
 		// retrieve k8s clients
-		k8sClient, _, err = watcher.GetK8sClients(waitCRDs)
+		k8sClient, _, err = watcher.GetK8sClients(func(_ *rest.Config) error { return nil })
 		if err != nil {
 			return err
 		}
@@ -539,57 +549,6 @@ func tetragonExecuteCtx(ctx context.Context, cancel context.CancelFunc, ready fu
 	return obs.StartReady(ctx, ready)
 }
 
-func waitCRDs(config *rest.Config) error {
-	crds := make(map[string]struct{})
-
-	if option.Config.EnableTracingPolicyCRD {
-		crds[v1alpha1.TPName] = struct{}{}
-		crds[v1alpha1.TPNamespacedName] = struct{}{}
-	}
-	if option.Config.EnablePodInfo {
-		crds[v1alpha1.PIName] = struct{}{}
-	}
-
-	if len(crds) == 0 {
-		log.Info("No CRDs are enabled")
-		return nil
-	}
-
-	log.WithField("crds", crds).Info("Waiting for required CRDs")
-	var wg sync.WaitGroup
-	wg.Add(1)
-	crdClient := apiextensionsclientset.NewForConfigOrDie(config)
-	crdInformer := apiextensionsinformer.NewCustomResourceDefinitionInformer(crdClient, 0*time.Second, nil)
-	_, err := crdInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj interface{}) {
-			crdObject, ok := obj.(*v1.CustomResourceDefinition)
-			if !ok {
-				log.WithField("obj", obj).Warn("Received an invalid object")
-				return
-			}
-			if _, ok := crds[crdObject.Name]; ok {
-				log.WithField("crd", crdObject.Name).Info("Found CRD")
-				delete(crds, crdObject.Name)
-				if len(crds) == 0 {
-					log.Info("Found all the required CRDs")
-					wg.Done()
-				}
-			}
-		},
-	})
-	if err != nil {
-		log.WithError(err).Error("failed to add event handler")
-		return err
-	}
-	stop := make(chan struct{})
-	go func() {
-		crdInformer.Run(stop)
-	}()
-	wg.Wait()
-	close(stop)
-	return nil
-}
-
 func loadTpFromDir(ctx context.Context, dir string) error {
 	if _, err := os.Stat(dir); err != nil {
 		// Do not fail if the default directory doesn't exist,

pkg/manager/manager.go

@@ -11,10 +11,12 @@ import (
 	"github.com/cilium/tetragon/pkg/k8s/apis/cilium.io/v1alpha1"
 	"github.com/cilium/tetragon/pkg/logger"
 	corev1 "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/cache"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	ctrlManager "sigs.k8s.io/controller-runtime/pkg/manager"
@@ -38,6 +40,7 @@ func Get() *ControllerManager {
 		scheme := runtime.NewScheme()
 		utilruntime.Must(clientgoscheme.AddToScheme(scheme))
 		utilruntime.Must(v1alpha1.AddToScheme(scheme))
+		utilruntime.Must(apiextensionsv1.AddToScheme(scheme))
 		controllerManager, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{Scheme: scheme})
 		if err != nil {
 			panic(err)
@@ -75,3 +78,41 @@ func (cm *ControllerManager) ListNamespaces() ([]corev1.Namespace, error) {
 	}
 	return namespaceList.Items, nil
 }
+
+func (cm *ControllerManager) WaitCRDs(ctx context.Context, crds map[string]struct{}) error {
+	log := logger.GetLogger()
+	log.WithField("crds", crds).Info("Waiting for required CRDs")
+	var wg sync.WaitGroup
+	wg.Add(1)
+	crdInformer, err := cm.Manager.GetCache().GetInformer(ctx, &apiextensionsv1.CustomResourceDefinition{})
+	if err != nil {
+		return err
+	}
+	_, err = crdInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			crdObject, ok := obj.(*apiextensionsv1.CustomResourceDefinition)
+			if !ok {
+				log.WithField("obj", obj).Warn("Received an invalid object")
+				return
+			}
+			if _, ok := crds[crdObject.Name]; ok {
+				log.WithField("crd", crdObject.Name).Info("Found CRD")
+				delete(crds, crdObject.Name)
+				if len(crds) == 0 {
+					log.Info("Found all the required CRDs")
+					wg.Done()
+				}
+			}
+		},
+	})
+	if err != nil {
+		log.WithError(err).Error("failed to add event handler")
+		return err
+	}
+	wg.Wait()
+	err = cm.Manager.GetCache().RemoveInformer(ctx, &apiextensionsv1.CustomResourceDefinition{})
+	if err != nil {
+		log.WithError(err).Warn("failed to remove CRD informer")
+	}
+	return nil
+}