Merge pull request #530 from mhrivnak/leaderelection

Added leader election.

Author: mhrivnak

pkg/leader/doc.go

@@ -0,0 +1,46 @@
+// Copyright 2018 The Operator-SDK 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 leader implements Leader For Life, a simple alternative to lease-based
+leader election.
+
+Both the Leader For Life and lease-based approaches to leader election are
+built on the concept that each candidate will attempt to create a resource with
+the same GVK, namespace, and name. Whichever candidate succeeds becomes the
+leader. The rest receive "already exists" errors and wait for a new
+opportunity.
+
+Leases provide a way to indirectly observe whether the leader still exists. The
+leader must periodically renew its lease, usually by updating a timestamp in
+its lock record. If it fails to do so, it is presumed dead, and a new election
+takes place. If the leader is in fact still alive but unreachable, it is
+expected to gracefully step down. A variety of factors can cause a leader to
+fail at updating its lease, but continue acting as the leader before succeeding
+at stepping down.
+
+In the "leader for life" approach, a specific Pod is the leader. Once
+established (by creating a lock record), the Pod is the leader until it is
+destroyed. There is no possibility for multiple pods to think they are the
+leader at the same time. The leader does not need to renew a lease, consider
+stepping down, or do anything related to election activity once it becomes the
+leader.
+
+The lock record in this case is a ConfigMap whose OwnerReference is set to the
+Pod that is the leader. When the leader is destroyed, the ConfigMap gets
+garbage-collected, enabling a different candidate Pod to become the leader.
+
+Leader for Life requires that all candidate Pods be in the same Namespace.
+*/
+package leader

pkg/leader/leader.go

@@ -0,0 +1,187 @@
+// Copyright 2018 The Operator-SDK 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 leader
+
+import (
+	"context"
+	"errors"
+	"io/ioutil"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/rest"
+	crclient "sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// errNoNS indicates that a namespace could not be found for the current
+// environment
+var errNoNS = errors.New("namespace not found for current environment")
+
+// maxBackoffInterval defines the maximum amount of time to wait between
+// attempts to become the leader.
+const maxBackoffInterval = time.Second * 16
+
+// Become ensures that the current pod is the leader within its namespace. If
+// run outside a cluster, it will skip leader election and return nil. It
+// continuously tries to create a ConfigMap with the provided name and the
+// current pod set as the owner reference. Only one can exist at a time with
+// the same name, so the pod that successfully creates the ConfigMap is the
+// leader. Upon termination of that pod, the garbage collector will delete the
+// ConfigMap, enabling a different pod to become the leader.
+func Become(ctx context.Context, lockName string) error {
+	logrus.Info("trying to become the leader")
+
+	ns, err := myNS()
+	if err != nil {
+		if err == errNoNS {
+			logrus.Info("Skipping leader election; not running in a cluster")
+			return nil
+		}
+		return err
+	}
+
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		return err
+	}
+
+	client, err := crclient.New(config, crclient.Options{})
+	if err != nil {
+		return err
+	}
+
+	owner, err := myOwnerRef(ctx, client, ns)
+	if err != nil {
+		return err
+	}
+
+	// check for existing lock from this pod, in case we got restarted
+	existing := &corev1.ConfigMap{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "ConfigMap",
+		},
+	}
+	key := crclient.ObjectKey{Namespace: ns, Name: lockName}
+	err = client.Get(ctx, key, existing)
+
+	switch {
+	case err == nil:
+		for _, existingOwner := range existing.GetOwnerReferences() {
+			if existingOwner.Name == owner.Name {
+				logrus.Info("Found existing lock with my name. I was likely restarted.")
+				logrus.Info("Continuing as the leader.")
+				return nil
+			} else {
+				logrus.Infof("Found existing lock from %s", existingOwner.Name)
+			}
+		}
+	case apierrors.IsNotFound(err):
+		logrus.Info("No pre-existing lock was found.")
+	default:
+		logrus.Error("unknown error trying to get ConfigMap")
+		return err
+	}
+
+	cm := &corev1.ConfigMap{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "ConfigMap",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            lockName,
+			Namespace:       ns,
+			OwnerReferences: []metav1.OwnerReference{*owner},
+		},
+	}
+
+	// try to create a lock
+	backoff := time.Second
+	for {
+		err := client.Create(ctx, cm)
+		switch {
+		case err == nil:
+			logrus.Info("Became the leader.")
+			return nil
+		case apierrors.IsAlreadyExists(err):
+			logrus.Info("Not the leader. Waiting.")
+			select {
+			case <-time.After(wait.Jitter(backoff, .2)):
+				if backoff < maxBackoffInterval {
+					backoff *= 2
+				}
+				continue
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		default:
+			logrus.Errorf("unknown error creating configmap: %v", err)
+			return err
+		}
+	}
+}
+
+// myNS returns the name of the namespace in which this code is currently running.
+func myNS() (string, error) {
+	nsBytes, err := ioutil.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace")
+	if err != nil {
+		if os.IsNotExist(err) {
+			logrus.Debug("current namespace not found")
+			return "", errNoNS
+		}
+		return "", err
+	}
+	ns := strings.TrimSpace(string(nsBytes))
+	logrus.Debugf("found namespace: %s", ns)
+	return ns, nil
+}
+
+// myOwnerRef returns an OwnerReference that corresponds to the pod in which
+// this code is currently running.
+func myOwnerRef(ctx context.Context, client crclient.Client, ns string) (*metav1.OwnerReference, error) {
+	hostname, err := os.Hostname()
+	if err != nil {
+		return nil, err
+	}
+	logrus.Debugf("found hostname: %s", hostname)
+
+	myPod := &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "Pod",
+		},
+	}
+
+	key := crclient.ObjectKey{Namespace: ns, Name: hostname}
+	err = client.Get(ctx, key, myPod)
+	if err != nil {
+		logrus.Errorf("failed to get pod: %v", err)
+		return nil, err
+	}
+
+	owner := &metav1.OwnerReference{
+		APIVersion: "v1",
+		Kind:       "Pod",
+		Name:       myPod.ObjectMeta.Name,
+		UID:        myPod.ObjectMeta.UID,
+	}
+	return owner, nil
+}