Merge pull request #128 from M00nF1sh/election

add leaderElection

Author: M00nF1sh

cmd/app-mesh-controller/root.go

@@ -23,11 +23,14 @@ import (
 )
 
 var (
-	cfgFile     string
-	master      string
-	kubeconfig  string
-	region      string
-	threadiness int
+	cfgFile                 string
+	master                  string
+	kubeconfig              string
+	region                  string
+	threadiness             int
+	leaderElection          bool
+	leaderElectionID        string
+	leaderElectionNamespace string
 )
 
 func init() {
@@ -37,10 +40,16 @@ func init() {
 	rootCmd.Flags().StringVar(&kubeconfig, "kubeconfig", "", "Path to your kubeconfig")
 	rootCmd.Flags().StringVar(&region, "aws-region", "", "AWS Region")
 	rootCmd.Flags().IntVar(&threadiness, "threadiness", controller.DefaultThreadiness, "Worker concurrency.")
+	rootCmd.Flags().BoolVar(&leaderElection, "election", controller.DefaultElection, `Whether to do leader election for controller`)
+	rootCmd.Flags().StringVar(&leaderElectionID, "election-id", controller.DefaultElectionID, "Namespace of leader-election configmap for ingress controller")
+	rootCmd.Flags().StringVar(&leaderElectionNamespace, "election-namespace", controller.DefaultElectionNamespace, "Namespace of leader-election configmap for ingress controller. If unspecified, the namespace of this controller pod will be used")
 
 	viper.BindPFlag("master", rootCmd.Flags().Lookup("master"))
 	viper.BindPFlag("kubeconfig", rootCmd.Flags().Lookup("kubeconfig"))
 	viper.BindPFlag("aws-region", rootCmd.Flags().Lookup("aws-region"))
+	viper.BindPFlag("election", rootCmd.Flags().Lookup("election"))
+	viper.BindPFlag("election-id", rootCmd.Flags().Lookup("election-id"))
+	viper.BindPFlag("election-namespace", rootCmd.Flags().Lookup("election-namespace"))
 }
 
 func main() {
@@ -77,7 +86,7 @@ var rootCmd = &cobra.Command{
 			klog.V(1).Infof("FLAG: --%s=%q", flag.Name, flag.Value)
 		})
 
-		var stopCh chan struct{}
+		stopCh := make(chan struct{})
 
 		cfg, err := getConfig()
 		if err != nil {
@@ -114,6 +123,9 @@ var rootCmd = &cobra.Command{
 			meshInformerFactory.Appmesh().V1beta1().VirtualNodes(),
 			meshInformerFactory.Appmesh().V1beta1().VirtualServices(),
 			stats,
+			leaderElection,
+			leaderElectionID,
+			leaderElectionNamespace,
 		)
 
 		if err != nil {

deploy/all.yaml

@@ -599,6 +599,13 @@ rules:
   - apiGroups: [""]
     resources: ["pods"]
     verbs: ["*"]
+  - apiGroups: [""]
+    resources: ["configmaps"]
+    verbs: ["create"]
+  - apiGroups: [""]
+    resources: ["configmaps"]
+    resourceNames: ["app-mesh-controller-leader"]
+    verbs: ["*"]
   - apiGroups: ["appmesh.k8s.aws"]
     resources: ["meshes", "virtualnodes", "virtualservices", "meshes/status", "virtualnodes/status", "virtualservices/status"]
     verbs: ["*"]

go.sum

@@ -55,6 +55,7 @@ github.com/google/gofuzz v0.0.0-20161122191042-44d81051d367/go.mod h1:HP5RmnzzSN
 github.com/google/gofuzz v0.0.0-20170612174753-24818f796faf/go.mod h1:HP5RmnzzSNb993RKQDq4+1A4ia9nllfqcQFTQJedwGI=
 github.com/google/gofuzz v1.0.0 h1:A8PeW59pxE9IoFRqBp37U+mSNaQoZ46F1f0f863XSXw=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gnostic v0.0.0-20170729233727-0c5108395e2d/go.mod h1:sJBsCZ4ayReDTBIg8b9dl28c5xFWyhBTVRp3pOg5EKY=
 github.com/googleapis/gnostic v0.2.0 h1:l6N3VoaVzTncYYW+9yOz2LJJammFZGBO13sqgEhpy9g=

pkg/controller/controller.go

@@ -5,6 +5,10 @@ import (
 	"fmt"
 	"time"
 
+	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+
 	appmeshv1beta1 "github.com/aws/aws-app-mesh-controller-for-k8s/pkg/apis/appmesh/v1beta1"
 	"github.com/aws/aws-app-mesh-controller-for-k8s/pkg/aws"
 	meshclientset "github.com/aws/aws-app-mesh-controller-for-k8s/pkg/client/clientset/versioned"
@@ -29,7 +33,11 @@ import (
 
 const (
 	//setting default threadiness to number of go-routines
-	DefaultThreadiness                  = 5
+	DefaultThreadiness       = 5
+	DefaultElection          = true
+	DefaultElectionID        = "app-mesh-controller-leader"
+	DefaultElectionNamespace = ""
+
 	controllerAgentName                 = "app-mesh-controller"
 	meshDeletionFinalizerName           = "meshDeletion.finalizers.appmesh.k8s.aws"
 	virtualNodeDeletionFinalizerName    = "virtualNodeDeletion.finalizers.appmesh.k8s.aws"
@@ -73,6 +81,18 @@ type Controller struct {
 
 	// stats records mesh Prometheus metrics
 	stats *metrics.Recorder
+
+	// LeaderElection determines whether or not to use leader election when
+	// starting the manager.
+	leaderElection bool
+
+	// LeaderElectionID determines the name of the configmap that leader election
+	// will use for holding the leader lock.
+	leaderElectionID string
+
+	// LeaderElectionNamespace determines the namespace in which the leader
+	// election configmap will be created.
+	leaderElectionNamespace string
 }
 
 func NewController(
@@ -83,7 +103,10 @@ func NewController(
 	meshInformer meshinformers.MeshInformer,
 	virtualNodeInformer meshinformers.VirtualNodeInformer,
 	virtualServiceInformer meshinformers.VirtualServiceInformer,
-	stats *metrics.Recorder) (*Controller, error) {
+	stats *metrics.Recorder,
+	leaderElection bool,
+	leaderElectionID string,
+	leaderElectionNamespace string) (*Controller, error) {
 
 	utilruntime.Must(meshscheme.AddToScheme(scheme.Scheme))
 	klog.V(4).Info("Creating event broadcaster")
@@ -93,24 +116,27 @@ func NewController(
 	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})
 
 	controller := &Controller{
-		name:                 controllerAgentName,
-		cloud:                cloud,
-		kubeclientset:        kubeclientset,
-		meshclientset:        meshclientset,
-		podsLister:           podInformer.Lister(),
-		podsSynced:           podInformer.Informer().HasSynced,
-		meshLister:           meshInformer.Lister(),
-		meshSynced:           meshInformer.Informer().HasSynced,
-		virtualNodeLister:    virtualNodeInformer.Lister(),
-		virtualNodeSynced:    virtualNodeInformer.Informer().HasSynced,
-		virtualServiceLister: virtualServiceInformer.Lister(),
-		virtualServiceSynced: virtualServiceInformer.Informer().HasSynced,
-		mq:                   workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
-		nq:                   workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
-		sq:                   workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
-		pq:                   workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
-		recorder:             recorder,
-		stats:                stats,
+		name:                    controllerAgentName,
+		cloud:                   cloud,
+		kubeclientset:           kubeclientset,
+		meshclientset:           meshclientset,
+		podsLister:              podInformer.Lister(),
+		podsSynced:              podInformer.Informer().HasSynced,
+		meshLister:              meshInformer.Lister(),
+		meshSynced:              meshInformer.Informer().HasSynced,
+		virtualNodeLister:       virtualNodeInformer.Lister(),
+		virtualNodeSynced:       virtualNodeInformer.Informer().HasSynced,
+		virtualServiceLister:    virtualServiceInformer.Lister(),
+		virtualServiceSynced:    virtualServiceInformer.Informer().HasSynced,
+		mq:                      workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+		nq:                      workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+		sq:                      workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+		pq:                      workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+		recorder:                recorder,
+		stats:                   stats,
+		leaderElection:          leaderElection,
+		leaderElectionID:        leaderElectionID,
+		leaderElectionNamespace: leaderElectionNamespace,
 	}
 
 	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -198,21 +224,76 @@ func (c *Controller) Run(threadiness int, stopCh chan struct{}) error {
 		return fmt.Errorf("failed to wait for caches to sync")
 	}
 
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		select {
+		case <-stopCh:
+			cancel()
+		case <-ctx.Done():
+		}
+	}()
+	if c.leaderElection {
+		return c.runWorkersWithLeaderElection(ctx, threadiness)
+	}
+	c.runWorkers(ctx, threadiness)
+	return nil
+}
+
+func (c *Controller) runWorkersWithLeaderElection(ctx context.Context, threadiness int) error {
+	leaderElectionNamespace := c.leaderElectionNamespace
+	if leaderElectionNamespace == "" {
+		var err error
+		if leaderElectionNamespace, err = getInClusterNamespace(); err != nil {
+			return err
+		}
+	}
+
+	leaderElectionIdentity := string(uuid.NewUUID())
+	leaderElectionLock, err := resourcelock.New(resourcelock.ConfigMapsResourceLock,
+		leaderElectionNamespace,
+		c.leaderElectionID,
+		c.kubeclientset.CoreV1(),
+		c.kubeclientset.CoordinationV1(),
+		resourcelock.ResourceLockConfig{
+			Identity: leaderElectionIdentity,
+		})
+	if err != nil {
+		return err
+	}
+	elector, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+		Lock:          leaderElectionLock,
+		LeaseDuration: 60 * time.Second,
+		RenewDeadline: 15 * time.Second,
+		RetryPeriod:   5 * time.Second,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				c.runWorkers(ctx, threadiness)
+			},
+			OnStoppedLeading: func() {
+				klog.Info("losing leader")
+			},
+		},
+	})
+	if err != nil {
+		return err
+	}
+	elector.Run(ctx)
+	return nil
+}
+
+func (c *Controller) runWorkers(ctx context.Context, threadiness int) {
 	klog.Info("Starting workers")
 	// Launch workers to process Mesh resources
 	for i := 0; i < threadiness; i++ {
-		go wait.Until(c.meshWorker, time.Second, stopCh)
-		go wait.Until(c.vNodeWorker, time.Second, stopCh)
-		go wait.Until(c.vServiceWorker, time.Second, stopCh)
-		go wait.Until(c.podWorker, time.Second, stopCh)
-		go wait.Until(c.cloudmapReconciler, 1*time.Minute, stopCh)
+		go wait.Until(c.meshWorker, time.Second, ctx.Done())
+		go wait.Until(c.vNodeWorker, time.Second, ctx.Done())
+		go wait.Until(c.vServiceWorker, time.Second, ctx.Done())
+		go wait.Until(c.podWorker, time.Second, ctx.Done())
+		go wait.Until(c.cloudmapReconciler, 1*time.Minute, ctx.Done())
 	}
-
 	klog.Info("Started workers")
-	<-stopCh
+	<-ctx.Done()
 	klog.Info("Shutting down workers")
-
-	return nil
 }
 
 // podAdded adds the pods endpoint to matching CloudMap Services.

pkg/controller/controller_util.go

@@ -2,10 +2,15 @@ package controller
 
 import (
 	"fmt"
-	"k8s.io/apimachinery/pkg/api/meta"
+	"io/ioutil"
+	"os"
 	"strings"
+
+	"k8s.io/apimachinery/pkg/api/meta"
 )
 
+const inClusterNamespacePath = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
+
 func containsFinalizer(obj interface{}, finalizer string) (bool, error) {
 	metaobj, err := meta.Accessor(obj)
 	if err != nil {
@@ -62,3 +67,22 @@ func namespacedResourceName(resourceName string, defaultResourceNamespace string
 	}
 	return resourceName + "-" + defaultResourceNamespace
 }
+
+// getInClusterNamespace returns the namespace of the controller pod.
+func getInClusterNamespace() (string, error) {
+	// Check whether the namespace file exists.
+	// If not, we are not running in a cluster and can't get the namespace.
+	_, err := os.Stat(inClusterNamespacePath)
+	if os.IsNotExist(err) {
+		return "", fmt.Errorf("not running in-cluster, please specify --election-namespace")
+	} else if err != nil {
+		return "", fmt.Errorf("error checking namespace file: %v", err)
+	}
+
+	// Load the namespace file and return its content
+	namespace, err := ioutil.ReadFile(inClusterNamespacePath)
+	if err != nil {
+		return "", fmt.Errorf("error reading namespace file: %v", err)
+	}
+	return string(namespace), nil
+}