From 15a49a5f4577c958f988c2d676ba3e59e5bfe104 Mon Sep 17 00:00:00 2001 From: Aaron Liang Date: Fri, 24 Mar 2023 16:54:26 +0000 Subject: [PATCH] Add feature flag for controller leader election --- cmd/allocator/main.go | 2 +- cmd/controller/main.go | 87 +++- cmd/extensions/pprof.go | 1 + cmd/ping/main.go | 2 +- cmd/sdk-server/main.go | 2 +- install/helm/agones/defaultfeaturegates.yaml | 1 + install/helm/agones/templates/controller.yaml | 29 +- .../templates/serviceaccounts/controller.yaml | 3 + install/helm/agones/values.yaml | 1 + pkg/util/runtime/features.go | 8 +- pkg/util/signals/signals.go | 7 +- .../client-go/tools/leaderelection/OWNERS | 11 + .../tools/leaderelection/healthzadaptor.go | 69 +++ .../tools/leaderelection/leaderelection.go | 418 ++++++++++++++++++ .../client-go/tools/leaderelection/metrics.go | 109 +++++ .../resourcelock/configmaplock.go | 126 ++++++ .../resourcelock/endpointslock.go | 121 +++++ .../leaderelection/resourcelock/interface.go | 227 ++++++++++ .../leaderelection/resourcelock/leaselock.go | 139 ++++++ .../leaderelection/resourcelock/multilock.go | 104 +++++ 20 files changed, 1446 insertions(+), 21 deletions(-) create mode 100644 vendor/k8s.io/client-go/tools/leaderelection/OWNERS create mode 100644 vendor/k8s.io/client-go/tools/leaderelection/healthzadaptor.go create mode 100644 vendor/k8s.io/client-go/tools/leaderelection/leaderelection.go create mode 100644 vendor/k8s.io/client-go/tools/leaderelection/metrics.go create mode 100644 vendor/k8s.io/client-go/tools/leaderelection/resourcelock/configmaplock.go create mode 100644 vendor/k8s.io/client-go/tools/leaderelection/resourcelock/endpointslock.go create mode 100644 vendor/k8s.io/client-go/tools/leaderelection/resourcelock/interface.go create mode 100644 vendor/k8s.io/client-go/tools/leaderelection/resourcelock/leaselock.go create mode 100644 vendor/k8s.io/client-go/tools/leaderelection/resourcelock/multilock.go diff --git a/cmd/allocator/main.go b/cmd/allocator/main.go index 4d7b825a1e..4f18c3ee5e 100644 --- a/cmd/allocator/main.go +++ b/cmd/allocator/main.go @@ -369,7 +369,7 @@ func newServiceHandler(kubeClient kubernetes.Interface, agonesClient versioned.I totalRemoteAllocationTimeout, allocationBatchWaitTime) - ctx := signals.NewSigKillContext() + ctx, _ := signals.NewSigKillContext() h := serviceHandler{ allocationCallback: func(gsa *allocationv1.GameServerAllocation) (k8sruntime.Object, error) { return allocator.Allocate(ctx, gsa) diff --git a/cmd/controller/main.go b/cmd/controller/main.go index 12719c6878..dda7a47424 100644 --- a/cmd/controller/main.go +++ b/cmd/controller/main.go @@ -40,6 +40,7 @@ import ( "agones.dev/agones/pkg/util/runtime" "agones.dev/agones/pkg/util/signals" "agones.dev/agones/pkg/util/webhooks" + "github.com/google/uuid" "github.com/heptiolabs/healthcheck" "github.com/pkg/errors" prom "github.com/prometheus/client_golang/prometheus" @@ -50,9 +51,12 @@ import ( corev1 "k8s.io/api/core/v1" extclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/clientcmd" + "k8s.io/client-go/tools/leaderelection" + "k8s.io/client-go/tools/leaderelection/resourcelock" ) const ( @@ -106,7 +110,7 @@ func setupLogging(logDir string, logSizeLimitMB int) { // main starts the operator for the gameserver CRD func main() { - ctx := signals.NewSigKillContext() + ctx, cancel := signals.NewSigKillContext() ctlConf := parseEnvFlags() if ctlConf.LogDir != "" { @@ -234,19 +238,59 @@ func main() { rs = append(rs, gasController) } - kubeInformerFactory.Start(ctx.Done()) - agonesInformerFactory.Start(ctx.Done()) + // Start of the leader election code + id := uuid.New().String() + + lock := &resourcelock.LeaseLock{ + LeaseMeta: metav1.ObjectMeta{ + Name: "agones-controller", + Namespace: "agones-system", + }, + Client: kubeClient.CoordinationV1(), + LockConfig: resourcelock.ResourceLockConfig{ + Identity: id, + }, + } + + logger.Infof("This controller ID is %s", id) + + startLeading := func(ctx context.Context) { + logger.Infof("Leader elected: %s", id) + + kubeInformerFactory.Start(ctx.Done()) + agonesInformerFactory.Start(ctx.Done()) + + for _, r := range rs { + go func(rr runner) { + if runErr := rr.Run(ctx, ctlConf.NumWorkers); runErr != nil { + logger.WithError(runErr).Fatalf("could not start runner: %T", rr) + } + }(r) + } - for _, r := range rs { - go func(rr runner) { - if runErr := rr.Run(ctx, ctlConf.NumWorkers); runErr != nil { - logger.WithError(runErr).Fatalf("could not start runner: %T", rr) - } - }(r) + <-ctx.Done() + logger.Info("Shut down agones controllers") } - <-ctx.Done() - logger.Info("Shut down agones controllers") + if !runtime.FeatureEnabled(runtime.FeatureControllerLeaderElection) { + startLeading(ctx) + } else { + runLeaderElection(ctx, lock, + startLeading, + func() { + logger.Infof("leader lost: %s", id) + time.Sleep(10 * time.Second) + cancel() + os.Exit(0) + }, + func(identity string) { + if identity == id { + return + } + logger.Infof("new leader elected: %s", identity) + }, + ) + } } func parseEnvFlags() config { @@ -433,6 +477,27 @@ type httpServer struct { http.ServeMux } +func runLeaderElection(ctx context.Context, lock *resourcelock.LeaseLock, startLeading func(_ context.Context), stopLeading func(), newLeader func(identity string)) { + leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{ + Lock: lock, + // IMPORTANT: you MUST ensure that any code you have that + // is protected by the lease must terminate **before** + // you call cancel. Otherwise, you could have a background + // loop still running and another process could + // get elected before your background loop finished, violating + // the stated goal of the lease. + ReleaseOnCancel: true, + LeaseDuration: 15 * time.Second, + RenewDeadline: 10 * time.Second, + RetryPeriod: 2 * time.Second, + Callbacks: leaderelection.LeaderCallbacks{ + OnStartedLeading: startLeading, + OnStoppedLeading: stopLeading, + OnNewLeader: newLeader, + }, + }) +} + func (h *httpServer) Run(_ context.Context, _ int) error { logger.Info("Starting http server...") srv := &http.Server{ diff --git a/cmd/extensions/pprof.go b/cmd/extensions/pprof.go index 6c33f94d90..6c1bec83c8 100644 --- a/cmd/extensions/pprof.go +++ b/cmd/extensions/pprof.go @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +//go:build profile // +build profile package main diff --git a/cmd/ping/main.go b/cmd/ping/main.go index 7bcd4d95fd..3eea91961c 100644 --- a/cmd/ping/main.go +++ b/cmd/ping/main.go @@ -49,7 +49,7 @@ func main() { logger.WithField("version", pkg.Version).WithField("featureGates", runtime.EncodeFeatures()). WithField("ctlConf", ctlConf).Info("Starting ping...") - ctx := signals.NewSigKillContext() + ctx, _ := signals.NewSigKillContext() udpSrv := serveUDP(ctx, ctlConf) defer udpSrv.close() diff --git a/cmd/sdk-server/main.go b/cmd/sdk-server/main.go index 17005a8577..9a54646dbd 100644 --- a/cmd/sdk-server/main.go +++ b/cmd/sdk-server/main.go @@ -78,7 +78,7 @@ func main() { time.Sleep(time.Duration(ctlConf.Delay) * time.Second) } - ctx := signals.NewSigKillContext() + ctx, _ := signals.NewSigKillContext() grpcServer := grpc.NewServer() // don't graceful stop, because if we get a SIGKILL signal diff --git a/install/helm/agones/defaultfeaturegates.yaml b/install/helm/agones/defaultfeaturegates.yaml index 98823e0e40..3ee5b5b210 100644 --- a/install/helm/agones/defaultfeaturegates.yaml +++ b/install/helm/agones/defaultfeaturegates.yaml @@ -30,6 +30,7 @@ SplitControllerAndExtensions: false # Pre-Alpha features CountsAndLists: false +ControllerLeaderElection: false # Example feature Example: false diff --git a/install/helm/agones/templates/controller.yaml b/install/helm/agones/templates/controller.yaml index cf0821e065..5d147dc8fa 100644 --- a/install/helm/agones/templates/controller.yaml +++ b/install/helm/agones/templates/controller.yaml @@ -31,13 +31,19 @@ spec: app: {{ template "agones.name" . }} release: {{ .Release.Name }} heritage: {{ .Release.Service }} - replicas: 1 +{{- if and $featureGates.ControllerLeaderElection (lt .Values.agones.controller.replicas 2) }} +{{- fail "Cannot have less than 2 replicas when leader election is enabled!" }} +{{- end}} +{{- if and not $featureGates.ControllerLeaderElection (ge .Values.agones.controller.replicas 2) }} +{{- fail "Cannot have leader election enabled when replicas is greater than 1!" }} +{{- end}} + replicas: {{ .Values.agones.controller.replicas | quote }} strategy: type: Recreate template: metadata: annotations: -{{- if not $featureGates.SplitControllerAndExtensions }} +{{- if or not $featureGates.SplitControllerAndExtensions not $featureGates.ControllerLeaderElection}} cluster-autoscaler.kubernetes.io/safe-to-evict: {{ .Values.agones.controller.safeToEvict | quote }} {{- end }} {{- if .Values.agones.controller.generateTLS }} @@ -183,3 +189,22 @@ spec: imagePullSecrets: - name: {{.Values.agones.image.controller.pullSecret}} {{- end }} +--- +apiVersion: policy/v1 +kind: PodDisruptionBudget +metadata: + name: agones-controller-pdb +spec: +{{- if .Values.agones.extensions.pdb.minAvailable }} +{{- if .Values.agones.extensions.pdb.maxUnavailable }} +{{- fail "minAvailable and maxUnavailable are mutually exclusive!" }} +{{- end}} +{{- end}} + minAvailable: {{ .Values.agones.extensions.pdb.minAvailable }} + maxUnavailable: {{ .Values.agones.extensions.pdb.maxUnavailable }} + selector: + matchLabels: + agones.dev/role: controller + app: {{ template "agones.name" . }} + release: {{ .Release.Name }} + heritage: {{ .Release.Service }} diff --git a/install/helm/agones/templates/serviceaccounts/controller.yaml b/install/helm/agones/templates/serviceaccounts/controller.yaml index a9a4b4b9c9..9389c71405 100644 --- a/install/helm/agones/templates/serviceaccounts/controller.yaml +++ b/install/helm/agones/templates/serviceaccounts/controller.yaml @@ -81,6 +81,9 @@ rules: - apiGroups: ["autoscaling.agones.dev"] resources: ["fleetautoscalers/status"] verbs: ["update"] +- apiGroups: ["coordination.k8s.io"] + resources: ["leases"] + verbs: ["create", "delete", "get", "list", "update", "watch"] --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding diff --git a/install/helm/agones/values.yaml b/install/helm/agones/values.yaml index dd60e0e872..57044e1e04 100644 --- a/install/helm/agones/values.yaml +++ b/install/helm/agones/values.yaml @@ -94,6 +94,7 @@ agones: failureThreshold: 3 timeoutSeconds: 1 allocationBatchWaitTime: 500ms + replicas: 2 extensions: <<: *controllerValues pdb: diff --git a/pkg/util/runtime/features.go b/pkg/util/runtime/features.go index 7a5d09baef..e6d14b4359 100644 --- a/pkg/util/runtime/features.go +++ b/pkg/util/runtime/features.go @@ -74,6 +74,9 @@ const ( // (a generic implenetation of the player tracking feature). FeatureCountsAndLists Feature = "CountsAndLists" + // FeatureControllerLeaderElection is a feature flag that enables/disables leader election for controller + FeatureControllerLeaderElection = "ControllerLeaderElection" + //////////////// // Example feature @@ -122,8 +125,9 @@ var ( FeatureSplitControllerAndExtensions: false, // Pre-Alpha features - FeatureCountsAndLists: false, - FeatureFleetAllocateOverflow: false, + FeatureCountsAndLists: false, + FeatureFleetAllocateOverflow: false, + FeatureControllerLeaderElection: false, // Example feature FeatureExample: false, diff --git a/pkg/util/signals/signals.go b/pkg/util/signals/signals.go index 8e474d0d5d..c23a736fa8 100644 --- a/pkg/util/signals/signals.go +++ b/pkg/util/signals/signals.go @@ -24,9 +24,10 @@ import ( ) // NewSigKillContext returns a Context that cancels when os.Interrupt or os.Kill is received -func NewSigKillContext() context.Context { - ctx, _ := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) - return ctx +func NewSigKillContext() (context.Context, context.CancelFunc) { + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + + return ctx, cancel } // NewSigTermHandler creates a channel to listen to SIGTERM and runs the handle function diff --git a/vendor/k8s.io/client-go/tools/leaderelection/OWNERS b/vendor/k8s.io/client-go/tools/leaderelection/OWNERS new file mode 100644 index 0000000000..908bdacdfe --- /dev/null +++ b/vendor/k8s.io/client-go/tools/leaderelection/OWNERS @@ -0,0 +1,11 @@ +# See the OWNERS docs at https://go.k8s.io/owners + +approvers: + - mikedanese +reviewers: + - wojtek-t + - deads2k + - mikedanese + - ingvagabund +emeritus_approvers: + - timothysc diff --git a/vendor/k8s.io/client-go/tools/leaderelection/healthzadaptor.go b/vendor/k8s.io/client-go/tools/leaderelection/healthzadaptor.go new file mode 100644 index 0000000000..b935372919 --- /dev/null +++ b/vendor/k8s.io/client-go/tools/leaderelection/healthzadaptor.go @@ -0,0 +1,69 @@ +/* +Copyright 2015 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package leaderelection + +import ( + "net/http" + "sync" + "time" +) + +// HealthzAdaptor associates the /healthz endpoint with the LeaderElection object. +// It helps deal with the /healthz endpoint being set up prior to the LeaderElection. +// This contains the code needed to act as an adaptor between the leader +// election code the health check code. It allows us to provide health +// status about the leader election. Most specifically about if the leader +// has failed to renew without exiting the process. In that case we should +// report not healthy and rely on the kubelet to take down the process. +type HealthzAdaptor struct { + pointerLock sync.Mutex + le *LeaderElector + timeout time.Duration +} + +// Name returns the name of the health check we are implementing. +func (l *HealthzAdaptor) Name() string { + return "leaderElection" +} + +// Check is called by the healthz endpoint handler. +// It fails (returns an error) if we own the lease but had not been able to renew it. +func (l *HealthzAdaptor) Check(req *http.Request) error { + l.pointerLock.Lock() + defer l.pointerLock.Unlock() + if l.le == nil { + return nil + } + return l.le.Check(l.timeout) +} + +// SetLeaderElection ties a leader election object to a HealthzAdaptor +func (l *HealthzAdaptor) SetLeaderElection(le *LeaderElector) { + l.pointerLock.Lock() + defer l.pointerLock.Unlock() + l.le = le +} + +// NewLeaderHealthzAdaptor creates a basic healthz adaptor to monitor a leader election. +// timeout determines the time beyond the lease expiry to be allowed for timeout. +// checks within the timeout period after the lease expires will still return healthy. +func NewLeaderHealthzAdaptor(timeout time.Duration) *HealthzAdaptor { + result := &HealthzAdaptor{ + timeout: timeout, + } + return result +} diff --git a/vendor/k8s.io/client-go/tools/leaderelection/leaderelection.go b/vendor/k8s.io/client-go/tools/leaderelection/leaderelection.go new file mode 100644 index 0000000000..03a13e6b63 --- /dev/null +++ b/vendor/k8s.io/client-go/tools/leaderelection/leaderelection.go @@ -0,0 +1,418 @@ +/* +Copyright 2015 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package leaderelection implements leader election of a set of endpoints. +// It uses an annotation in the endpoints object to store the record of the +// election state. This implementation does not guarantee that only one +// client is acting as a leader (a.k.a. fencing). +// +// A client only acts on timestamps captured locally to infer the state of the +// leader election. The client does not consider timestamps in the leader +// election record to be accurate because these timestamps may not have been +// produced by a local clock. The implemention does not depend on their +// accuracy and only uses their change to indicate that another client has +// renewed the leader lease. Thus the implementation is tolerant to arbitrary +// clock skew, but is not tolerant to arbitrary clock skew rate. +// +// However the level of tolerance to skew rate can be configured by setting +// RenewDeadline and LeaseDuration appropriately. The tolerance expressed as a +// maximum tolerated ratio of time passed on the fastest node to time passed on +// the slowest node can be approximately achieved with a configuration that sets +// the same ratio of LeaseDuration to RenewDeadline. For example if a user wanted +// to tolerate some nodes progressing forward in time twice as fast as other nodes, +// the user could set LeaseDuration to 60 seconds and RenewDeadline to 30 seconds. +// +// While not required, some method of clock synchronization between nodes in the +// cluster is highly recommended. It's important to keep in mind when configuring +// this client that the tolerance to skew rate varies inversely to master +// availability. +// +// Larger clusters often have a more lenient SLA for API latency. This should be +// taken into account when configuring the client. The rate of leader transitions +// should be monitored and RetryPeriod and LeaseDuration should be increased +// until the rate is stable and acceptably low. It's important to keep in mind +// when configuring this client that the tolerance to API latency varies inversely +// to master availability. +// +// DISCLAIMER: this is an alpha API. This library will likely change significantly +// or even be removed entirely in subsequent releases. Depend on this API at +// your own risk. +package leaderelection + +import ( + "bytes" + "context" + "fmt" + "sync" + "time" + + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/wait" + rl "k8s.io/client-go/tools/leaderelection/resourcelock" + "k8s.io/utils/clock" + + "k8s.io/klog/v2" +) + +const ( + JitterFactor = 1.2 +) + +// NewLeaderElector creates a LeaderElector from a LeaderElectionConfig +func NewLeaderElector(lec LeaderElectionConfig) (*LeaderElector, error) { + if lec.LeaseDuration <= lec.RenewDeadline { + return nil, fmt.Errorf("leaseDuration must be greater than renewDeadline") + } + if lec.RenewDeadline <= time.Duration(JitterFactor*float64(lec.RetryPeriod)) { + return nil, fmt.Errorf("renewDeadline must be greater than retryPeriod*JitterFactor") + } + if lec.LeaseDuration < 1 { + return nil, fmt.Errorf("leaseDuration must be greater than zero") + } + if lec.RenewDeadline < 1 { + return nil, fmt.Errorf("renewDeadline must be greater than zero") + } + if lec.RetryPeriod < 1 { + return nil, fmt.Errorf("retryPeriod must be greater than zero") + } + if lec.Callbacks.OnStartedLeading == nil { + return nil, fmt.Errorf("OnStartedLeading callback must not be nil") + } + if lec.Callbacks.OnStoppedLeading == nil { + return nil, fmt.Errorf("OnStoppedLeading callback must not be nil") + } + + if lec.Lock == nil { + return nil, fmt.Errorf("Lock must not be nil.") + } + le := LeaderElector{ + config: lec, + clock: clock.RealClock{}, + metrics: globalMetricsFactory.newLeaderMetrics(), + } + le.metrics.leaderOff(le.config.Name) + return &le, nil +} + +type LeaderElectionConfig struct { + // Lock is the resource that will be used for locking + Lock rl.Interface + + // LeaseDuration is the duration that non-leader candidates will + // wait to force acquire leadership. This is measured against time of + // last observed ack. + // + // A client needs to wait a full LeaseDuration without observing a change to + // the record before it can attempt to take over. When all clients are + // shutdown and a new set of clients are started with different names against + // the same leader record, they must wait the full LeaseDuration before + // attempting to acquire the lease. Thus LeaseDuration should be as short as + // possible (within your tolerance for clock skew rate) to avoid a possible + // long waits in the scenario. + // + // Core clients default this value to 15 seconds. + LeaseDuration time.Duration + // RenewDeadline is the duration that the acting master will retry + // refreshing leadership before giving up. + // + // Core clients default this value to 10 seconds. + RenewDeadline time.Duration + // RetryPeriod is the duration the LeaderElector clients should wait + // between tries of actions. + // + // Core clients default this value to 2 seconds. + RetryPeriod time.Duration + + // Callbacks are callbacks that are triggered during certain lifecycle + // events of the LeaderElector + Callbacks LeaderCallbacks + + // WatchDog is the associated health checker + // WatchDog may be null if it's not needed/configured. + WatchDog *HealthzAdaptor + + // ReleaseOnCancel should be set true if the lock should be released + // when the run context is cancelled. If you set this to true, you must + // ensure all code guarded by this lease has successfully completed + // prior to cancelling the context, or you may have two processes + // simultaneously acting on the critical path. + ReleaseOnCancel bool + + // Name is the name of the resource lock for debugging + Name string +} + +// LeaderCallbacks are callbacks that are triggered during certain +// lifecycle events of the LeaderElector. These are invoked asynchronously. +// +// possible future callbacks: +// * OnChallenge() +type LeaderCallbacks struct { + // OnStartedLeading is called when a LeaderElector client starts leading + OnStartedLeading func(context.Context) + // OnStoppedLeading is called when a LeaderElector client stops leading + OnStoppedLeading func() + // OnNewLeader is called when the client observes a leader that is + // not the previously observed leader. This includes the first observed + // leader when the client starts. + OnNewLeader func(identity string) +} + +// LeaderElector is a leader election client. +type LeaderElector struct { + config LeaderElectionConfig + // internal bookkeeping + observedRecord rl.LeaderElectionRecord + observedRawRecord []byte + observedTime time.Time + // used to implement OnNewLeader(), may lag slightly from the + // value observedRecord.HolderIdentity if the transition has + // not yet been reported. + reportedLeader string + + // clock is wrapper around time to allow for less flaky testing + clock clock.Clock + + // used to lock the observedRecord + observedRecordLock sync.Mutex + + metrics leaderMetricsAdapter +} + +// Run starts the leader election loop. Run will not return +// before leader election loop is stopped by ctx or it has +// stopped holding the leader lease +func (le *LeaderElector) Run(ctx context.Context) { + defer runtime.HandleCrash() + defer func() { + le.config.Callbacks.OnStoppedLeading() + }() + + if !le.acquire(ctx) { + return // ctx signalled done + } + ctx, cancel := context.WithCancel(ctx) + defer cancel() + go le.config.Callbacks.OnStartedLeading(ctx) + le.renew(ctx) +} + +// RunOrDie starts a client with the provided config or panics if the config +// fails to validate. RunOrDie blocks until leader election loop is +// stopped by ctx or it has stopped holding the leader lease +func RunOrDie(ctx context.Context, lec LeaderElectionConfig) { + le, err := NewLeaderElector(lec) + if err != nil { + panic(err) + } + if lec.WatchDog != nil { + lec.WatchDog.SetLeaderElection(le) + } + le.Run(ctx) +} + +// GetLeader returns the identity of the last observed leader or returns the empty string if +// no leader has yet been observed. +// This function is for informational purposes. (e.g. monitoring, logs, etc.) +func (le *LeaderElector) GetLeader() string { + return le.getObservedRecord().HolderIdentity +} + +// IsLeader returns true if the last observed leader was this client else returns false. +func (le *LeaderElector) IsLeader() bool { + return le.getObservedRecord().HolderIdentity == le.config.Lock.Identity() +} + +// acquire loops calling tryAcquireOrRenew and returns true immediately when tryAcquireOrRenew succeeds. +// Returns false if ctx signals done. +func (le *LeaderElector) acquire(ctx context.Context) bool { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + succeeded := false + desc := le.config.Lock.Describe() + klog.Infof("attempting to acquire leader lease %v...", desc) + wait.JitterUntil(func() { + succeeded = le.tryAcquireOrRenew(ctx) + le.maybeReportTransition() + if !succeeded { + klog.V(4).Infof("failed to acquire lease %v", desc) + return + } + le.config.Lock.RecordEvent("became leader") + le.metrics.leaderOn(le.config.Name) + klog.Infof("successfully acquired lease %v", desc) + cancel() + }, le.config.RetryPeriod, JitterFactor, true, ctx.Done()) + return succeeded +} + +// renew loops calling tryAcquireOrRenew and returns immediately when tryAcquireOrRenew fails or ctx signals done. +func (le *LeaderElector) renew(ctx context.Context) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + wait.Until(func() { + timeoutCtx, timeoutCancel := context.WithTimeout(ctx, le.config.RenewDeadline) + defer timeoutCancel() + err := wait.PollImmediateUntil(le.config.RetryPeriod, func() (bool, error) { + return le.tryAcquireOrRenew(timeoutCtx), nil + }, timeoutCtx.Done()) + + le.maybeReportTransition() + desc := le.config.Lock.Describe() + if err == nil { + klog.V(5).Infof("successfully renewed lease %v", desc) + return + } + le.config.Lock.RecordEvent("stopped leading") + le.metrics.leaderOff(le.config.Name) + klog.Infof("failed to renew lease %v: %v", desc, err) + cancel() + }, le.config.RetryPeriod, ctx.Done()) + + // if we hold the lease, give it up + if le.config.ReleaseOnCancel { + le.release() + } +} + +// release attempts to release the leader lease if we have acquired it. +func (le *LeaderElector) release() bool { + if !le.IsLeader() { + return true + } + now := metav1.Now() + leaderElectionRecord := rl.LeaderElectionRecord{ + LeaderTransitions: le.observedRecord.LeaderTransitions, + LeaseDurationSeconds: 1, + RenewTime: now, + AcquireTime: now, + } + if err := le.config.Lock.Update(context.TODO(), leaderElectionRecord); err != nil { + klog.Errorf("Failed to release lock: %v", err) + return false + } + + le.setObservedRecord(&leaderElectionRecord) + return true +} + +// tryAcquireOrRenew tries to acquire a leader lease if it is not already acquired, +// else it tries to renew the lease if it has already been acquired. Returns true +// on success else returns false. +func (le *LeaderElector) tryAcquireOrRenew(ctx context.Context) bool { + now := metav1.Now() + leaderElectionRecord := rl.LeaderElectionRecord{ + HolderIdentity: le.config.Lock.Identity(), + LeaseDurationSeconds: int(le.config.LeaseDuration / time.Second), + RenewTime: now, + AcquireTime: now, + } + + // 1. obtain or create the ElectionRecord + oldLeaderElectionRecord, oldLeaderElectionRawRecord, err := le.config.Lock.Get(ctx) + if err != nil { + if !errors.IsNotFound(err) { + klog.Errorf("error retrieving resource lock %v: %v", le.config.Lock.Describe(), err) + return false + } + if err = le.config.Lock.Create(ctx, leaderElectionRecord); err != nil { + klog.Errorf("error initially creating leader election record: %v", err) + return false + } + + le.setObservedRecord(&leaderElectionRecord) + + return true + } + + // 2. Record obtained, check the Identity & Time + if !bytes.Equal(le.observedRawRecord, oldLeaderElectionRawRecord) { + le.setObservedRecord(oldLeaderElectionRecord) + + le.observedRawRecord = oldLeaderElectionRawRecord + } + if len(oldLeaderElectionRecord.HolderIdentity) > 0 && + le.observedTime.Add(le.config.LeaseDuration).After(now.Time) && + !le.IsLeader() { + klog.V(4).Infof("lock is held by %v and has not yet expired", oldLeaderElectionRecord.HolderIdentity) + return false + } + + // 3. We're going to try to update. The leaderElectionRecord is set to it's default + // here. Let's correct it before updating. + if le.IsLeader() { + leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTime + leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + } else { + leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + 1 + } + + // update the lock itself + if err = le.config.Lock.Update(ctx, leaderElectionRecord); err != nil { + klog.Errorf("Failed to update lock: %v", err) + return false + } + + le.setObservedRecord(&leaderElectionRecord) + return true +} + +func (le *LeaderElector) maybeReportTransition() { + if le.observedRecord.HolderIdentity == le.reportedLeader { + return + } + le.reportedLeader = le.observedRecord.HolderIdentity + if le.config.Callbacks.OnNewLeader != nil { + go le.config.Callbacks.OnNewLeader(le.reportedLeader) + } +} + +// Check will determine if the current lease is expired by more than timeout. +func (le *LeaderElector) Check(maxTolerableExpiredLease time.Duration) error { + if !le.IsLeader() { + // Currently not concerned with the case that we are hot standby + return nil + } + // If we are more than timeout seconds after the lease duration that is past the timeout + // on the lease renew. Time to start reporting ourselves as unhealthy. We should have + // died but conditions like deadlock can prevent this. (See #70819) + if le.clock.Since(le.observedTime) > le.config.LeaseDuration+maxTolerableExpiredLease { + return fmt.Errorf("failed election to renew leadership on lease %s", le.config.Name) + } + + return nil +} + +// setObservedRecord will set a new observedRecord and update observedTime to the current time. +// Protect critical sections with lock. +func (le *LeaderElector) setObservedRecord(observedRecord *rl.LeaderElectionRecord) { + le.observedRecordLock.Lock() + defer le.observedRecordLock.Unlock() + + le.observedRecord = *observedRecord + le.observedTime = le.clock.Now() +} + +// getObservedRecord returns observersRecord. +// Protect critical sections with lock. +func (le *LeaderElector) getObservedRecord() rl.LeaderElectionRecord { + le.observedRecordLock.Lock() + defer le.observedRecordLock.Unlock() + + return le.observedRecord +} diff --git a/vendor/k8s.io/client-go/tools/leaderelection/metrics.go b/vendor/k8s.io/client-go/tools/leaderelection/metrics.go new file mode 100644 index 0000000000..65917bf88e --- /dev/null +++ b/vendor/k8s.io/client-go/tools/leaderelection/metrics.go @@ -0,0 +1,109 @@ +/* +Copyright 2018 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package leaderelection + +import ( + "sync" +) + +// This file provides abstractions for setting the provider (e.g., prometheus) +// of metrics. + +type leaderMetricsAdapter interface { + leaderOn(name string) + leaderOff(name string) +} + +// GaugeMetric represents a single numerical value that can arbitrarily go up +// and down. +type SwitchMetric interface { + On(name string) + Off(name string) +} + +type noopMetric struct{} + +func (noopMetric) On(name string) {} +func (noopMetric) Off(name string) {} + +// defaultLeaderMetrics expects the caller to lock before setting any metrics. +type defaultLeaderMetrics struct { + // leader's value indicates if the current process is the owner of name lease + leader SwitchMetric +} + +func (m *defaultLeaderMetrics) leaderOn(name string) { + if m == nil { + return + } + m.leader.On(name) +} + +func (m *defaultLeaderMetrics) leaderOff(name string) { + if m == nil { + return + } + m.leader.Off(name) +} + +type noMetrics struct{} + +func (noMetrics) leaderOn(name string) {} +func (noMetrics) leaderOff(name string) {} + +// MetricsProvider generates various metrics used by the leader election. +type MetricsProvider interface { + NewLeaderMetric() SwitchMetric +} + +type noopMetricsProvider struct{} + +func (_ noopMetricsProvider) NewLeaderMetric() SwitchMetric { + return noopMetric{} +} + +var globalMetricsFactory = leaderMetricsFactory{ + metricsProvider: noopMetricsProvider{}, +} + +type leaderMetricsFactory struct { + metricsProvider MetricsProvider + + onlyOnce sync.Once +} + +func (f *leaderMetricsFactory) setProvider(mp MetricsProvider) { + f.onlyOnce.Do(func() { + f.metricsProvider = mp + }) +} + +func (f *leaderMetricsFactory) newLeaderMetrics() leaderMetricsAdapter { + mp := f.metricsProvider + if mp == (noopMetricsProvider{}) { + return noMetrics{} + } + return &defaultLeaderMetrics{ + leader: mp.NewLeaderMetric(), + } +} + +// SetProvider sets the metrics provider for all subsequently created work +// queues. Only the first call has an effect. +func SetProvider(metricsProvider MetricsProvider) { + globalMetricsFactory.setProvider(metricsProvider) +} diff --git a/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/configmaplock.go b/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/configmaplock.go new file mode 100644 index 0000000000..5702728982 --- /dev/null +++ b/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/configmaplock.go @@ -0,0 +1,126 @@ +/* +Copyright 2017 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcelock + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + corev1client "k8s.io/client-go/kubernetes/typed/core/v1" +) + +// TODO: This is almost a exact replica of Endpoints lock. +// going forwards as we self host more and more components +// and use ConfigMaps as the means to pass that configuration +// data we will likely move to deprecate the Endpoints lock. + +type configMapLock struct { + // ConfigMapMeta should contain a Name and a Namespace of a + // ConfigMapMeta object that the LeaderElector will attempt to lead. + ConfigMapMeta metav1.ObjectMeta + Client corev1client.ConfigMapsGetter + LockConfig ResourceLockConfig + cm *v1.ConfigMap +} + +// Get returns the election record from a ConfigMap Annotation +func (cml *configMapLock) Get(ctx context.Context) (*LeaderElectionRecord, []byte, error) { + var record LeaderElectionRecord + var err error + cml.cm, err = cml.Client.ConfigMaps(cml.ConfigMapMeta.Namespace).Get(ctx, cml.ConfigMapMeta.Name, metav1.GetOptions{}) + if err != nil { + return nil, nil, err + } + if cml.cm.Annotations == nil { + cml.cm.Annotations = make(map[string]string) + } + recordStr, found := cml.cm.Annotations[LeaderElectionRecordAnnotationKey] + recordBytes := []byte(recordStr) + if found { + if err := json.Unmarshal(recordBytes, &record); err != nil { + return nil, nil, err + } + } + return &record, recordBytes, nil +} + +// Create attempts to create a LeaderElectionRecord annotation +func (cml *configMapLock) Create(ctx context.Context, ler LeaderElectionRecord) error { + recordBytes, err := json.Marshal(ler) + if err != nil { + return err + } + cml.cm, err = cml.Client.ConfigMaps(cml.ConfigMapMeta.Namespace).Create(ctx, &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: cml.ConfigMapMeta.Name, + Namespace: cml.ConfigMapMeta.Namespace, + Annotations: map[string]string{ + LeaderElectionRecordAnnotationKey: string(recordBytes), + }, + }, + }, metav1.CreateOptions{}) + return err +} + +// Update will update an existing annotation on a given resource. +func (cml *configMapLock) Update(ctx context.Context, ler LeaderElectionRecord) error { + if cml.cm == nil { + return errors.New("configmap not initialized, call get or create first") + } + recordBytes, err := json.Marshal(ler) + if err != nil { + return err + } + if cml.cm.Annotations == nil { + cml.cm.Annotations = make(map[string]string) + } + cml.cm.Annotations[LeaderElectionRecordAnnotationKey] = string(recordBytes) + cm, err := cml.Client.ConfigMaps(cml.ConfigMapMeta.Namespace).Update(ctx, cml.cm, metav1.UpdateOptions{}) + if err != nil { + return err + } + cml.cm = cm + return nil +} + +// RecordEvent in leader election while adding meta-data +func (cml *configMapLock) RecordEvent(s string) { + if cml.LockConfig.EventRecorder == nil { + return + } + events := fmt.Sprintf("%v %v", cml.LockConfig.Identity, s) + subject := &v1.ConfigMap{ObjectMeta: cml.cm.ObjectMeta} + // Populate the type meta, so we don't have to get it from the schema + subject.Kind = "ConfigMap" + subject.APIVersion = v1.SchemeGroupVersion.String() + cml.LockConfig.EventRecorder.Eventf(subject, v1.EventTypeNormal, "LeaderElection", events) +} + +// Describe is used to convert details on current resource lock +// into a string +func (cml *configMapLock) Describe() string { + return fmt.Sprintf("%v/%v", cml.ConfigMapMeta.Namespace, cml.ConfigMapMeta.Name) +} + +// Identity returns the Identity of the lock +func (cml *configMapLock) Identity() string { + return cml.LockConfig.Identity +} diff --git a/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/endpointslock.go b/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/endpointslock.go new file mode 100644 index 0000000000..af3fa16269 --- /dev/null +++ b/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/endpointslock.go @@ -0,0 +1,121 @@ +/* +Copyright 2016 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcelock + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + corev1client "k8s.io/client-go/kubernetes/typed/core/v1" +) + +type endpointsLock struct { + // EndpointsMeta should contain a Name and a Namespace of an + // Endpoints object that the LeaderElector will attempt to lead. + EndpointsMeta metav1.ObjectMeta + Client corev1client.EndpointsGetter + LockConfig ResourceLockConfig + e *v1.Endpoints +} + +// Get returns the election record from a Endpoints Annotation +func (el *endpointsLock) Get(ctx context.Context) (*LeaderElectionRecord, []byte, error) { + var record LeaderElectionRecord + var err error + el.e, err = el.Client.Endpoints(el.EndpointsMeta.Namespace).Get(ctx, el.EndpointsMeta.Name, metav1.GetOptions{}) + if err != nil { + return nil, nil, err + } + if el.e.Annotations == nil { + el.e.Annotations = make(map[string]string) + } + recordStr, found := el.e.Annotations[LeaderElectionRecordAnnotationKey] + recordBytes := []byte(recordStr) + if found { + if err := json.Unmarshal(recordBytes, &record); err != nil { + return nil, nil, err + } + } + return &record, recordBytes, nil +} + +// Create attempts to create a LeaderElectionRecord annotation +func (el *endpointsLock) Create(ctx context.Context, ler LeaderElectionRecord) error { + recordBytes, err := json.Marshal(ler) + if err != nil { + return err + } + el.e, err = el.Client.Endpoints(el.EndpointsMeta.Namespace).Create(ctx, &v1.Endpoints{ + ObjectMeta: metav1.ObjectMeta{ + Name: el.EndpointsMeta.Name, + Namespace: el.EndpointsMeta.Namespace, + Annotations: map[string]string{ + LeaderElectionRecordAnnotationKey: string(recordBytes), + }, + }, + }, metav1.CreateOptions{}) + return err +} + +// Update will update and existing annotation on a given resource. +func (el *endpointsLock) Update(ctx context.Context, ler LeaderElectionRecord) error { + if el.e == nil { + return errors.New("endpoint not initialized, call get or create first") + } + recordBytes, err := json.Marshal(ler) + if err != nil { + return err + } + if el.e.Annotations == nil { + el.e.Annotations = make(map[string]string) + } + el.e.Annotations[LeaderElectionRecordAnnotationKey] = string(recordBytes) + e, err := el.Client.Endpoints(el.EndpointsMeta.Namespace).Update(ctx, el.e, metav1.UpdateOptions{}) + if err != nil { + return err + } + el.e = e + return nil +} + +// RecordEvent in leader election while adding meta-data +func (el *endpointsLock) RecordEvent(s string) { + if el.LockConfig.EventRecorder == nil { + return + } + events := fmt.Sprintf("%v %v", el.LockConfig.Identity, s) + subject := &v1.Endpoints{ObjectMeta: el.e.ObjectMeta} + // Populate the type meta, so we don't have to get it from the schema + subject.Kind = "Endpoints" + subject.APIVersion = v1.SchemeGroupVersion.String() + el.LockConfig.EventRecorder.Eventf(subject, v1.EventTypeNormal, "LeaderElection", events) +} + +// Describe is used to convert details on current resource lock +// into a string +func (el *endpointsLock) Describe() string { + return fmt.Sprintf("%v/%v", el.EndpointsMeta.Namespace, el.EndpointsMeta.Name) +} + +// Identity returns the Identity of the lock +func (el *endpointsLock) Identity() string { + return el.LockConfig.Identity +} diff --git a/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/interface.go b/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/interface.go new file mode 100644 index 0000000000..c6e23bda16 --- /dev/null +++ b/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/interface.go @@ -0,0 +1,227 @@ +/* +Copyright 2016 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcelock + +import ( + "context" + "fmt" + clientset "k8s.io/client-go/kubernetes" + restclient "k8s.io/client-go/rest" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + coordinationv1 "k8s.io/client-go/kubernetes/typed/coordination/v1" + corev1 "k8s.io/client-go/kubernetes/typed/core/v1" +) + +const ( + LeaderElectionRecordAnnotationKey = "control-plane.alpha.kubernetes.io/leader" + endpointsResourceLock = "endpoints" + configMapsResourceLock = "configmaps" + LeasesResourceLock = "leases" + // When using EndpointsLeasesResourceLock, you need to ensure that + // API Priority & Fairness is configured with non-default flow-schema + // that will catch the necessary operations on leader-election related + // endpoint objects. + // + // The example of such flow scheme could look like this: + // apiVersion: flowcontrol.apiserver.k8s.io/v1beta2 + // kind: FlowSchema + // metadata: + // name: my-leader-election + // spec: + // distinguisherMethod: + // type: ByUser + // matchingPrecedence: 200 + // priorityLevelConfiguration: + // name: leader-election # reference the PL + // rules: + // - resourceRules: + // - apiGroups: + // - "" + // namespaces: + // - '*' + // resources: + // - endpoints + // verbs: + // - get + // - create + // - update + // subjects: + // - kind: ServiceAccount + // serviceAccount: + // name: '*' + // namespace: kube-system + EndpointsLeasesResourceLock = "endpointsleases" + // When using EndpointsLeasesResourceLock, you need to ensure that + // API Priority & Fairness is configured with non-default flow-schema + // that will catch the necessary operations on leader-election related + // configmap objects. + // + // The example of such flow scheme could look like this: + // apiVersion: flowcontrol.apiserver.k8s.io/v1beta2 + // kind: FlowSchema + // metadata: + // name: my-leader-election + // spec: + // distinguisherMethod: + // type: ByUser + // matchingPrecedence: 200 + // priorityLevelConfiguration: + // name: leader-election # reference the PL + // rules: + // - resourceRules: + // - apiGroups: + // - "" + // namespaces: + // - '*' + // resources: + // - configmaps + // verbs: + // - get + // - create + // - update + // subjects: + // - kind: ServiceAccount + // serviceAccount: + // name: '*' + // namespace: kube-system + ConfigMapsLeasesResourceLock = "configmapsleases" +) + +// LeaderElectionRecord is the record that is stored in the leader election annotation. +// This information should be used for observational purposes only and could be replaced +// with a random string (e.g. UUID) with only slight modification of this code. +// TODO(mikedanese): this should potentially be versioned +type LeaderElectionRecord struct { + // HolderIdentity is the ID that owns the lease. If empty, no one owns this lease and + // all callers may acquire. Versions of this library prior to Kubernetes 1.14 will not + // attempt to acquire leases with empty identities and will wait for the full lease + // interval to expire before attempting to reacquire. This value is set to empty when + // a client voluntarily steps down. + HolderIdentity string `json:"holderIdentity"` + LeaseDurationSeconds int `json:"leaseDurationSeconds"` + AcquireTime metav1.Time `json:"acquireTime"` + RenewTime metav1.Time `json:"renewTime"` + LeaderTransitions int `json:"leaderTransitions"` +} + +// EventRecorder records a change in the ResourceLock. +type EventRecorder interface { + Eventf(obj runtime.Object, eventType, reason, message string, args ...interface{}) +} + +// ResourceLockConfig common data that exists across different +// resource locks +type ResourceLockConfig struct { + // Identity is the unique string identifying a lease holder across + // all participants in an election. + Identity string + // EventRecorder is optional. + EventRecorder EventRecorder +} + +// Interface offers a common interface for locking on arbitrary +// resources used in leader election. The Interface is used +// to hide the details on specific implementations in order to allow +// them to change over time. This interface is strictly for use +// by the leaderelection code. +type Interface interface { + // Get returns the LeaderElectionRecord + Get(ctx context.Context) (*LeaderElectionRecord, []byte, error) + + // Create attempts to create a LeaderElectionRecord + Create(ctx context.Context, ler LeaderElectionRecord) error + + // Update will update and existing LeaderElectionRecord + Update(ctx context.Context, ler LeaderElectionRecord) error + + // RecordEvent is used to record events + RecordEvent(string) + + // Identity will return the locks Identity + Identity() string + + // Describe is used to convert details on current resource lock + // into a string + Describe() string +} + +// Manufacture will create a lock of a given type according to the input parameters +func New(lockType string, ns string, name string, coreClient corev1.CoreV1Interface, coordinationClient coordinationv1.CoordinationV1Interface, rlc ResourceLockConfig) (Interface, error) { + endpointsLock := &endpointsLock{ + EndpointsMeta: metav1.ObjectMeta{ + Namespace: ns, + Name: name, + }, + Client: coreClient, + LockConfig: rlc, + } + configmapLock := &configMapLock{ + ConfigMapMeta: metav1.ObjectMeta{ + Namespace: ns, + Name: name, + }, + Client: coreClient, + LockConfig: rlc, + } + leaseLock := &LeaseLock{ + LeaseMeta: metav1.ObjectMeta{ + Namespace: ns, + Name: name, + }, + Client: coordinationClient, + LockConfig: rlc, + } + switch lockType { + case endpointsResourceLock: + return nil, fmt.Errorf("endpoints lock is removed, migrate to %s", EndpointsLeasesResourceLock) + case configMapsResourceLock: + return nil, fmt.Errorf("configmaps lock is removed, migrate to %s", ConfigMapsLeasesResourceLock) + case LeasesResourceLock: + return leaseLock, nil + case EndpointsLeasesResourceLock: + return &MultiLock{ + Primary: endpointsLock, + Secondary: leaseLock, + }, nil + case ConfigMapsLeasesResourceLock: + return &MultiLock{ + Primary: configmapLock, + Secondary: leaseLock, + }, nil + default: + return nil, fmt.Errorf("Invalid lock-type %s", lockType) + } +} + +// NewFromKubeconfig will create a lock of a given type according to the input parameters. +// Timeout set for a client used to contact to Kubernetes should be lower than +// RenewDeadline to keep a single hung request from forcing a leader loss. +// Setting it to max(time.Second, RenewDeadline/2) as a reasonable heuristic. +func NewFromKubeconfig(lockType string, ns string, name string, rlc ResourceLockConfig, kubeconfig *restclient.Config, renewDeadline time.Duration) (Interface, error) { + // shallow copy, do not modify the kubeconfig + config := *kubeconfig + timeout := renewDeadline / 2 + if timeout < time.Second { + timeout = time.Second + } + config.Timeout = timeout + leaderElectionClient := clientset.NewForConfigOrDie(restclient.AddUserAgent(&config, "leader-election")) + return New(lockType, ns, name, leaderElectionClient.CoreV1(), leaderElectionClient.CoordinationV1(), rlc) +} diff --git a/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/leaselock.go b/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/leaselock.go new file mode 100644 index 0000000000..ab80d7f158 --- /dev/null +++ b/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/leaselock.go @@ -0,0 +1,139 @@ +/* +Copyright 2018 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcelock + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + coordinationv1 "k8s.io/api/coordination/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + coordinationv1client "k8s.io/client-go/kubernetes/typed/coordination/v1" +) + +type LeaseLock struct { + // LeaseMeta should contain a Name and a Namespace of a + // LeaseMeta object that the LeaderElector will attempt to lead. + LeaseMeta metav1.ObjectMeta + Client coordinationv1client.LeasesGetter + LockConfig ResourceLockConfig + lease *coordinationv1.Lease +} + +// Get returns the election record from a Lease spec +func (ll *LeaseLock) Get(ctx context.Context) (*LeaderElectionRecord, []byte, error) { + var err error + ll.lease, err = ll.Client.Leases(ll.LeaseMeta.Namespace).Get(ctx, ll.LeaseMeta.Name, metav1.GetOptions{}) + if err != nil { + return nil, nil, err + } + record := LeaseSpecToLeaderElectionRecord(&ll.lease.Spec) + recordByte, err := json.Marshal(*record) + if err != nil { + return nil, nil, err + } + return record, recordByte, nil +} + +// Create attempts to create a Lease +func (ll *LeaseLock) Create(ctx context.Context, ler LeaderElectionRecord) error { + var err error + ll.lease, err = ll.Client.Leases(ll.LeaseMeta.Namespace).Create(ctx, &coordinationv1.Lease{ + ObjectMeta: metav1.ObjectMeta{ + Name: ll.LeaseMeta.Name, + Namespace: ll.LeaseMeta.Namespace, + }, + Spec: LeaderElectionRecordToLeaseSpec(&ler), + }, metav1.CreateOptions{}) + return err +} + +// Update will update an existing Lease spec. +func (ll *LeaseLock) Update(ctx context.Context, ler LeaderElectionRecord) error { + if ll.lease == nil { + return errors.New("lease not initialized, call get or create first") + } + ll.lease.Spec = LeaderElectionRecordToLeaseSpec(&ler) + + lease, err := ll.Client.Leases(ll.LeaseMeta.Namespace).Update(ctx, ll.lease, metav1.UpdateOptions{}) + if err != nil { + return err + } + + ll.lease = lease + return nil +} + +// RecordEvent in leader election while adding meta-data +func (ll *LeaseLock) RecordEvent(s string) { + if ll.LockConfig.EventRecorder == nil { + return + } + events := fmt.Sprintf("%v %v", ll.LockConfig.Identity, s) + subject := &coordinationv1.Lease{ObjectMeta: ll.lease.ObjectMeta} + // Populate the type meta, so we don't have to get it from the schema + subject.Kind = "Lease" + subject.APIVersion = coordinationv1.SchemeGroupVersion.String() + ll.LockConfig.EventRecorder.Eventf(subject, corev1.EventTypeNormal, "LeaderElection", events) +} + +// Describe is used to convert details on current resource lock +// into a string +func (ll *LeaseLock) Describe() string { + return fmt.Sprintf("%v/%v", ll.LeaseMeta.Namespace, ll.LeaseMeta.Name) +} + +// Identity returns the Identity of the lock +func (ll *LeaseLock) Identity() string { + return ll.LockConfig.Identity +} + +func LeaseSpecToLeaderElectionRecord(spec *coordinationv1.LeaseSpec) *LeaderElectionRecord { + var r LeaderElectionRecord + if spec.HolderIdentity != nil { + r.HolderIdentity = *spec.HolderIdentity + } + if spec.LeaseDurationSeconds != nil { + r.LeaseDurationSeconds = int(*spec.LeaseDurationSeconds) + } + if spec.LeaseTransitions != nil { + r.LeaderTransitions = int(*spec.LeaseTransitions) + } + if spec.AcquireTime != nil { + r.AcquireTime = metav1.Time{spec.AcquireTime.Time} + } + if spec.RenewTime != nil { + r.RenewTime = metav1.Time{spec.RenewTime.Time} + } + return &r + +} + +func LeaderElectionRecordToLeaseSpec(ler *LeaderElectionRecord) coordinationv1.LeaseSpec { + leaseDurationSeconds := int32(ler.LeaseDurationSeconds) + leaseTransitions := int32(ler.LeaderTransitions) + return coordinationv1.LeaseSpec{ + HolderIdentity: &ler.HolderIdentity, + LeaseDurationSeconds: &leaseDurationSeconds, + AcquireTime: &metav1.MicroTime{ler.AcquireTime.Time}, + RenewTime: &metav1.MicroTime{ler.RenewTime.Time}, + LeaseTransitions: &leaseTransitions, + } +} diff --git a/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/multilock.go b/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/multilock.go new file mode 100644 index 0000000000..5ee1dcbb50 --- /dev/null +++ b/vendor/k8s.io/client-go/tools/leaderelection/resourcelock/multilock.go @@ -0,0 +1,104 @@ +/* +Copyright 2019 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resourcelock + +import ( + "bytes" + "context" + "encoding/json" + + apierrors "k8s.io/apimachinery/pkg/api/errors" +) + +const ( + UnknownLeader = "leaderelection.k8s.io/unknown" +) + +// MultiLock is used for lock's migration +type MultiLock struct { + Primary Interface + Secondary Interface +} + +// Get returns the older election record of the lock +func (ml *MultiLock) Get(ctx context.Context) (*LeaderElectionRecord, []byte, error) { + primary, primaryRaw, err := ml.Primary.Get(ctx) + if err != nil { + return nil, nil, err + } + + secondary, secondaryRaw, err := ml.Secondary.Get(ctx) + if err != nil { + // Lock is held by old client + if apierrors.IsNotFound(err) && primary.HolderIdentity != ml.Identity() { + return primary, primaryRaw, nil + } + return nil, nil, err + } + + if primary.HolderIdentity != secondary.HolderIdentity { + primary.HolderIdentity = UnknownLeader + primaryRaw, err = json.Marshal(primary) + if err != nil { + return nil, nil, err + } + } + return primary, ConcatRawRecord(primaryRaw, secondaryRaw), nil +} + +// Create attempts to create both primary lock and secondary lock +func (ml *MultiLock) Create(ctx context.Context, ler LeaderElectionRecord) error { + err := ml.Primary.Create(ctx, ler) + if err != nil && !apierrors.IsAlreadyExists(err) { + return err + } + return ml.Secondary.Create(ctx, ler) +} + +// Update will update and existing annotation on both two resources. +func (ml *MultiLock) Update(ctx context.Context, ler LeaderElectionRecord) error { + err := ml.Primary.Update(ctx, ler) + if err != nil { + return err + } + _, _, err = ml.Secondary.Get(ctx) + if err != nil && apierrors.IsNotFound(err) { + return ml.Secondary.Create(ctx, ler) + } + return ml.Secondary.Update(ctx, ler) +} + +// RecordEvent in leader election while adding meta-data +func (ml *MultiLock) RecordEvent(s string) { + ml.Primary.RecordEvent(s) + ml.Secondary.RecordEvent(s) +} + +// Describe is used to convert details on current resource lock +// into a string +func (ml *MultiLock) Describe() string { + return ml.Primary.Describe() +} + +// Identity returns the Identity of the lock +func (ml *MultiLock) Identity() string { + return ml.Primary.Identity() +} + +func ConcatRawRecord(primaryRaw, secondaryRaw []byte) []byte { + return bytes.Join([][]byte{primaryRaw, secondaryRaw}, []byte(",")) +}