在集群中存在某种业务场景，一批相同功能的进程同时运行，但是同一时刻，只能有一个工作，只有当正在工作的进程异常时，才会由另一个进程进行接管。这种业务逻辑通常用于实现一主多从。
如果有人认为，传统应用需要部署多个通常是为了容灾，而在k8s上运行的Pod受控制器管理，如果Pod异常或者Pod所在宿主机宕机，Pod是可以漂移到其他节点的，所以，不需要部署多个Pod，只需要部署一个Pod就行。k8s上的Pod确实可以漂移，但是，如果宿主机宕机，k8s认为Pod异常，并在其他节点重建Pod是有周期的，不能在查询不到Pod状态时立刻就将Pod驱逐掉，也许节点只是临时不可用呢？例如，负载很高，因此，判断宿主机宕机需要有个时间短。
因此，在k8s中运行一主多从是为了能够实现主的快速切换。
k8s中也有这种业务场景，在多master场景下，只能有一个master上的进程工作，例如，scheduler和controller-manager。以scheduler来说，它的工作是给Pod分配合适的宿主机，如果有多个scheduler同时运行，就会出现竞争，因此，如果允许这种场景存在的话，就又需要实现一种调度逻辑：某个Pod由哪个scheduler进行调度，这相当于又要实现一层调度。但是，实际上调度工作是相对比较简单的，不需要多个scheduler进行负载，只需要一个scheduler进行调度就行。因此，k8s提供了leader-election的能力。
leader-election的具体工作方式是：各候选者将自身的信息写入某一个资源，如果写成功，某个后选择就称为了主，其他就是备，同时，在之后主会定期更新资源的时间，如果超过一段时间未更新时间，其他候选者发现资源的最后更新时间超过一定值，就会认为主挂掉，然后会向资源写入自身信息，从而成为新的主。
基于该原理，有一个现成的镜像可以使用：instana/leader-elector。
apiVersion: apps/v1
kind: Deployment
metadata:
labels:
app: leader
name: leader
spec:
replicas: 3
selector:
matchLabels:
app: leader
template:
metadata:
labels:
app: leader
spec:
containers:
- image: instana/leader-elector:0.5.13
name: leader
command: ["/app/server","--id=$(POD_NAME)","--election=leader-election","--http=0.0.0.0:4040"]
env:
- name: POD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.name
上面的yaml有两个需要注意的地方：
创建deploy后，会启动三个Pod，通过kubectl logs可以看到只有一个Pod成为主，也就是向资源名称为leader-election的Endpoint写入了自身的Pod名称。然后通过代理(kubectl proxy)访问：http://localhost:8001/api/v1/namespaces/default/pods/:4040/proxy/，就会看到主的Pod名称。
知道了leader-election的大概原理，也知道了上面的镜像可以直接实现主的选举，那么如何使用呢？
直接将上面的leader-elector镜像作为Sidecar，将Pod名称作为候选者的唯一标识，然后将Pod名称也注入到环境变量，在业务进程起来后，定时调用http://localhost:4040就可以获取主，如果发现主的名称与自身的Pod的名称一致，就执行业务逻辑，否则一直等待。
使用Sidecar的好处是比较方便，开发成本低，不便的地方就是，适用场景有限，只能写入Endpoint资源。因此，在某些场景下，可以使用SDK，直接基于leader-election库开发。
创建一个Lease类型的锁(当然，也可以是其他类型，但是lease更加轻量)，创建资源时需要指定资源的命名空间、名称、标识(这一批Pod都会该命名空间的资源写入自身的唯一标识)。然后调用leaderelection库中的RunOrDie()函数，此时会指定：
// leaderelection/leaderelection.go
func (le *LeaderElector) Run(ctx context.Context) {
defer runtime.HandleCrash()
defer func() {
le.config.Callbacks.OnStoppedLeading()
}()
// 申请资源锁，有三种情形：
// 1 出错，则返回false，Run()直接退出
// 2 获取到锁了，则返回true，执行回调函数
// 3 未获取到锁，acquire()函数不会返回
if !le.acquire(ctx) {
return // ctx signalled done
}
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// 申请成功后，执行回调函数
go le.config.Callbacks.OnStartedLeading(ctx)
// 定时刷新租约
le.renew(ctx)
}
// 申请资源锁
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)
// 每隔RetryPeriod去申请资源锁，或者更新
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
}
func (le *LeaderElector) renew(ctx context.Context) {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// 每隔RetryPeriod尝试更新租约
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()
}
}
// 尝试获取或者更新资源锁
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 获取资源锁记录
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 将资源锁记录与缓存的上一次的值进行对比
// 如果当前不是leader，并且资源锁没有过期，则退出
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() {
// 当前是leader，锁资源未过期，将之前的资源锁的数据填充到新的资源锁中(申请锁时间，切换次数)
leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTime
leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions
} else {
// 当前不是leader
leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + 1
}
// 更新资源锁
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
}