added leader election per controller

Author: GrigoriyMikhalkin

pkg/controller/controller.go

@@ -18,23 +18,65 @@ package controller
 
 import (
 	"fmt"
+	"time"
 
+	"github.com/go-logr/logr"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/internal/controller"
+	logf "sigs.k8s.io/controller-runtime/pkg/internal/log"
+	internalrecorder "sigs.k8s.io/controller-runtime/pkg/internal/recorder"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/recorder"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
+const (
+	// Values taken from: https://github.com/kubernetes/apiserver/blob/master/pkg/apis/config/v1alpha1/defaults.go
+	defaultLeaseDuration = 15 * time.Second
+	defaultRenewDeadline = 10 * time.Second
+	defaultRetryPeriod   = 2 * time.Second
+)
+
 // Options are the arguments for creating a new Controller
 type Options struct {
 	// MaxConcurrentReconciles is the maximum number of concurrent Reconciles which can be run. Defaults to 1.
 	MaxConcurrentReconciles int
 
 	// Reconciler reconciles an object
 	Reconciler reconcile.Reconciler
+
+	// LeaderElection determines whether or not to use leader election when
+	// starting the controller.
+	LeaderElection bool
+
+	// LeaderElectionNamespace determines the namespace in which the leader
+	// election configmap will be created.
+	LeaderElectionNamespace string
+
+	// LeaderElectionID determines the name of the configmap that leader election
+	// will use for holding the leader lock.
+	LeaderElectionID string
+
+	// LeaseDuration is the duration that non-leader candidates will
+	// wait to force acquire leadership. This is measured against time of
+	// last observed ack. Default is 15 seconds.
+	LeaseDuration *time.Duration
+	// RenewDeadline is the duration that the acting master will retry
+	// refreshing leadership before giving up. Default is 10 seconds.
+	RenewDeadline *time.Duration
+	// RetryPeriod is the duration the LeaderElector clients should wait
+	// between tries of actions. Default is 2 seconds.
+	RetryPeriod *time.Duration
+
+	// Dependency injection for testing
+	newResourceLock func(config *rest.Config, recorderProvider recorder.Provider, options leaderelection.Options) (resourcelock.Interface, error)
 }
 
 // Controller implements a Kubernetes API.  A Controller manages a work queue fed reconcile.Requests
@@ -69,15 +111,22 @@ func New(name string, mgr manager.Manager, options Options) (Controller, error)
 		return nil, fmt.Errorf("must specify Name for Controller")
 	}
 
-	if options.MaxConcurrentReconciles <= 0 {
-		options.MaxConcurrentReconciles = 1
-	}
+	log := logf.RuntimeLog.WithValues("controller", name)
+
+	// Set default values for options fields
+	options = setOptionsDefaults(options, name)
 
 	// Inject dependencies into Reconciler
 	if err := mgr.SetFields(options.Reconciler); err != nil {
 		return nil, err
 	}
 
+	// Create the resource lock
+	resourceLock, err := createResourceLock(mgr, options, log)
+	if err != nil {
+		return nil, err
+	}
+
 	// Create controller with dependencies set
 	c := &controller.Controller{
 		Do:                      options.Reconciler,
@@ -89,8 +138,59 @@ func New(name string, mgr manager.Manager, options Options) (Controller, error)
 		Queue:                   workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), name),
 		MaxConcurrentReconciles: options.MaxConcurrentReconciles,
 		Name:                    name,
+		ResourceLock:            resourceLock,
+		LeaseDuration:           *options.LeaseDuration,
+		RenewDeadline:           *options.RenewDeadline,
+		RetryPeriod:             *options.RetryPeriod,
 	}
 
 	// Add the controller as a Manager components
 	return c, mgr.Add(c)
 }
+
+// setOptionsDefaults set default values for Options fields
+func setOptionsDefaults(options Options, name string) Options {
+	if options.MaxConcurrentReconciles <= 0 {
+		options.MaxConcurrentReconciles = 1
+	}
+
+	if options.newResourceLock == nil {
+		options.newResourceLock = leaderelection.NewResourceLock
+	}
+
+	if options.LeaderElection && options.LeaderElectionID == "" {
+		options.LeaderElectionID = fmt.Sprint(name, "-controller")
+	}
+
+	leaseDuration, renewDeadline, retryPeriod := defaultLeaseDuration, defaultRenewDeadline, defaultRetryPeriod
+	if options.LeaseDuration == nil {
+		options.LeaseDuration = &leaseDuration
+	}
+
+	if options.RenewDeadline == nil {
+		options.RenewDeadline = &renewDeadline
+	}
+
+	if options.RetryPeriod == nil {
+		options.RetryPeriod = &retryPeriod
+	}
+
+	return options
+}
+
+func createResourceLock(mgr manager.Manager, options Options, log logr.Logger) (resourcelock.Interface, error) {
+	recorderProvider, err := internalrecorder.NewProvider(mgr.GetConfig(), scheme.Scheme, log)
+	if err != nil {
+		return nil, err
+	}
+
+	resourceLock, err := options.newResourceLock(mgr.GetConfig(), recorderProvider, leaderelection.Options{
+		LeaderElection:          options.LeaderElection,
+		LeaderElectionID:        options.LeaderElectionID,
+		LeaderElectionNamespace: options.LeaderElectionNamespace,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return resourceLock, nil
+}

pkg/controller/controller_integration_test.go

@@ -14,15 +14,14 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package controller_test
+package controller
 
 import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
-	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
@@ -55,7 +54,7 @@ var _ = Describe("controller", func() {
 			Expect(err).NotTo(HaveOccurred())
 
 			By("Creating the Controller")
-			instance, err := controller.New("foo-controller", cm, controller.Options{
+			instance, err := New("foo-controller", cm, Options{
 				Reconciler: reconcile.Func(
 					func(request reconcile.Request) (reconcile.Result, error) {
 						reconciled <- request

pkg/controller/controller_suite_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package controller_test
+package controller
 
 import (
 	"testing"

pkg/controller/controller_test.go

@@ -14,17 +14,20 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package controller_test
+package controller
 
 import (
 	"fmt"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/recorder"
 	"sigs.k8s.io/controller-runtime/pkg/runtime/inject"
 )
 
@@ -46,7 +49,7 @@ var _ = Describe("controller.Controller", func() {
 		It("should return an error if Name is not Specified", func(done Done) {
 			m, err := manager.New(cfg, manager.Options{})
 			Expect(err).NotTo(HaveOccurred())
-			c, err := controller.New("", m, controller.Options{Reconciler: rec})
+			c, err := New("", m, Options{Reconciler: rec})
 			Expect(c).To(BeNil())
 			Expect(err.Error()).To(ContainSubstring("must specify Name for Controller"))
 
@@ -57,7 +60,7 @@ var _ = Describe("controller.Controller", func() {
 			m, err := manager.New(cfg, manager.Options{})
 			Expect(err).NotTo(HaveOccurred())
 
-			c, err := controller.New("foo", m, controller.Options{})
+			c, err := New("foo", m, Options{})
 			Expect(c).To(BeNil())
 			Expect(err.Error()).To(ContainSubstring("must specify Reconciler"))
 
@@ -68,7 +71,7 @@ var _ = Describe("controller.Controller", func() {
 			m, err := manager.New(cfg, manager.Options{})
 			Expect(err).NotTo(HaveOccurred())
 
-			c, err := controller.New("foo", m, controller.Options{Reconciler: &failRec{}})
+			c, err := New("foo", m, Options{Reconciler: &failRec{}})
 			Expect(c).To(BeNil())
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("expected error"))
@@ -80,16 +83,49 @@ var _ = Describe("controller.Controller", func() {
 			m, err := manager.New(cfg, manager.Options{})
 			Expect(err).NotTo(HaveOccurred())
 
-			c1, err := controller.New("c1", m, controller.Options{Reconciler: rec})
+			c1, err := New("c1", m, Options{Reconciler: rec})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(c1).ToNot(BeNil())
 
-			c2, err := controller.New("c2", m, controller.Options{Reconciler: rec})
+			c2, err := New("c2", m, Options{Reconciler: rec})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(c2).ToNot(BeNil())
 
 			close(done)
 		})
+
+		It("with leader election enabled should default ID to controller name if ID is not set", func(done Done) {
+			var rl resourcelock.Interface
+			m, err := manager.New(cfg, manager.Options{})
+			Expect(err).NotTo(HaveOccurred())
+			c, err := New("test", m, Options{
+				Reconciler:              rec,
+				LeaderElection:          true,
+				LeaderElectionNamespace: "default",
+				newResourceLock: func(config *rest.Config, recorderProvider recorder.Provider, options leaderelection.Options) (resourcelock.Interface, error) {
+					var err error
+					rl, err = leaderelection.NewResourceLock(config, recorderProvider, options)
+					return rl, err
+				},
+			})
+			Expect(c).ToNot(BeNil())
+			Expect(err).ToNot(HaveOccurred())
+			Expect(rl.Describe()).To(Equal("default/test-controller"))
+			close(done)
+		})
+
+		It("should return an error if namespace not set and not running in cluster", func() {
+			m, err := manager.New(cfg, manager.Options{})
+			Expect(err).NotTo(HaveOccurred())
+			c, err := New("test", m, Options{
+				Reconciler:       rec,
+				LeaderElection:   true,
+				LeaderElectionID: "test-controller",
+			})
+			Expect(c).To(BeNil())
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("unable to find leader election namespace: not running in-cluster, please specify LeaderElectionNamespace"))
+		})
 	})
 })
 

pkg/internal/controller/controller.go

@@ -17,6 +17,7 @@ limitations under the License.
 package controller
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"time"
@@ -25,6 +26,8 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
@@ -92,6 +95,21 @@ type Controller struct {
 	// Kubernetes API.
 	Recorder record.EventRecorder
 
+	errChan chan error
+
+	// resourceLock forms the basis for leader election
+	ResourceLock resourcelock.Interface
+
+	// leaseDuration is the duration that non-leader candidates will
+	// wait to force acquire leadership.
+	LeaseDuration time.Duration
+	// renewDeadline is the duration that the acting master will retry
+	// refreshing leadership before giving up.
+	RenewDeadline time.Duration
+	// retryPeriod is the duration the LeaderElector clients should wait
+	// between tries of actions.
+	RetryPeriod time.Duration
+
 	// TODO(community): Consider initializing a logger with the Controller Name as the tag
 }
 
@@ -130,6 +148,26 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 	defer utilruntime.HandleCrash()
 	defer c.Queue.ShutDown()
 
+	c.errChan = make(chan error)
+	if c.ResourceLock != nil {
+		err := c.startLeaderElection(stop)
+		if err != nil {
+			return err
+		}
+	} else {
+		go c.runController(stop)
+	}
+
+	select {
+	case <-stop:
+		log.Info("Stopping workers", "controller", c.Name)
+		return nil
+	case err := <-c.errChan:
+		return err
+	}
+}
+
+func (c *Controller) runController(stop <-chan struct{}) {
 	// Start the SharedIndexInformer factories to begin populating the SharedIndexInformer caches
 	log.Info("Starting Controller", "controller", c.Name)
 
@@ -143,7 +181,8 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 		err := fmt.Errorf("failed to wait for %s caches to sync", c.Name)
 		log.Error(err, "Could not wait for Cache to sync", "controller", c.Name)
 		c.mu.Unlock()
-		return err
+		c.errChan <- err
+		return
 	}
 
 	if c.JitterPeriod == 0 {
@@ -162,9 +201,29 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 
 	c.Started = true
 	c.mu.Unlock()
+}
+
+func (c *Controller) startLeaderElection(stop <-chan struct{}) (err error) {
+	l, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+		Lock:          c.ResourceLock,
+		LeaseDuration: c.LeaseDuration,
+		RenewDeadline: c.RenewDeadline,
+		RetryPeriod:   c.RetryPeriod,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(_ context.Context) {
+				c.runController(stop)
+			},
+			OnStoppedLeading: func() {
+				c.errChan <- fmt.Errorf("leader election lost")
+			},
+		},
+	})
+	if err != nil {
+		return err
+	}
 
-	<-stop
-	log.Info("Stopping workers", "controller", c.Name)
+	// Start the leader elector process
+	go l.Run(context.Background())
 	return nil
 }