Add leader election to controller manager

Author: JoelSpeed

pkg/leaderelection/doc.go

@@ -0,0 +1,20 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+/*
+Package leaderelection contains a constructors for a leader election resource lock
+*/
+package leaderelection

pkg/leaderelection/leader_election.go

@@ -0,0 +1,78 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package leaderelection
+
+import (
+	"fmt"
+	"os"
+
+	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"sigs.k8s.io/controller-runtime/pkg/recorder"
+)
+
+// Options provides the required configuration to create a new resource lock
+type Options struct {
+	// LeaderElection determines whether or not to use leader election when
+	// starting the manager.
+	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
+}
+
+// NewResourceLock creates a new config map resource lock for use in a leader
+// election loop
+func NewResourceLock(config *rest.Config, recorderProvider recorder.Provider, options Options) (resourcelock.Interface, error) {
+	if !options.LeaderElection {
+		return nil, nil
+	}
+
+	if options.LeaderElectionID == "" || options.LeaderElectionNamespace == "" {
+		return nil, fmt.Errorf("if leader election is enabled, both LeaderElectionID and LeaderElectionNamespace must be set")
+	}
+
+	// Leader id, needs to be unique
+	id, err := os.Hostname()
+	if err != nil {
+		return nil, err
+	}
+	id = id + "_" + string(uuid.NewUUID())
+
+	// Construct client for leader election
+	client, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return nil, err
+	}
+
+	// TODO(JoelSpeed): switch to leaderelection object in 1.12
+	return resourcelock.New(resourcelock.ConfigMapsResourceLock,
+		options.LeaderElectionNamespace,
+		options.LeaderElectionID,
+		client.CoreV1(),
+		resourcelock.ResourceLockConfig{
+			Identity:      id,
+			EventRecorder: recorderProvider.GetEventRecorderFor(id),
+		})
+}

pkg/manager/internal.go

@@ -17,10 +17,14 @@ limitations under the License.
 package manager
 
 import (
+	"fmt"
 	"sync"
+	"time"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -56,6 +60,9 @@ type controllerManager struct {
 	// (and EventHandlers, Sources and Predicates).
 	recorderProvider recorder.Provider
 
+	// resourceLock
+	resourceLock resourcelock.Interface
+
 	mu      sync.Mutex
 	started bool
 	errChan chan error
@@ -133,6 +140,52 @@ func (cm *controllerManager) GetRecorder(name string) record.EventRecorder {
 }
 
 func (cm *controllerManager) Start(stop <-chan struct{}) error {
+	if cm.resourceLock == nil {
+		go cm.start(stop)
+		select {
+		case <-stop:
+			// we are done
+			return nil
+		case err := <-cm.errChan:
+			// Error starting a controller
+			return err
+		}
+	}
+
+	l, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+		Lock: cm.resourceLock,
+		// Values taken from: https://github.com/kubernetes/apiserver/blob/master/pkg/apis/config/v1alpha1/defaults.go
+		// TODO(joelspeed): These timings should be configurable
+		LeaseDuration: 15 * time.Second,
+		RenewDeadline: 10 * time.Second,
+		RetryPeriod:   2 * time.Second,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: cm.start,
+			OnStoppedLeading: func() {
+				// Most implementations of leader election log.Fatal() here.
+				// Since Start is wrapped in log.Fatal when called, we can just return
+				// an error here which will cause the program to exit.
+				cm.errChan <- fmt.Errorf("leader election lost")
+			},
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	go l.Run()
+
+	select {
+	case <-stop:
+		// We are done
+		return nil
+	case err := <-cm.errChan:
+		// Error starting a controller
+		return err
+	}
+}
+
+func (cm *controllerManager) start(stop <-chan struct{}) {
 	func() {
 		cm.mu.Lock()
 		defer cm.mu.Unlock()
@@ -169,9 +222,6 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 	select {
 	case <-stop:
 		// We are done
-		return nil
-	case err := <-cm.errChan:
-		// Error starting a controller
-		return err
+		return
 	}
 }

pkg/manager/manager.go

@@ -25,11 +25,13 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 	internalrecorder "sigs.k8s.io/controller-runtime/pkg/internal/recorder"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/recorder"
 )
 
@@ -83,10 +85,23 @@ type Options struct {
 	// value only if you know what you are doing. Defaults to 10 hours if unset.
 	SyncPeriod *time.Duration
 
+	// LeaderElection determines whether or not to use leader election when
+	// starting the manager.
+	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
+
 	// Dependency injection for testing
 	newCache            func(config *rest.Config, opts cache.Options) (cache.Cache, error)
 	newClient           func(config *rest.Config, options client.Options) (client.Client, error)
 	newRecorderProvider func(config *rest.Config, scheme *runtime.Scheme, logger logr.Logger) (recorder.Provider, error)
+	newResourceLock     func(config *rest.Config, recorderProvider recorder.Provider, options leaderelection.Options) (resourcelock.Interface, error)
 }
 
 // Runnable allows a component to be started.
@@ -140,6 +155,16 @@ func New(config *rest.Config, options Options) (Manager, error) {
 		return nil, err
 	}
 
+	// Create the resource lock to enable leader election)
+	resourceLock, err := options.newResourceLock(config, recorderProvider, leaderelection.Options{
+		LeaderElection:          options.LeaderElection,
+		LeaderElectionID:        options.LeaderElectionID,
+		LeaderElectionNamespace: options.LeaderElectionNamespace,
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	return &controllerManager{
 		config:           config,
 		scheme:           options.Scheme,
@@ -148,6 +173,7 @@ func New(config *rest.Config, options Options) (Manager, error) {
 		fieldIndexes:     cache,
 		client:           client.DelegatingClient{Reader: cache, Writer: writeObj, StatusClient: writeObj},
 		recorderProvider: recorderProvider,
+		resourceLock:     resourceLock,
 	}, nil
 }
 
@@ -177,5 +203,10 @@ func setOptionsDefaults(options Options) Options {
 		options.newRecorderProvider = internalrecorder.NewProvider
 	}
 
+	// Allow newResourceLock to be mocked
+	if options.newResourceLock == nil {
+		options.newResourceLock = leaderelection.NewResourceLock
+	}
+
 	return options
 }

pkg/manager/manager_test.go

@@ -96,6 +96,21 @@ var _ = Describe("manger.Manager", func() {
 
 			close(done)
 		})
+		Context("with leader election enabled", func() {
+			It("should return an error if ID not set", func() {
+				m, err := New(cfg, Options{LeaderElection: true, LeaderElectionNamespace: "default"})
+				Expect(m).To(BeNil())
+				Expect(err).To(HaveOccurred())
+				Expect(err.Error()).To(ContainSubstring("if leader election is enabled, both LeaderElectionID and LeaderElectionNamespace must be set"))
+			})
+
+			It("should return an error if namespace not set", func() {
+				m, err := New(cfg, Options{LeaderElection: true, LeaderElectionID: "controller-runtime"})
+				Expect(m).To(BeNil())
+				Expect(err).To(HaveOccurred())
+				Expect(err.Error()).To(ContainSubstring("if leader election is enabled, both LeaderElectionID and LeaderElectionNamespace must be set"))
+			})
+		})
 	})
 
 	Describe("Start", func() {