Add leadership election

internal/pkg/cmd/reloader.go

@@ -1,11 +1,14 @@
 package cmd
 
 import (
+	"context"
 	"errors"
 	"fmt"
-	"github.com/stakater/Reloader/internal/pkg/constants"
 	"os"
 	"strings"
+	"time"
+
+	"github.com/stakater/Reloader/internal/pkg/constants"
 
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -15,6 +18,15 @@ 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"
+)
+
+const (
+	lockName        string = "stakaer-reloader-lock"
+	podNameEnv      string = "POD_NAME"
+	podNamespaceEnv string = "POD_NAMESPACE"
 )
 
 // NewReloaderCommand starts the reloader controller
@@ -38,21 +50,34 @@ func NewReloaderCommand() *cobra.Command {
 	cmd.PersistentFlags().StringVar(&options.IsArgoRollouts, "is-Argo-Rollouts", "false", "Add support for argo rollouts")
 	cmd.PersistentFlags().StringVar(&options.ReloadStrategy, constants.ReloadStrategyFlag, constants.EnvVarsReloadStrategy, "Specifies the desired reload strategy")
 	cmd.PersistentFlags().StringVar(&options.ReloadOnCreate, "reload-on-create", "false", "Add support to watch create events")
+	cmd.PersistentFlags().BoolVar(&options.EnableHA, "enable-ha", false, "Adds support for running multiple replicas via leadership election")
 
 	return cmd
 }
 
 func validateFlags(*cobra.Command, []string) error {
 	// Ensure the reload strategy is one of the following...
+	var validReloadStrategy bool
 	valid := []string{constants.EnvVarsReloadStrategy, constants.AnnotationsReloadStrategy}
 	for _, s := range valid {
 		if s == options.ReloadStrategy {
-			return nil
+			validReloadStrategy = true
 		}
 	}
 
-	err := fmt.Sprintf("%s must be one of: %s", constants.ReloadStrategyFlag, strings.Join(valid, ", "))
-	return errors.New(err)
+	if !validReloadStrategy {
+		err := fmt.Sprintf("%s must be one of: %s", constants.ReloadStrategyFlag, strings.Join(valid, ", "))
+		return errors.New(err)
+	}
+
+	// Validate that HA options are correct
+	if options.EnableHA {
+		if _, _, err := validateHAEnvs(); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 func configureLogging(logFormat string) error {
@@ -68,6 +93,25 @@ func configureLogging(logFormat string) error {
 	return nil
 }
 
+func validateHAEnvs() (string, string, error) {
+	podName, podNamespace := getHAEnvs()
+
+	if podName == "" {
+		return podName, podNamespace, fmt.Errorf("%s not set, cannot run in HA mode without %s set", podNameEnv, podNameEnv)
+	}
+	if podNamespace == "" {
+		return podName, podNamespace, fmt.Errorf("%s not set, cannot run in HA mode without %s set", podNamespaceEnv, podNamespaceEnv)
+	}
+	return podName, podNamespace, nil
+}
+
+func getHAEnvs() (string, string) {
+	podName := os.Getenv(podNameEnv)
+	podNamespace := os.Getenv(podNamespaceEnv)
+
+	return podName, podNamespace
+}
+
 func startReloader(cmd *cobra.Command, args []string) {
 	err := configureLogging(options.LogFormat)
 	if err != nil {
@@ -99,6 +143,7 @@ func startReloader(cmd *cobra.Command, args []string) {
 
 	collectors := metrics.SetupPrometheusEndpoint()
 
+	var controllers []*controller.Controller
 	for k := range kube.ResourceMap {
 		if ignoredResourcesList.Contains(k) {
 			continue
@@ -109,6 +154,13 @@ func startReloader(cmd *cobra.Command, args []string) {
 			logrus.Fatalf("%s", err)
 		}
 
+		// If HA is enabled then we need to run leadership election
+		if options.EnableHA {
+			c.SetLeader(false)
+		}
+
+		controllers = append(controllers, c)
+
 		// Now let's start the controller
 		stop := make(chan struct{})
 		defer close(stop)
@@ -116,10 +168,64 @@ func startReloader(cmd *cobra.Command, args []string) {
 		go c.Run(1, stop)
 	}
 
-	// Wait forever
+	// Run the leadership election
+	if options.EnableHA {
+		podName, podNamespace := getHAEnvs()
+		lock := getNewLock(clientset, lockName, podName, podNamespace)
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		runLeaderElection(lock, ctx, podName, controllers)
+	}
+
 	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,
+		},
+	}
+}
+
+func runLeaderElection(lock *resourcelock.LeaseLock, ctx context.Context, id string, controllers []*controller.Controller) {
+	leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
+		Lock:            lock,
+		ReleaseOnCancel: true,
+		// TODO Validate that keys persist in the cache for at least one leadership election cycle
+		LeaseDuration: 15 * time.Second,
+		RenewDeadline: 10 * time.Second,
+		RetryPeriod:   2 * time.Second,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(c context.Context) {
+				setLeader(controllers, true)
+			},
+			OnStoppedLeading: func() {
+				setLeader(controllers, false)
+			},
+			OnNewLeader: func(current_id string) {
+				if current_id == id {
+					//klog.Info("still the leader!")
+					return
+				}
+				//klog.Info("new leader is %s", current_id)
+			},
+		},
+	})
+}
+
+func setLeader(controllers []*controller.Controller, isLeader bool) {
+	for _, c := range controllers {
+		c := c
+		c.SetLeader(isLeader)
+	}
+}
+
 func getIgnoredNamespacesList(cmd *cobra.Command) (util.List, error) {
 	return getStringSliceFromFlags(cmd, "namespaces-to-ignore")
 }

internal/pkg/controller/controller.go

@@ -2,9 +2,10 @@ package controller
 
 import (
 	"fmt"
-	"github.com/stakater/Reloader/internal/pkg/options"
 	"time"
 
+	"github.com/stakater/Reloader/internal/pkg/options"
+
 	"github.com/sirupsen/logrus"
 	"github.com/stakater/Reloader/internal/pkg/handler"
 	"github.com/stakater/Reloader/internal/pkg/metrics"
@@ -29,6 +30,7 @@ type Controller struct {
 	namespace         string
 	ignoredNamespaces util.List
 	collectors        metrics.Collectors
+	isLeader          bool
 }
 
 // controllerInitialized flag determines whether controlled is being initialized
@@ -56,6 +58,7 @@ func NewController(
 	c.informer = informer
 	c.queue = queue
 	c.collectors = collectors
+	c.isLeader = true
 	return &c, nil
 }
 
@@ -140,7 +143,7 @@ func (c *Controller) processNextItem() bool {
 	defer c.queue.Done(resourceHandler)
 
 	// Invoke the method containing the business logic
-	err := resourceHandler.(handler.ResourceHandler).Handle()
+	err := resourceHandler.(handler.ResourceHandler).Handle(c.isLeader)
 	// Handle the error if something went wrong during the execution of the business logic
 	c.handleErr(err, resourceHandler)
 	return true
@@ -171,3 +174,7 @@ func (c *Controller) handleErr(err error, key interface{}) {
 	runtime.HandleError(err)
 	logrus.Infof("Dropping the key %q out of the queue: %v", key, err)
 }
+
+func (c *Controller) SetLeader(isLeader bool) {
+	c.isLeader = isLeader
+}

internal/pkg/handler/create.go

@@ -1,6 +1,8 @@
 package handler
 
 import (
+	"fmt"
+
 	"github.com/sirupsen/logrus"
 	"github.com/stakater/Reloader/internal/pkg/metrics"
 	"github.com/stakater/Reloader/internal/pkg/util"
@@ -14,13 +16,16 @@ type ResourceCreatedHandler struct {
 }
 
 // Handle processes the newly created resource
-func (r ResourceCreatedHandler) Handle() error {
+func (r ResourceCreatedHandler) Handle(isLeader bool) error {
 	if r.Resource == nil {
 		logrus.Errorf("Resource creation handler received nil resource")
 	} else {
 		config, _ := r.GetConfig()
 		// process resource based on its type
-		return doRollingUpgrade(config, r.Collectors)
+		if isLeader {
+			return doRollingUpgrade(config, r.Collectors)
+		}
+		return fmt.Errorf("instance is not leader, will not perform rolling upgrade on %s %s/%s", config.Type, config.ResourceName, config.Namespace)
 	}
 	return nil
 }

internal/pkg/handler/handler.go

@@ -6,6 +6,6 @@ import (
 
 // ResourceHandler handles the creation and update of resources
 type ResourceHandler interface {
-	Handle() error
+	Handle(isLeader bool) error
 	GetConfig() (util.Config, string)
 }

internal/pkg/handler/update.go

@@ -1,6 +1,8 @@
 package handler
 
 import (
+	"fmt"
+
 	"github.com/sirupsen/logrus"
 	"github.com/stakater/Reloader/internal/pkg/metrics"
 	"github.com/stakater/Reloader/internal/pkg/util"
@@ -15,14 +17,17 @@ type ResourceUpdatedHandler struct {
 }
 
 // Handle processes the updated resource
-func (r ResourceUpdatedHandler) Handle() error {
+func (r ResourceUpdatedHandler) Handle(isLeader bool) error {
 	if r.Resource == nil || r.OldResource == nil {
 		logrus.Errorf("Resource update handler received nil resource")
 	} else {
 		config, oldSHAData := r.GetConfig()
 		if config.SHAValue != oldSHAData {
 			// process resource based on its type
-			return doRollingUpgrade(config, r.Collectors)
+			if isLeader {
+				return doRollingUpgrade(config, r.Collectors)
+			}
+			return fmt.Errorf("instance is not leader, will not perform rolling upgrade on %s %s/%s", config.Type, config.ResourceName, config.Namespace)
 		}
 	}
 	return nil

internal/pkg/options/flags.go

@@ -25,4 +25,6 @@ var (
 	ReloadStrategy = constants.EnvVarsReloadStrategy
 	// ReloadOnCreate Adds support to watch create events
 	ReloadOnCreate = "false"
+	// EnableHA adds support for running multiple replicas via leadership election
+	EnableHA = false
 )