Turns on leader election by default and adds e2e tests

Adds a readiness probe by default that reflects whether a pod is the leader.

fixes operator-framework#598

pkg/scaffold/cmd.go

@@ -37,14 +37,17 @@ func (s *Cmd) GetInput() (input.Input, error) {
 const cmdTmpl = `package main
 
 import (
+	"context"
 	"flag"
 	"log"
+	"os"
 	"runtime"
 
 	"{{ .Repo }}/pkg/apis"
 	"{{ .Repo }}/pkg/controller"
 
 	"github.com/operator-framework/operator-sdk/pkg/k8sutil"
+	"github.com/operator-framework/operator-sdk/pkg/leader"
 	sdkVersion "github.com/operator-framework/operator-sdk/version"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
@@ -73,6 +76,20 @@ func main() {
 		log.Fatal(err)
 	}
 
+	// Become the leader before proceeding
+	leader.Become(context.TODO(), "{{ .ProjectName }}-lock")
+
+	// Create file for readiness probe
+	f, err := os.Create("/tmp/operator-sdk-leader")
+	if err != nil {
+		log.Fatal(err)
+	}
+	err = f.Close()
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer os.Remove(f.Name())
+
 	// Create a new Cmd to provide shared dependencies and start components
 	mgr, err := manager.New(cfg, manager.Options{Namespace: namespace})
 	if err != nil {

pkg/scaffold/cmd_test.go

@@ -34,13 +34,16 @@ func TestCmd(t *testing.T) {
 const cmdExp = `package main
 
 import (
+	"context"
 	"flag"
 	"log"
+	"os"
 	"runtime"
 
 	"github.com/example-inc/app-operator/pkg/apis"
 	"github.com/example-inc/app-operator/pkg/controller"
 	"github.com/operator-framework/operator-sdk/pkg/k8sutil"
+	"github.com/operator-framework/operator-sdk/pkg/leader"
 	sdkVersion "github.com/operator-framework/operator-sdk/version"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
@@ -69,6 +72,20 @@ func main() {
 		log.Fatal(err)
 	}
 
+	// Become the leader before proceeding
+	leader.Become(context.TODO(), "app-operator-lock")
+
+	// Create file for readiness probe
+	f, err := os.Create("/tmp/operator-sdk-leader")
+	if err != nil {
+		log.Fatal(err)
+	}
+	err = f.Close()
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer os.Remove(f.Name())
+
 	// Create a new Cmd to provide shared dependencies and start components
 	mgr, err := manager.New(cfg, manager.Options{Namespace: namespace})
 	if err != nil {

pkg/scaffold/operator.go

@@ -59,6 +59,14 @@ spec:
           command:
           - {{.ProjectName}}
           imagePullPolicy: Always
+          readinessProbe:
+            exec:
+              command:
+                - stat
+                - /tmp/operator-sdk-leader
+            initialDelaySeconds: 4
+            periodSeconds: 10
+            failureThreshold: 1
           env:
             - name: WATCH_NAMESPACE
               valueFrom:

pkg/scaffold/operator_test.go

@@ -56,6 +56,14 @@ spec:
           command:
           - app-operator
           imagePullPolicy: Always
+          readinessProbe:
+            exec:
+              command:
+                - stat
+                - /tmp/operator-sdk-leader
+            initialDelaySeconds: 4
+            periodSeconds: 10
+            failureThreshold: 1
           env:
             - name: WATCH_NAMESPACE
               valueFrom:

test/e2e/memcached_test.go

@@ -16,6 +16,8 @@ package e2e
 
 import (
 	"bytes"
+	goctx "context"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"os/exec"
@@ -28,8 +30,10 @@ import (
 	"github.com/operator-framework/operator-sdk/test/e2e/e2eutil"
 	framework "github.com/operator-framework/operator-sdk/test/e2e/framework"
 
+	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 const (
@@ -113,6 +117,12 @@ func TestMemcached(t *testing.T) {
 		t.Fatalf("error: %v\nCommand Output: %s\n", err, string(cmdOut))
 	}
 
+	// Set replicas to 2 to test leader election. In production, this should almost always be set to 1.
+	cmdOut, err = exec.Command("sed", "-i", "-e", "s/replicas: 1/replicas: 2/", "deploy/operator.yaml").CombinedOutput()
+	if err != nil {
+		t.Fatalf("could not change operator replicas to 2: %v\nCommand Output:\n%v", err, string(cmdOut))
+	}
+
 	cmdOut, err = exec.Command("operator-sdk",
 		"add",
 		"api",
@@ -215,6 +225,80 @@ func TestMemcached(t *testing.T) {
 	})
 }
 
+func memcachedLeaderTest(t *testing.T, f *framework.Framework, ctx framework.TestCtx) error {
+	namespace, err := ctx.GetNamespace()
+	if err != nil {
+		return err
+	}
+
+	err = e2eutil.DeploymentReplicaCheck(t, f.KubeClient, namespace, "memcached-operator", 1, 6)
+	if err != nil {
+		return err
+	}
+
+	leader, err := verifyLeader(namespace, f)
+	if err != nil {
+		return err
+	}
+
+	// delete the leader's pod so a new leader will get elected
+	err = f.DynamicClient.Delete(goctx.TODO(), leader)
+	if err != nil {
+		return err
+	}
+
+	err = e2eutil.DeploymentReplicaCheck(t, f.KubeClient, namespace, "memcached-operator", 1, 6)
+	if err != nil {
+		return err
+	}
+
+	newLeader, err := verifyLeader(namespace, f)
+	if err != nil {
+		return err
+	}
+	if newLeader.Name == leader.Name {
+		return fmt.Errorf("leader pod name did not change across pod delete")
+	}
+
+	return nil
+}
+
+func verifyLeader(namespace string, f *framework.Framework) (*v1.Pod, error) {
+	// get configmap, which is the lock
+	lock := v1.ConfigMap{}
+	err := f.DynamicClient.Get(goctx.TODO(), types.NamespacedName{Name: "memcached-operator-lock", Namespace: namespace}, &lock)
+	if err != nil {
+		return nil, err
+	}
+
+	owners := lock.GetOwnerReferences()
+	if len(owners) != 1 {
+		return nil, fmt.Errorf("leader lock has %d owner refs, expected 1", len(owners))
+	}
+	owner := owners[0]
+
+	// get operator pods
+	pods := v1.PodList{}
+	opts := client.ListOptions{Namespace: namespace}
+	opts.SetLabelSelector("name=memcached-operator")
+	opts.SetFieldSelector("status.phase=Running")
+	err = f.DynamicClient.List(goctx.TODO(), &opts, &pods)
+	if err != nil {
+		return nil, err
+	}
+	if len(pods.Items) != 2 {
+		return nil, fmt.Errorf("expected 2 pods, found %d", len(pods.Items))
+	}
+
+	// find and return the leader
+	for _, pod := range pods.Items {
+		if pod.Name == owner.Name {
+			return &pod, nil
+		}
+	}
+	return nil, fmt.Errorf("did not find operator pod that was referenced by configmap")
+}
+
 func memcachedScaleTest(t *testing.T, f *framework.Framework, ctx framework.TestCtx) error {
 	// create example-memcached yaml file
 	err := ioutil.WriteFile("deploy/cr.yaml",
@@ -401,6 +485,10 @@ func MemcachedCluster(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	if err = memcachedLeaderTest(t, f, ctx); err != nil {
+		t.Fatal(err)
+	}
+
 	if err = memcachedScaleTest(t, f, ctx); err != nil {
 		t.Fatal(err)
 	}