Replace watch with builder.ControllerManagedBy

Signed-off-by: Manuel de Brito Fontes <[email protected]>

Author: aledbf

components/node-labeler/cmd/run.go

@@ -18,16 +18,18 @@ import (
 	"github.com/spf13/cobra"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/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/plugin/pkg/client/auth"
 	"k8s.io/client-go/util/retry"
+	"k8s.io/utils/pointer"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
-	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/metrics"
@@ -60,6 +62,10 @@ var runCmd = &cobra.Command{
 			LeaderElection:         true,
 			LeaderElectionID:       "node-labeler.gitpod.io",
 			Namespace:              namespace,
+			// default sync period is 10h.
+			// in case node-labeler is restarted and not change happens, we could waste (at least) 20m in a node
+			// that never will run workspaces and the additional nodes cluster-autoscaler adds to compensate
+			SyncPeriod: pointer.Duration(1 * time.Minute),
 		})
 		if err != nil {
 			log.WithError(err).Fatal("unable to start node-labeber")
@@ -74,35 +80,32 @@ var runCmd = &cobra.Command{
 			client,
 		}
 
-		c, err := controller.New("pod-watcher", mgr, controller.Options{
-			Reconciler:              r,
-			MaxConcurrentReconciles: 20,
+		filterPredicate, err := predicate.LabelSelectorPredicate(metav1.LabelSelector{
+			MatchLabels: map[string]string{
+				"required-by-node-labeler": "true",
+			},
 		})
+		if err != nil {
+			log.WithError(err).Fatal("unable to create predicate")
+		}
+
+		err = ctrl.NewControllerManagedBy(mgr).
+			Named("pod-watcher").
+			For(&corev1.Pod{}).
+			Watches(
+				&source.Kind{Type: &corev1.Pod{}},
+				&handler.EnqueueRequestForObject{},
+				builder.WithPredicates(filterPredicate),
+			).
+			WithOptions(controller.Options{MaxConcurrentReconciles: 20}).
+			Complete(r)
 		if err != nil {
 			log.WithError(err).Fatal("unable to bind controller watch event handler")
 		}
 
 		metrics.Registry.MustRegister(NodeLabelerCounterVec)
 		metrics.Registry.MustRegister(NodeLabelerTimeHistVec)
 
-		err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForObject{}, predicate.Funcs{
-			CreateFunc: func(ce event.CreateEvent) bool {
-				return processPodEvent(ce.Object)
-			},
-			UpdateFunc: func(ue event.UpdateEvent) bool {
-				return processPodEvent(ue.ObjectNew)
-			},
-			DeleteFunc: func(deleteEvent event.DeleteEvent) bool {
-				return processPodEvent(deleteEvent.Object)
-			},
-			GenericFunc: func(genericEvent event.GenericEvent) bool {
-				return false
-			},
-		})
-		if err != nil {
-			log.WithError(err).Fatal("unable to create controller")
-		}
-
 		err = mgr.AddHealthzCheck("healthz", healthz.Ping)
 		if err != nil {
 			log.WithError(err).Fatal("unable to set up health check")
@@ -132,14 +135,6 @@ var (
 	scheme = runtime.NewScheme()
 )
 
-func processPodEvent(pod client.Object) bool {
-	if strings.HasPrefix(pod.GetName(), registryFacade) || strings.HasPrefix(pod.GetName(), wsDaemon) {
-		return true
-	}
-
-	return false
-}
-
 type PodReconciler struct {
 	client.Client
 }
@@ -165,8 +160,6 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
 		port          string
 		component     string
 		labelToUpdate string
-
-		waitTimeout time.Duration = 5 * time.Second
 	)
 
 	switch {
@@ -181,7 +174,7 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
 		ipAddress = pod.Status.PodIP
 		port = strconv.Itoa(wsdaemonPort)
 	default:
-		log.WithField("pod", pod.Name).Info("Invalid pod. Skipping...")
+		// nothing to do
 		return reconcile.Result{}, nil
 	}
 
@@ -215,14 +208,17 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
 		return reconcile.Result{}, fmt.Errorf("obtaining node %s: %w", nodeName, err)
 	}
 
-	if node.Labels[labelToUpdate] == "true" {
-		// Label already exists.
+	if labelValue, exists := node.Labels[labelToUpdate]; exists && labelValue == "true" {
+		// nothing to do, the label already exists.
 		return reconcile.Result{}, nil
 	}
 
-	err = waitForTCPPortToBeReachable(ipAddress, port, 30*time.Second)
+	err = checkTCPPortIsReachable(ipAddress, port)
 	if err != nil {
-		return reconcile.Result{}, fmt.Errorf("waiting for TCP port: %v", err)
+		log.WithField("host", ipAddress).WithField("port", port).
+			WithField("pod", pod.Name).
+			WithError(err).Error("checking if TCP port is open")
+		return reconcile.Result{RequeueAfter: time.Second * 5}, nil
 	}
 
 	if component == registryFacade {
@@ -233,8 +229,6 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
 		}
 	}
 
-	time.Sleep(waitTimeout)
-
 	err = updateLabel(labelToUpdate, true, nodeName, r)
 	if err != nil {
 		return reconcile.Result{}, fmt.Errorf("trying to add the label: %v", err)
@@ -280,31 +274,14 @@ func updateLabel(label string, add bool, nodeName string, client client.Client)
 	})
 }
 
-func waitForTCPPortToBeReachable(host string, port string, timeout time.Duration) error {
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-
-	ticker := time.NewTicker(1 * time.Second)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return fmt.Errorf("port %v on host %v never reachable", port, host)
-		case <-ticker.C:
-			conn, err := net.DialTimeout("tcp", net.JoinHostPort(host, port), 500*time.Millisecond)
-			if err != nil {
-				continue
-			}
-
-			if conn != nil {
-				conn.Close()
-				return nil
-			}
-
-			continue
-		}
+func checkTCPPortIsReachable(host string, port string) error {
+	conn, err := net.DialTimeout("tcp", net.JoinHostPort(host, port), 1*time.Second)
+	if err != nil {
+		return err
 	}
+	defer conn.Close()
+
+	return nil
 }
 
 func checkRegistryFacade(host, port string) error {