diff --git a/internal/pkg/cmd/reloader.go b/internal/pkg/cmd/reloader.go index 963070ca..09f893e7 100644 --- a/internal/pkg/cmd/reloader.go +++ b/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") } diff --git a/internal/pkg/leadership/leadership.go b/internal/pkg/leadership/leadership.go new file mode 100644 index 00000000..e0f3db55 --- /dev/null +++ b/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) + } +}