Add leader election to controller manager

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/fake/doc.go

@@ -0,0 +1,21 @@
+/*
+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 fake mocks a resource lock for testing purposes.
+Always returns leadership.
+*/
+package fake

pkg/leaderelection/fake/leader_election.go

@@ -0,0 +1,90 @@
+/*
+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 fake
+
+import (
+	"os"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
+	"sigs.k8s.io/controller-runtime/pkg/recorder"
+)
+
+// NewResourceLock creates a new ResourceLock for use in testing
+// leader election.
+func NewResourceLock(config *rest.Config, recorderProvider recorder.Provider, options leaderelection.Options) (resourcelock.Interface, error) {
+	// Leader id, needs to be unique
+	id, err := os.Hostname()
+	if err != nil {
+		return nil, err
+	}
+	id = id + "_" + string(uuid.NewUUID())
+
+	return &ResourceLock{
+		id: id,
+		record: resourcelock.LeaderElectionRecord{
+			HolderIdentity:       id,
+			LeaseDurationSeconds: 15,
+			AcquireTime:          metav1.NewTime(time.Now()),
+			RenewTime:            metav1.NewTime(time.Now().Add(15 * time.Second)),
+			LeaderTransitions:    1,
+		},
+	}, nil
+}
+
+// ResourceLock implements the ResourceLockInterface.
+// By default returns that the current identity holds the lock.
+type ResourceLock struct {
+	id     string
+	record resourcelock.LeaderElectionRecord
+}
+
+// Get implements the ResourceLockInterface.
+func (f *ResourceLock) Get() (*resourcelock.LeaderElectionRecord, error) {
+	return &f.record, nil
+}
+
+// Create implements the ResourceLockInterface.
+func (f *ResourceLock) Create(ler resourcelock.LeaderElectionRecord) error {
+	f.record = ler
+	return nil
+}
+
+// Update implements the ResourceLockInterface.
+func (f *ResourceLock) Update(ler resourcelock.LeaderElectionRecord) error {
+	f.record = ler
+	return nil
+}
+
+// RecordEvent implements the ResourceLockInterface.
+func (f *ResourceLock) RecordEvent(s string) {
+	return
+}
+
+// Identity implements the ResourceLockInterface.
+func (f *ResourceLock) Identity() string {
+	return f.id
+}
+
+// Describe implements the ResourceLockInterface.
+func (f *ResourceLock) Describe() string {
+	return f.id
+}

pkg/leaderelection/leader_election.go

@@ -0,0 +1,109 @@
+/*
+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"
+	"io/ioutil"
+	"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"
+)
+
+const inClusterNamespacePath = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
+
+// 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
+	}
+
+	// Default the LeaderElectionID
+	if options.LeaderElectionID == "" {
+		options.LeaderElectionID = "controller-leader-election-helper"
+	}
+
+	// Default the namespace (if running in cluster)
+	if options.LeaderElectionNamespace == "" {
+		var err error
+		options.LeaderElectionNamespace, err = getInClusterNamespace()
+		if err != nil {
+			return nil, fmt.Errorf("unable to find leader election namespace: %v", err)
+		}
+	}
+
+	// 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),
+		})
+}
+
+func getInClusterNamespace() (string, error) {
+	// Check whether the namespace file exists.
+	// If not, we are not running in cluster so can't guess the namespace.
+	_, err := os.Stat(inClusterNamespacePath)
+	if os.IsNotExist(err) {
+		return "", fmt.Errorf("not running in-cluster, please specify LeaderElectionNamespace")
+	} else if err != nil {
+		return "", fmt.Errorf("error checking namespace file: %v", err)
+	}
+
+	// Load the namespace file and return itss content
+	namespace, err := ioutil.ReadFile(inClusterNamespacePath)
+	if err != nil {
+		return "", fmt.Errorf("error reading namespace file: %v", err)
+	}
+	return string(namespace), nil
+}

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
 	}
 }