Move leadership to its own package

internal/pkg/cmd/reloader.go

@@ -6,9 +6,9 @@ import (
 	"fmt"
 	"os"
 	"strings"
-	"time"
 
 	"github.com/stakater/Reloader/internal/pkg/constants"
+	"github.com/stakater/Reloader/internal/pkg/leadership"
 
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -18,9 +18,6 @@ import (
 	"github.com/stakater/Reloader/internal/pkg/util"
 	"github.com/stakater/Reloader/pkg/kube"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/tools/leaderelection"
-	"k8s.io/client-go/tools/leaderelection/resourcelock"
 )
 
 // NewReloaderCommand starts the reloader controller
@@ -161,79 +158,19 @@ func startReloader(cmd *cobra.Command, args []string) {
 		go c.Run(1, stop)
 	}
 
-	// Run the leadership election
+	// Run leadership election
 	if options.EnableHA {
-		var stopChannels []chan struct{}
-		for i := 0; i < len(controllers); i++ {
-			stop := make(chan struct{})
-			stopChannels = append(stopChannels, stop)
-		}
 		podName, podNamespace := getHAEnvs()
-		lock := getNewLock(clientset, constants.LockName, podName, podNamespace)
+		lock := leadership.GetNewLock(clientset, constants.LockName, podName, podNamespace)
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
-		runLeaderElection(lock, ctx, cancel, podName, controllers, stopChannels)
+		leadership.RunLeaderElection(lock, ctx, cancel, podName, controllers)
 		return
 	}
 
 	select {}
 }
 
-func getNewLock(clientset *kubernetes.Clientset, lockName, podname, namespace string) *resourcelock.LeaseLock {
-	return &resourcelock.LeaseLock{
-		LeaseMeta: v1.ObjectMeta{
-			Name:      lockName,
-			Namespace: namespace,
-		},
-		Client: clientset.CoordinationV1(),
-		LockConfig: resourcelock.ResourceLockConfig{
-			Identity: podname,
-		},
-	}
-}
-
-// runLeaderElection runs leadership election. If an instance of the controller is the leader and stops leading it will shutdown.
-func runLeaderElection(lock *resourcelock.LeaseLock, ctx context.Context, cancel context.CancelFunc, id string, controllers []*controller.Controller, stopChannels []chan struct{}) {
-	leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
-		Lock:            lock,
-		ReleaseOnCancel: true,
-		LeaseDuration:   15 * time.Second,
-		RenewDeadline:   10 * time.Second,
-		RetryPeriod:     2 * time.Second,
-		Callbacks: leaderelection.LeaderCallbacks{
-			OnStartedLeading: func(c context.Context) {
-				logrus.Info("became leader, starting controllers")
-				runControllers(controllers, stopChannels)
-			},
-			OnStoppedLeading: func() {
-				logrus.Info("no longer leader, shutting down")
-				stopControllers(stopChannels)
-				cancel()
-			},
-			OnNewLeader: func(current_id string) {
-				if current_id == id {
-					logrus.Info("still the leader!")
-					return
-				}
-				logrus.Infof("new leader is %s", current_id)
-			},
-		},
-	})
-}
-
-func runControllers(controllers []*controller.Controller, stopChannels []chan struct{}) {
-	for i, c := range controllers {
-		c := c
-		go c.Run(1, stopChannels[i])
-	}
-}
-
-func stopControllers(stopChannels []chan struct{}) {
-	for _, c := range stopChannels {
-		close(c)
-	}
-}
-
 func getIgnoredNamespacesList(cmd *cobra.Command) (util.List, error) {
 	return getStringSliceFromFlags(cmd, "namespaces-to-ignore")
 }

internal/pkg/leadership/leadership.go

@@ -0,0 +1,75 @@
+package leadership
+
+import (
+	"context"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stakater/Reloader/internal/pkg/controller"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+)
+
+func GetNewLock(clientset *kubernetes.Clientset, lockName, podname, namespace string) *resourcelock.LeaseLock {
+	return &resourcelock.LeaseLock{
+		LeaseMeta: v1.ObjectMeta{
+			Name:      lockName,
+			Namespace: namespace,
+		},
+		Client: clientset.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: podname,
+		},
+	}
+}
+
+// runLeaderElection runs leadership election. If an instance of the controller is the leader and stops leading it will shutdown.
+func RunLeaderElection(lock *resourcelock.LeaseLock, ctx context.Context, cancel context.CancelFunc, id string, controllers []*controller.Controller) {
+	// Construct channels for the controllers to use
+	var stopChannels []chan struct{}
+	for i := 0; i < len(controllers); i++ {
+		stop := make(chan struct{})
+		stopChannels = append(stopChannels, stop)
+	}
+
+	leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
+		Lock:            lock,
+		ReleaseOnCancel: true,
+		LeaseDuration:   15 * time.Second,
+		RenewDeadline:   10 * time.Second,
+		RetryPeriod:     2 * time.Second,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(c context.Context) {
+				logrus.Info("became leader, starting controllers")
+				runControllers(controllers, stopChannels)
+			},
+			OnStoppedLeading: func() {
+				logrus.Info("no longer leader, shutting down")
+				stopControllers(stopChannels)
+				cancel()
+			},
+			OnNewLeader: func(current_id string) {
+				if current_id == id {
+					logrus.Info("still the leader!")
+					return
+				}
+				logrus.Infof("new leader is %s", current_id)
+			},
+		},
+	})
+}
+
+func runControllers(controllers []*controller.Controller, stopChannels []chan struct{}) {
+	for i, c := range controllers {
+		c := c
+		go c.Run(1, stopChannels[i])
+	}
+}
+
+func stopControllers(stopChannels []chan struct{}) {
+	for _, c := range stopChannels {
+		close(c)
+	}
+}