Add leader election to reconciler

Guard the reconciler execution behind a leader election lock to ensure
that it runs on one pod of the daemonset. This prevents high memory
usage on the whereabouts pod and apiserver when running on large
clusters.

Fixes #386.

Signed-off-by: Alejandro Macedo <alex.macedopereira@gmail.com>

Author: AMacedoP

cmd/controlloop/controlloop.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"os"
@@ -10,10 +11,13 @@ import (
 	"github.com/fsnotify/fsnotify"
 	"github.com/go-co-op/gocron/v2"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/client-go/tools/record"
 
 	nadclient "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/client/clientset/versioned"
@@ -30,18 +34,21 @@ const (
 	allNamespaces               = ""
 	controllerName              = "pod-ip-controlloop"
 	reconcilerCronConfiguration = "/cron-schedule/config"
+	reconcilerLeaderLeaseName   = "whereabouts-reconciler-lock"
+	defaultWhereaboutsNamespace = "kube-system"
 )
 
 const (
 	_ int = iota
 	couldNotCreateController
-	cronSchedulerCreationError
-	fileWatcherError
-	couldNotCreateConfigWatcherError
+	couldNotInitializeReconcilerLeaderElection
 )
 
 const (
-	defaultLogLevel = "debug"
+	defaultLogLevel               = "debug"
+	reconcilerLeaderLeaseDuration = 15 * time.Second
+	reconcilerLeaderRenewDeadline = 10 * time.Second
+	reconcilerLeaderRetryPeriod   = 2 * time.Second
 )
 
 func main() {
@@ -66,53 +73,186 @@ func main() {
 	networkController.Start(stopChan)
 	defer networkController.Shutdown()
 
-	s, err := gocron.NewScheduler(gocron.WithLocation(time.UTC))
+	leaderElectionCtx, cancelLeaderElection := context.WithCancel(context.Background())
+	defer cancelLeaderElection()
+	go runReconcilerLeaderElectionLoop(leaderElectionCtx, errorChan)
+
+	for {
+		select {
+		case <-stopChan:
+			logging.Verbosef("shutting down network controller")
+			cancelLeaderElection()
+			return
+		case err := <-errorChan:
+			if err == nil {
+				logging.Verbosef("reconciler success")
+			} else {
+				logging.Verbosef("reconciler failure: %s", err)
+			}
+		}
+	}
+}
+
+func runReconcilerLeaderElectionLoop(ctx context.Context, errorChan chan error) {
+	namespace := whereaboutsNamespace()
+	identity := reconcilerLeaderIdentity()
+
+	cfg, err := rest.InClusterConfig()
 	if err != nil {
-		os.Exit(cronSchedulerCreationError)
+		_ = logging.Errorf("failed to generate in-cluster config for reconciler leader election: %v", err)
+		os.Exit(couldNotInitializeReconcilerLeaderElection)
+	}
+
+	k8sClientSet, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		_ = logging.Errorf("failed to create kubernetes client for reconciler leader election: %v", err)
+		os.Exit(couldNotInitializeReconcilerLeaderElection)
+	}
+
+	if ctx.Err() != nil {
+		return
+	}
+
+	electionCtx, cancelElection := context.WithCancel(ctx)
+	err = runReconcilerLeaderElection(
+		electionCtx,
+		k8sClientSet,
+		namespace,
+		identity,
+		errorChan,
+		cancelElection,
+	)
+	cancelElection()
+	if err != nil {
+		errorChan <- err
+	}
+}
+
+func runReconcilerLeaderElection(
+	ctx context.Context,
+	k8sClientSet kubernetes.Interface,
+	namespace string,
+	identity string,
+	errorChan chan error,
+	cancelElection context.CancelFunc,
+) error {
+	leaseLock := &resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{
+			Name:      reconcilerLeaderLeaseName,
+			Namespace: namespace,
+		},
+		Client: k8sClientSet.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: identity,
+		},
+	}
+
+	leaderElector, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+		Lock:            leaseLock,
+		LeaseDuration:   reconcilerLeaderLeaseDuration,
+		RenewDeadline:   reconcilerLeaderRenewDeadline,
+		RetryPeriod:     reconcilerLeaderRetryPeriod,
+		ReleaseOnCancel: true,
+		Name:            reconcilerLeaderLeaseName,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(leadingCtx context.Context) {
+				logging.Verbosef("acquired reconciler leadership (%s/%s) as %q", namespace, reconcilerLeaderLeaseName, identity)
+				if err := runScheduledReconciler(leadingCtx, errorChan); err != nil {
+					errorChan <- err
+					cancelElection()
+				}
+			},
+			OnStoppedLeading: func() {
+				logging.Verbosef("lost reconciler leadership (%s/%s)", namespace, reconcilerLeaderLeaseName)
+			},
+			OnNewLeader: func(currentLeader string) {
+				if currentLeader == identity {
+					logging.Verbosef("this pod is reconciler leader: %q", currentLeader)
+					return
+				}
+				logging.Verbosef("reconciler leader is now: %q", currentLeader)
+			},
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create reconciler leader elector: %w", err)
+	}
+
+	logging.Verbosef("starting reconciler leader election (%s/%s) with identity %q", namespace, reconcilerLeaderLeaseName, identity)
+	leaderElector.Run(ctx)
+	return nil
+}
+
+func runScheduledReconciler(ctx context.Context, errorChan chan error) error {
+	scheduler, err := gocron.NewScheduler(gocron.WithLocation(time.UTC))
+	if err != nil {
+		return fmt.Errorf("failed to create reconciler cron scheduler: %w", err)
 	}
 
 	watcher, err := fsnotify.NewWatcher()
 	if err != nil {
-		_ = logging.Errorf("error creating configuration watcher: %v", err)
-		os.Exit(fileWatcherError)
+		if shutdownErr := scheduler.Shutdown(); shutdownErr != nil {
+			_ = logging.Errorf("failed to shutdown reconciler scheduler after watcher initialization error: %v", shutdownErr)
+		}
+		return fmt.Errorf("error creating reconciler configuration watcher: %w", err)
 	}
-	defer watcher.Close()
+	defer func() {
+		if closeErr := watcher.Close(); closeErr != nil {
+			_ = logging.Errorf("error closing reconciler configuration watcher: %v", closeErr)
+		}
+	}()
 
 	reconcilerConfigWatcher, err := reconciler.NewConfigWatcher(
 		reconcilerCronConfiguration,
-		s,
+		scheduler,
 		watcher,
 		func() {
 			reconciler.ReconcileIPs(errorChan)
 		},
 	)
 	if err != nil {
-		os.Exit(couldNotCreateConfigWatcherError)
+		if shutdownErr := scheduler.Shutdown(); shutdownErr != nil {
+			_ = logging.Errorf("failed to shutdown reconciler scheduler after config watcher error: %v", shutdownErr)
+		}
+		return fmt.Errorf("could not create reconciler config watcher: %w", err)
 	}
-	s.Start()
 
+	scheduler.Start()
 	const reconcilerConfigMntFile = "/cron-schedule/..data"
-	p := func(e fsnotify.Event) bool {
-		return e.Name == reconcilerConfigMntFile && e.Op&fsnotify.Create == fsnotify.Create
+	reconcilerConfigWatcher.SyncConfiguration(func(event fsnotify.Event) bool {
+		return event.Name == reconcilerConfigMntFile && event.Op&fsnotify.Create == fsnotify.Create
+	})
+
+	logging.Verbosef("scheduled reconciler started")
+	<-ctx.Done()
+	logging.Verbosef("scheduled reconciler stopping")
+
+	if err := scheduler.Shutdown(); err != nil {
+		_ = logging.Errorf("error shutting reconciler scheduler: %v", err)
 	}
-	reconcilerConfigWatcher.SyncConfiguration(p)
 
-	for {
-		select {
-		case <-stopChan:
-			logging.Verbosef("shutting down network controller")
-			if err := s.Shutdown(); err != nil {
-				_ = logging.Errorf("error shutting : %v", err)
-			}
-			return
-		case err := <-errorChan:
-			if err == nil {
-				logging.Verbosef("reconciler success")
-			} else {
-				logging.Verbosef("reconciler failure: %s", err)
-			}
-		}
+	return nil
+}
+
+func whereaboutsNamespace() string {
+	if namespace, found := os.LookupEnv("WHEREABOUTS_NAMESPACE"); found && namespace != "" {
+		return namespace
+	}
+	return defaultWhereaboutsNamespace
+}
+
+func reconcilerLeaderIdentity() string {
+	if podName, found := os.LookupEnv("POD_NAME"); found && podName != "" {
+		return podName
+	}
+	hostname, err := os.Hostname()
+	if err == nil && hostname != "" {
+		return hostname
+	}
+	if nodeName, found := os.LookupEnv("NODENAME"); found && nodeName != "" {
+		return fmt.Sprintf("%s-%d", nodeName, os.Getpid())
 	}
+	return fmt.Sprintf("%s-%d", reconcilerLeaderLeaseName, os.Getpid())
 }
 
 func handleSignals(stopChannel chan struct{}, signals ...os.Signal) {

deployment/whereabouts-chart/templates/daemonset.yaml

@@ -48,6 +48,10 @@ spec:
               /token-watcher.sh &
               /ip-control-loop -log-level debug
           env:
+          - name: POD_NAME
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.name
           - name: NODENAME
             valueFrom:
               fieldRef:

doc/crds/daemonset-install.yaml

@@ -125,6 +125,10 @@ spec:
             /ip-control-loop -log-level debug
         image: ghcr.io/k8snetworkplumbingwg/whereabouts:latest
         env:
+        - name: POD_NAME
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.name
         - name: NODENAME
           valueFrom:
             fieldRef: