leaderelection: retrofit controller-manager with leaderelection client

Signed-off-by: Mike Danese <mikedanese@google.com>

cmd/kube-controller-manager/app/controllermanager.go

@@ -29,11 +29,15 @@ import (
 	"net"
 	"net/http"
 	"net/http/pprof"
+	"os"
 	"strconv"
 	"time"
 
 	"k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
+	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/client/leaderelection"
+	"k8s.io/kubernetes/pkg/client/record"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
 	"k8s.io/kubernetes/pkg/cloudprovider"
@@ -135,6 +139,49 @@ func Run(s *options.CMServer) error {
 		glog.Fatal(server.ListenAndServe())
 	}()
 
+	run := func(stop <-chan struct{}) {
+		err := StartControllers(s, kubeClient, kubeconfig, stop)
+		glog.Fatalf("error running controllers: %v", err)
+		panic("unreachable")
+	}
+
+	if !s.LeaderElection.LeaderElect {
+		run(nil)
+		panic("unreachable")
+	}
+
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(glog.Infof)
+	eventBroadcaster.StartRecordingToSink(kubeClient.Events(""))
+	recorder := eventBroadcaster.NewRecorder(api.EventSource{Component: "controller-manager"})
+
+	id, err := os.Hostname()
+	if err != nil {
+		return err
+	}
+
+	leaderelection.RunOrDie(leaderelection.LeaderElectionConfig{
+		EndpointsMeta: api.ObjectMeta{
+			Namespace: "kube-system",
+			Name:      "kube-controller-manager",
+		},
+		Client:        kubeClient,
+		Identity:      id,
+		EventRecorder: recorder,
+		LeaseDuration: s.LeaderElection.LeaseDuration.Duration,
+		RenewDeadline: s.LeaderElection.RenewDeadline.Duration,
+		RetryPeriod:   s.LeaderElection.RetryPeriod.Duration,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: run,
+			OnStoppedLeading: func() {
+				glog.Fatalf("leaderelection lost")
+			},
+		},
+	})
+	panic("unreachable")
+}
+
+func StartControllers(s *options.CMServer, kubeClient *client.Client, kubeconfig *client.Config, stop <-chan struct{}) error {
 	go endpointcontroller.NewEndpointController(clientForUserAgentOrDie(*kubeconfig, "endpoint-controller"), ResyncPeriod(s)).
 		Run(s.ConcurrentEndpointSyncs, util.NeverStop)
 

cmd/kube-controller-manager/app/options/options.go

@@ -24,6 +24,8 @@ import (
 	"net"
 	"time"
 
+	"k8s.io/kubernetes/pkg/apis/componentconfig"
+	"k8s.io/kubernetes/pkg/client/leaderelection"
 	"k8s.io/kubernetes/pkg/master/ports"
 
 	"github.com/spf13/pflag"
@@ -71,6 +73,8 @@ type CMServer struct {
 	Kubeconfig   string
 	KubeAPIQPS   float32
 	KubeAPIBurst int
+
+	LeaderElection componentconfig.LeaderElectionConfiguration
 }
 
 // VolumeConfigFlags is used to bind CLI flags to variables.  This top-level struct contains *all* enumerated
@@ -121,8 +125,9 @@ func NewCMServer() *CMServer {
 			PersistentVolumeRecyclerIncrementTimeoutHostPath: 30,
 			EnableHostPathProvisioning:                       false,
 		},
-		KubeAPIQPS:   20.0,
+		KubeAPIQPS:     20.0,
-		KubeAPIBurst: 30,
+		KubeAPIBurst:   30,
+		LeaderElection: leaderelection.DefaultLeaderElectionConfiguration(),
 	}
 	return &s
 }
@@ -179,4 +184,5 @@ func (s *CMServer) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&s.RootCAFile, "root-ca-file", s.RootCAFile, "If set, this root certificate authority will be included in service account's token secret. This must be a valid PEM-encoded CA bundle.")
 	fs.Float32Var(&s.KubeAPIQPS, "kube-api-qps", s.KubeAPIQPS, "QPS to use while talking with kubernetes apiserver")
 	fs.IntVar(&s.KubeAPIBurst, "kube-api-burst", s.KubeAPIBurst, "Burst to use while talking with kubernetes apiserver")
+	leaderelection.BindFlags(&s.LeaderElection, fs)
 }

docs/admin/kube-controller-manager.md

@@ -74,6 +74,10 @@ kube-controller-manager
       --kube-api-burst=30: Burst to use while talking with kubernetes apiserver
       --kube-api-qps=20: QPS to use while talking with kubernetes apiserver
       --kubeconfig="": Path to kubeconfig file with authorization and master location information.
+      --leader-elect[=false]: Start a leader election client and gain leadership before executing the main loop. Enable this when running replicated components for high availability.
+      --leader-elect-lease-duration=15s: The duration that non-leader candidates will wait after observing a leadership renewal until attempting to acquire leadership of a led but unrenewed leader slot. This is effectively the maximum duration that a leader can be stopped before it is replaced by another candidate. This is only applicable if leader election is enabled.
+      --leader-elect-renew-deadline=10s: The interval between attempts by the acting master to renew a leadership slot before it stops leading. This must be less than or equal to the lease duration. This is only applicable if leader election is enabled.
+      --leader-elect-retry-period=2s: The duration the clients should wait between attempting acquisition and renewal of a leadership. This is only applicable if leader election is enabled.
       --log-flush-frequency=5s: Maximum number of seconds between log flushes
       --master="": The address of the Kubernetes API server (overrides any value in kubeconfig)
       --min-resync-period=12h0m0s: The resync period in reflectors will be random between MinResyncPeriod and 2*MinResyncPeriod
@@ -99,7 +103,7 @@ kube-controller-manager
       --terminated-pod-gc-threshold=12500: Number of terminated pods that can exist before the terminated pod garbage collector starts deleting terminated pods. If <= 0, the terminated pod garbage collector is disabled.
 ```
 
-###### Auto generated by spf13/cobra on 9-Dec-2015
+###### Auto generated by spf13/cobra on 18-Jan-2016
 
 
 <!-- BEGIN MUNGE: GENERATED_ANALYTICS -->

docs/admin/kube-scheduler.md

@@ -60,8 +60,8 @@ kube-scheduler
       --kube-api-burst=100: Burst to use while talking with kubernetes apiserver
       --kube-api-qps=50: QPS to use while talking with kubernetes apiserver
       --kubeconfig="": Path to kubeconfig file with authorization and master location information.
-      --leader-elect[=false]: Start a leader election client and gain leadership before executing scheduler loop. Enable this when running replicated schedulers.
+      --leader-elect[=false]: Start a leader election client and gain leadership before executing the main loop. Enable this when running replicated components for high availability.
-      --leader-elect-lease-duration=15s: The duration that non-leader candidates will wait after observing a leadershiprenewal until attempting to acquire leadership of a led but unrenewed leader slot. This is effectively the maximum duration that a leader can be stopped before it is replaced by another candidate. This is only applicable if leader election is enabled.
+      --leader-elect-lease-duration=15s: The duration that non-leader candidates will wait after observing a leadership renewal until attempting to acquire leadership of a led but unrenewed leader slot. This is effectively the maximum duration that a leader can be stopped before it is replaced by another candidate. This is only applicable if leader election is enabled.
       --leader-elect-renew-deadline=10s: The interval between attempts by the acting master to renew a leadership slot before it stops leading. This must be less than or equal to the lease duration. This is only applicable if leader election is enabled.
       --leader-elect-retry-period=2s: The duration the clients should wait between attempting acquisition and renewal of a leadership. This is only applicable if leader election is enabled.
       --log-flush-frequency=5s: Maximum number of seconds between log flushes
@@ -72,7 +72,7 @@ kube-scheduler
       --scheduler-name="default-scheduler": Name of the scheduler, used to select which pods will be processed by this scheduler, based on pod's annotation with key 'scheduler.alpha.kubernetes.io/name'
 ```
 
-###### Auto generated by spf13/cobra on 13-Jan-2016
+###### Auto generated by spf13/cobra on 18-Jan-2016
 
 
 <!-- BEGIN MUNGE: GENERATED_ANALYTICS -->

pkg/client/leaderelection/leaderelection.go

@@ -345,10 +345,10 @@ func DefaultLeaderElectionConfiguration() componentconfig.LeaderElectionConfigur
 func BindFlags(l *componentconfig.LeaderElectionConfiguration, fs *pflag.FlagSet) {
 	fs.BoolVar(&l.LeaderElect, "leader-elect", l.LeaderElect, ""+
 		"Start a leader election client and gain leadership before "+
-		"executing scheduler loop. Enable this when running replicated "+
+		"executing the main loop. Enable this when running replicated "+
-		"schedulers.")
+		"components for high availability.")
 	fs.DurationVar(&l.LeaseDuration.Duration, "leader-elect-lease-duration", l.LeaseDuration.Duration, ""+
-		"The duration that non-leader candidates will wait after observing a leadership"+
+		"The duration that non-leader candidates will wait after observing a leadership "+
 		"renewal until attempting to acquire leadership of a led but unrenewed leader "+
 		"slot. This is effectively the maximum duration that a leader can be stopped "+
 		"before it is replaced by another candidate. This is only applicable if leader "+