add leader election for snapshotter

cmd/csi-snapshotter/leader.go

@@ -0,0 +1,92 @@
+/*
+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 main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/golang/glog"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"k8s.io/client-go/tools/record"
+)
+
+const (
+	leaseDuration = 15 * time.Second
+	renewDeadline = 10 * time.Second
+	retryPeriod   = 5 * time.Second
+)
+
+// WaitForLeader waits until this particular external snapshotter becomes a leader.
+func WaitForLeader(clientset *kubernetes.Clientset, namespace string, lockName string) {
+	hostname, err := os.Hostname()
+	if err != nil {
+		glog.Error(err)
+		os.Exit(1)
+	}
+	// add a uniquifier so that two processes on the same host don't accidentally both become active
+	identity := hostname + "_" + string(uuid.NewUUID())
+
+	broadcaster := record.NewBroadcaster()
+	broadcaster.StartRecordingToSink(&corev1.EventSinkImpl{Interface: clientset.CoreV1().Events(namespace)})
+	eventRecorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: fmt.Sprintf("%s %s", lockName, string(identity))})
+
+	rlConfig := resourcelock.ResourceLockConfig{
+		Identity:      identity,
+		EventRecorder: eventRecorder,
+	}
+	lock, err := resourcelock.New(resourcelock.ConfigMapsResourceLock, namespace, lockName, clientset.CoreV1(), rlConfig)
+	if err != nil {
+		glog.Error(err)
+		os.Exit(1)
+	}
+
+	elected := make(chan struct{})
+
+	leaderConfig := leaderelection.LeaderElectionConfig{
+		Lock:          lock,
+		LeaseDuration: leaseDuration,
+		RenewDeadline: renewDeadline,
+		RetryPeriod:   retryPeriod,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				glog.V(2).Info("Became leader, starting")
+				close(elected)
+			},
+			OnStoppedLeading: func() {
+				glog.Error("Stopped leading")
+				os.Exit(1)
+			},
+			OnNewLeader: func(identity string) {
+				glog.V(3).Infof("Current leader: %s", identity)
+			},
+		},
+	}
+
+	go leaderelection.RunOrDie(context.TODO(), leaderConfig)
+
+	// wait for being elected
+	<-elected
+}

cmd/csi-snapshotter/main.go

@@ -58,6 +58,8 @@ var (
 	resyncPeriod                    = flag.Duration("resync-period", 60*time.Second, "Resync interval of the controller.")
 	snapshotNamePrefix              = flag.String("snapshot-name-prefix", "snapshot", "Prefix to apply to the name of a created snapshot")
 	snapshotNameUUIDLength          = flag.Int("snapshot-name-uuid-length", -1, "Length in characters for the generated uuid of a created snapshot. Defaults behavior is to NOT truncate.")
+	enableLeaderElection            = flag.Bool("leader-election", false, "Enable leader election.")
+	leaderElectionNamespace         = flag.String("leader-election-namespace", "", "Namespace where this snapshotter runs.")
 	showVersion                     = flag.Bool("version", false, "Show version.")
 )
 
@@ -156,6 +158,17 @@ func main() {
 		os.Exit(1)
 	}
 
+	if *enableLeaderElection {
+		// Leader election was requested.
+		if leaderElectionNamespace == nil || *leaderElectionNamespace == "" {
+			glog.Error("-leader-election-namespace must not be empty")
+			os.Exit(1)
+		}
+		// Name of config map with leader election lock
+		lockName := "external-snapshotter-leader-" + *snapshotter
+		WaitForLeader(kubeClient, *leaderElectionNamespace, lockName)
+	}
+
 	glog.V(2).Infof("Start NewCSISnapshotController with snapshotter [%s] kubeconfig [%s] connectionTimeout [%+v] csiAddress [%s] createSnapshotContentRetryCount [%d] createSnapshotContentInterval [%+v] resyncPeriod [%+v] snapshotNamePrefix [%s] snapshotNameUUIDLength [%d]", *snapshotter, *kubeconfig, *connectionTimeout, *csiAddress, createSnapshotContentRetryCount, *createSnapshotContentInterval, *resyncPeriod, *snapshotNamePrefix, snapshotNameUUIDLength)
 
 	ctrl := controller.NewCSISnapshotController(

deploy/kubernetes/rbac.yaml

@@ -63,3 +63,34 @@ roleRef:
   # change the name also here if the ClusterRole gets renamed
   name: external-snapshotter-runner
   apiGroup: rbac.authorization.k8s.io
+
+---
+# Snapshotter must be able to work with config map in current namespace
+# if (and only if) leadership election is enabled
+kind: Role
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  # replace with non-default namespace name
+  namespace: default
+  name: external-snapshotter-cfg
+rules:
+- apiGroups: [""]
+  resources: ["configmaps"]
+  verbs: ["get", "watch", "list", "delete", "update", "create"]
+
+---
+kind: RoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: csi-snapshotter-role-cfg
+  # replace with non-default namespace name
+  namespace: default
+subjects:
+- kind: ServiceAccount
+  name: csi-snapshotter
+  # replace with non-default namespace name
+  namespace: default
+roleRef:
+  kind: Role
+  name: external-snapshotter-cfg
+  apiGroup: rbac.authorization.k8s.io