From b937c4ea8d9423961033085108ce27387a288154 Mon Sep 17 00:00:00 2001 From: stefanprodan Date: Sun, 7 Jul 2019 02:21:13 +0300 Subject: [PATCH] Implement leader election Add enable-leader-election and leader-election-namespace flags --- cmd/flagger/main.go | 120 +++++++++++++++++++++++++++++++++++--------- go.sum | 1 + 2 files changed, 97 insertions(+), 24 deletions(-) diff --git a/cmd/flagger/main.go b/cmd/flagger/main.go index f7a308d0..1aabb4a8 100644 --- a/cmd/flagger/main.go +++ b/cmd/flagger/main.go @@ -1,6 +1,7 @@ package main import ( + "context" "flag" "fmt" "log" @@ -8,7 +9,7 @@ import ( "strings" "time" - semver "github.com/Masterminds/semver" + "github.com/Masterminds/semver" clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned" informers "github.com/weaveworks/flagger/pkg/client/informers/externalversions" "github.com/weaveworks/flagger/pkg/controller" @@ -20,31 +21,37 @@ import ( "github.com/weaveworks/flagger/pkg/signals" "github.com/weaveworks/flagger/pkg/version" "go.uber.org/zap" + "k8s.io/apimachinery/pkg/util/uuid" "k8s.io/client-go/kubernetes" _ "k8s.io/client-go/plugin/pkg/client/auth/gcp" "k8s.io/client-go/tools/cache" "k8s.io/client-go/tools/clientcmd" + "k8s.io/client-go/tools/leaderelection" + "k8s.io/client-go/tools/leaderelection/resourcelock" + "k8s.io/client-go/transport" _ "k8s.io/code-generator/cmd/client-gen/generators" ) var ( - masterURL string - kubeconfig string - metricsServer string - controlLoopInterval time.Duration - logLevel string - port string - msteamsURL string - slackURL string - slackUser string - slackChannel string - threadiness int - zapReplaceGlobals bool - zapEncoding string - namespace string - meshProvider string - selectorLabels string - ver bool + masterURL string + kubeconfig string + metricsServer string + controlLoopInterval time.Duration + logLevel string + port string + msteamsURL string + slackURL string + slackUser string + slackChannel string + threadiness int + zapReplaceGlobals bool + zapEncoding string + namespace string + meshProvider string + selectorLabels string + enableLeaderElection bool + leaderElectionNamespace string + ver bool ) func init() { @@ -62,8 +69,10 @@ func init() { flag.BoolVar(&zapReplaceGlobals, "zap-replace-globals", false, "Whether to change the logging level of the global zap logger.") flag.StringVar(&zapEncoding, "zap-encoding", "json", "Zap logger encoding.") flag.StringVar(&namespace, "namespace", "", "Namespace that flagger would watch canary object.") - flag.StringVar(&meshProvider, "mesh-provider", "istio", "Service mesh provider, can be istio, appmesh, supergloo, nginx or smi.") + flag.StringVar(&meshProvider, "mesh-provider", "istio", "Service mesh provider, can be istio, linkerd, appmesh, supergloo, nginx or smi.") flag.StringVar(&selectorLabels, "selector-labels", "app,name,app.kubernetes.io/name", "List of pod labels that Flagger uses to create pod selectors.") + flag.BoolVar(&enableLeaderElection, "enable-leader-election", false, "Enable leader election.") + flag.StringVar(&leaderElectionNamespace, "leader-election-namespace", "kube-system", "Namespace used to create the leader election config map.") flag.BoolVar(&ver, "version", false, "Print version") } @@ -194,14 +203,77 @@ func main() { } } - // start controller - go func(ctrl *controller.Controller) { - if err := ctrl.Run(threadiness, stopCh); err != nil { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + cfg.Wrap(transport.ContextCanceller(ctx, fmt.Errorf("the leader is shutting down"))) + + go func() { + <-stopCh + cancel() + }() + + runController := func() { + if err := c.Run(threadiness, stopCh); err != nil { logger.Fatalf("Error running controller: %v", err) } - }(c) + } - <-stopCh + if enableLeaderElection { + ns := leaderElectionNamespace + if namespace != "" { + ns = namespace + } + startLeaderElection(ctx, runController, ns, kubeClient, logger) + } else { + runController() + } +} + +func startLeaderElection(ctx context.Context, run func(), ns string, kubeClient kubernetes.Interface, logger *zap.SugaredLogger) { + configMapName := "flagger-leader-election" + id, err := os.Hostname() + if err != nil { + logger.Fatalf("Error running controller: %v", err) + } + id = id + "_" + string(uuid.NewUUID()) + + lock, err := resourcelock.New( + resourcelock.ConfigMapsResourceLock, + ns, + configMapName, + kubeClient.CoreV1(), + kubeClient.CoordinationV1(), + resourcelock.ResourceLockConfig{ + Identity: id, + }, + ) + if err != nil { + logger.Fatalf("Error running controller: %v", err) + } + + logger.Infof("Starting leader election id: %s configmap: %s namespace: %s", id, configMapName, ns) + leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{ + Lock: lock, + ReleaseOnCancel: true, + LeaseDuration: 60 * time.Second, + RenewDeadline: 15 * time.Second, + RetryPeriod: 5 * time.Second, + Callbacks: leaderelection.LeaderCallbacks{ + OnStartedLeading: func(ctx context.Context) { + logger.Info("Acting as elected leader") + run() + }, + OnStoppedLeading: func() { + logger.Infof("Leader election lost") + }, + OnNewLeader: func(identity string) { + if identity != id { + logger.Infof("Another instance has been elected as leader: %v", identity) + } + }, + }, + }) } func initNotifier(logger *zap.SugaredLogger) (client notifier.Interface) { diff --git a/go.sum b/go.sum index 3eb69fc9..b000eb21 100644 --- a/go.sum +++ b/go.sum @@ -154,6 +154,7 @@ 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/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs= github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc= +github.com/google/uuid v1.0.0 h1:b4Gk+7WdP/d3HZH8EJsZpvV7EtDOgaZLtnaNGIu1adA= github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg= github.com/googleapis/gnostic v0.0.0-20170426233943-68f4ded48ba9/go.mod h1:sJBsCZ4ayReDTBIg8b9dl28c5xFWyhBTVRp3pOg5EKY=