use csi-lib-utils for leader election, also add support for lease based leader election

Signed-off-by: Andrew Sy Kim <[email protected]>

Author: andrewsykim

cmd/csi-provisioner/csi-provisioner.go

@@ -17,8 +17,10 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	goflag "flag"
 	"fmt"
+	"io/ioutil"
 	"math/rand"
 	"os"
 	"strconv"
@@ -27,6 +29,7 @@ import (
 
 	flag "github.com/spf13/pflag"
 
+	"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"
@@ -51,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")
 	provisioner          = flag.String("provisioner", "", "This option is deprecated")
 
+	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 'lease' (strongly recommended)")
+
 	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
 
@@ -184,8 +193,50 @@ func init() {
 		serverVersion.GitVersion,
 		provisionerOptions...,
 	)
+
+	run := func(context.Context) {
+		provisionController.Run(wait.NeverStop)
+	}
+
+	if !*enableLeaderElection {
+		run(context.TODO())
+	} else {
+		// getInClusterNamespace is copied from sigs.k8s.io/sig-storage-lib-external-provisioner/controller
+		// to preserve backwards compatibilty
+		namespace := getInClusterNamespace()
+		// 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" {
+			le = leaderelection.NewLeaderElectionWithEndpoints(clientset, lockName, namespace, run)
+		} else if *leaderElectionType == "lease" {
+			le = leaderelection.NewLeaderElection(clientset, lockName, namespace, 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)
+		}
+	}
+
 }
 
-func main() {
-	provisionController.Run(wait.NeverStop)
+// getInClusterNamespace returns the namespace in which the controller runs.
+func getInClusterNamespace() string {
+	if ns := os.Getenv("POD_NAMESPACE"); ns != "" {
+		return ns
+	}
+
+	// Fall back to the namespace associated with the service account token, if available
+	if data, err := ioutil.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"); err == nil {
+		if ns := strings.TrimSpace(string(data)); len(ns) > 0 {
+			return ns
+		}
+	}
+
+	return "default"
 }