Use leader election library from csi-lib-utils

Gopkg.toml

@@ -34,11 +34,6 @@
   name = "sigs.k8s.io/sig-storage-lib-external-provisioner"
   version = ">=v3.0.0-beta"
 
-# TODO: remove when official 0.4.0 tagged
-[[constraint]]
-  name = "github.com/kubernetes-csi/csi-lib-utils"
-  version = ">=0.4.0-rc1"
-
 # TODO: remove when released
 [[constraint]]
   name = "github.com/kubernetes-csi/external-snapshotter"
@@ -57,6 +52,14 @@
   name = "k8s.io/component-base"
   version = "kubernetes-1.14.0"
 
+[[constraint]]
+  name = "k8s.io/client-go"
+  version = "kubernetes-1.14.0"
+
+[[constraint]]
+  name = "github.com/kubernetes-csi/csi-lib-utils"
+  version = ">=0.6.1"
+
 [prune]
   non-go = true
   go-tests = true

README.md

@@ -49,6 +49,8 @@ Note that the external-provisioner does not scale with more replicas. Only one e
 
 * `--enable-leader-election`: Enables leader election. This is mandatory when there are multiple replicas of the same external-provisioner running for one CSI driver. Only one of them may be active (=leader). A new leader will be re-elected when current leader dies or becomes unresponsive for ~15 seconds.
 
+* `--leader-election-type`: The resource type to use for leader election, options are 'endpoints' (default) or 'leases' (recommended)
+
 * `--timeout <duration>`: Timeout of all calls to CSI driver. It should be set to value that accommodates majority of `ControllerCreateVolume` and `ControllerDeleteVolume` calls. See [CSI error and timeout handling](#csi-error-and-timeout-handling) for details. 15 seconds is used by default.
 
 * `--retry-interval-start <duration>` - Initial retry interval of failed provisioning or deletion. It doubles with each failure, up to `--retry-interval-max` and then it stops increasing. Default value is 1 second. See [CSI error and timeout handling](#csi-error-and-timeout-handling) for details.

cmd/csi-provisioner/csi-provisioner.go

@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	goflag "flag"
 	"fmt"
 	"math/rand"
@@ -28,6 +29,7 @@ import (
 	flag "github.com/spf13/pflag"
 
 	"github.com/kubernetes-csi/csi-lib-utils/deprecatedflags"
+	"github.com/kubernetes-csi/csi-lib-utils/leaderelection"
 	ctrl "github.com/kubernetes-csi/external-provisioner/pkg/controller"
 	snapclientset "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned"
 	"sigs.k8s.io/sig-storage-lib-external-provisioner/controller"
@@ -52,19 +54,25 @@ var (
 	volumeNamePrefix     = flag.String("volume-name-prefix", "pvc", "Prefix to apply to the name of a created volume.")
 	volumeNameUUIDLength = flag.Int("volume-name-uuid-length", -1, "Truncates generated UUID of a created volume to this length. Defaults behavior is to NOT truncate.")
 	showVersion          = flag.Bool("version", false, "Show version.")
-	enableLeaderElection = flag.Bool("enable-leader-election", false, "Enables leader election. If leader election is enabled, additional RBAC rules are required. Please refer to the Kubernetes CSI documentation for instructions on setting up these RBAC rules.")
 	retryIntervalStart   = flag.Duration("retry-interval-start", time.Second, "Initial retry interval of failed provisioning or deletion. It doubles with each failure, up to retry-interval-max.")
 	retryIntervalMax     = flag.Duration("retry-interval-max", 5*time.Minute, "Maximum retry interval of failed provisioning or deletion.")
 	workerThreads        = flag.Uint("worker-threads", 100, "Number of provisioner worker threads, in other words nr. of simultaneous CSI calls.")
 	operationTimeout     = flag.Duration("timeout", 10*time.Second, "Timeout for waiting for creation or deletion of a volume")
 	_                    = deprecatedflags.Add("provisioner")
 
+	enableLeaderElection = flag.Bool("enable-leader-election", false, "Enables leader election. If leader election is enabled, additional RBAC rules are required. Please refer to the Kubernetes CSI documentation for instructions on setting up these RBAC rules.")
+	leaderElectionType   = flag.String("leader-election-type", "endpoints", "the type of leader election, options are 'endpoints' (default) or 'leases' (strongly recommended). The 'endpoints' option is deprecated in favor of 'leases'.")
+
 	featureGates        map[string]bool
 	provisionController *controller.ProvisionController
 	version             = "unknown"
 )
 
-func init() {
+type leaderElection interface {
+	Run() error
+}
+
+func main() {
 	var config *rest.Config
 	var err error
 
@@ -178,8 +186,32 @@ func init() {
 		serverVersion.GitVersion,
 		provisionerOptions...,
 	)
-}
 
-func main() {
-	provisionController.Run(wait.NeverStop)
+	run := func(context.Context) {
+		provisionController.Run(wait.NeverStop)
+	}
+
+	if !*enableLeaderElection {
+		run(context.TODO())
+	} else {
+		// this lock name pattern is also copied from sigs.k8s.io/sig-storage-lib-external-provisioner/controller
+		// to preserve backwards compatibility
+		lockName := strings.Replace(provisionerName, "/", "-", -1)
+
+		var le leaderElection
+		if *leaderElectionType == "endpoints" {
+			klog.Warning("The 'endpoints' leader election type is deprecated and will be removed in a future release. Use '--leader-election-type=leases' instead.")
+			le = leaderelection.NewLeaderElectionWithEndpoints(clientset, lockName, run)
+		} else if *leaderElectionType == "leases" {
+			le = leaderelection.NewLeaderElection(clientset, lockName, run)
+		} else {
+			klog.Error("--leader-election-type must be either 'endpoints' or 'lease'")
+			os.Exit(1)
+		}
+
+		if err := le.Run(); err != nil {
+			klog.Fatalf("failed to initialize leader election: %v", err)
+		}
+	}
+
 }

deploy/kubernetes/rbac.yaml

@@ -76,9 +76,14 @@ metadata:
   namespace: default
   name: external-provisioner-cfg
 rules:
+# Only one of the following rules for endpoints or leases is required based on
+# what is set for `--leader-election-type`. Endpoints are deprecated in favor of Leases.
 - apiGroups: [""]
   resources: ["endpoints"]
   verbs: ["get", "watch", "list", "delete", "update", "create"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["get", "watch", "list", "delete", "update", "create"]
 
 ---
 kind: RoleBinding