Kubernetes控制器主从选举

Kubernetes控制器主从选举

在开发Kubernetes控制器时，可以通过client-go中的leaderelection包实现控制器的主从选举 ，实现控制器的高可用。下面介绍基于资源锁的选主。

| 基本原理

对资源创建资源锁，创建成功资源锁成为主。

资源锁类型分为以下几种类型：

const ( EndpointsResourceLock = "endpoints" ConfigMapsResourceLock = "configmaps" LeasesResourceLock = "leases" EndpointsLeasesResourceLock = "endpointsleases" ConfigMapsLeasesResourceLock = "configmapsleases")

获取资源锁后，对应资源的Metadata将会更新，增加Annotations:

sukai@sukai:~/.kube$ kubectl -n spark-operator get cm spark-operator-lock -o yamlapiVersion: v1kind: ConfigMapmetadata: annotations: control-plane.alpha.kubernetes.io/leader: '{"holderIdentity":"sukai_8617c3fd-15a5-4c51-a4da-12fc86999d83","leaseDurationSeconds":15,"acquireTime":"2021-04-26T07:18:57Z","renewTime":"2021-04-26T08:07:49Z","leaderTransitions":16}' creationTimestamp: "2021-04-26T06:38:38Z" managedFields: - apiVersion: v1 fieldsType: FieldsV1 fieldsV1: f:metadata: f:annotations: {} manager: leader_01.exe operation: Update time: "2021-04-26T07:18:33Z" - apiVersion: v1 fieldsType: FieldsV1 fieldsV1: f:metadata: f:annotations: f:control-plane.alpha.kubernetes.io/leader: {} manager: leader_03.exe operation: Update time: "2021-04-26T07:18:55Z" name: spark-operator-lock namespace: spark-operator resourceVersion: "1527867" uid: 7888aa1e-72c8-4994-b3df-d018a4e3cd93

Annotation记录为：

HolderIdentity锁持有者，也就是主

LeaseDurationSeconds 租约周期

AcquireTime最后一次获取锁的时间

RenewTime最后一次续约时间

LeaderTransitions锁选举次数，主成功选举一次加1.

// 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 versionedtype 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"`}

资源锁有租约时间，所有的候选者在等待一个完整租约时间后，发现资源锁没有续租，则可以竞争这个资源锁。比如选举配置项：LeaseDuration为15秒，RenewDeadline为14秒，RetryPeriod为4秒，则表示主获取一个资源锁的周期为15秒，在14秒内没有续约则超期，候选者和主在每隔4秒进行获取锁或者续约操作一次。租约已经超期，锁租约时间到了以后候选者就可以获取锁。

| 代码示例

下面以spark-on-k8s-operator为例，看一下如何使用基于资源锁的选举。

首先创建一个空结构体的信号管道，startCh用于接收选举完成的信号。

startCh := make(chan struct{}, 1)

定义资源锁持有者的标识，这里以主机名+UUID方式。

id, err := os.Hostname() if err != nil { glog.Fatal(err) } id = id + "_" + string(uuid.NewUUID())

初始化一个资源锁，定义类型为configmap资源锁，这里定义在spark-operator命名空间，创建名为spark-operator-lock的configmap，使用的kubernetes API的clientSet接口，使用id作为标识，和资源锁变化事件生产者，FakeRecorder将所有事件都扔了。

var ( leaderElectionLockNamespace = flag.String("leader-election-lock-namespace", "spark-operator", "Namespace in which to create the ConfigMap for leader election.") leaderElectionLockName = flag.String("leader-election-lock-name", "spark-operator-lock", "Name of the ConfigMap for leader election."))

resourceLock, err := resourcelock.New(resourcelock.ConfigMapsResourceLock, *leaderElectionLockNamespace, *leaderElectionLockName, kubeClient.CoreV1(), kubeClient.CoordinationV1(), resourcelock.ResourceLockConfig{ Identity: id, // TODO: This is a workaround for a nil dereference in client-go. This line can be removed when that dependency is updated. EventRecorder: &record.FakeRecorder{}, }) if err != nil { glog.Fatal(err) }

初始化选举配置，定义资源锁，租约相关的时间周期，以及成功选举和选举失败的回调函数。这里回调函数就是关闭channel。

electionCfg := leaderelection.LeaderElectionConfig{ Lock: resourceLock, LeaseDuration: *leaderElectionLeaseDuration, RenewDeadline: *leaderElectionRenewDeadline, RetryPeriod: *leaderElectionRetryPeriod, Callbacks: leaderelection.LeaderCallbacks{ OnStartedLeading: func(c context.Context) { // glog.Info("start") close(startCh) }, OnStoppedLeading: func() { // glog.Info("stopped") close(stopCh) }, }, }

根据选举配置electionCfg初始化一个选举实例，调用Run开始选举。

elector, err := leaderelection.NewLeaderElector(electionCfg) if err != nil { glog.Fatal(err) } go elector.Run(context.Background())

阻塞读取startCh通道，当选举成功回调函数close(startCh)，将开始往下执行业务逻辑。

<-startCh glog.Info("Starting application controller goroutines") select { case <-signalCh: close(stopCh) case <-stopCh: }