为什么需要 Pod 之间的 Leader Election

一般来说,由 Deployment 创建的 1 个或多个 Pod 都是对等关系,彼此之间提供一样的服务。但是在某些场合,多个 Pod 之间需要有一个 Leader 的角色,即:

  • Pod 之间有且只有一个 Leader

  • Leader 在一定周期不可用时,其他 Pod 会再选出一个 Leader

  • 由处于 Leader 身份的 Pod 来完成某些特殊的业务逻辑(通常是写操作)

比如,当多个 Pod 之间只需要一个写者时,如果不采用 Leader Election,那么就必须在 Pod 启动之初人为地配置一个 Leader。如果配置的 Leader 在后续的服务中失效且没有对应机制来生成新的 Leader,那么对应 Pod 服务就可能处于不可用状态,违背高可用原则。

典型地,Kubernetes 的核心组件 kube-controller-manager 就需要一个需要 Leader 的场景。当 kube-controller-manager 的启动参数设置 --leader-elect=true 时,对应节点的 kube-controller-manager 在启动时会执行选主操作。当选出一个 Leader 之后,由 Leader 来启动所有的控制器。如果 Leader Pod 不可用,将会自动选出新的 Leader Pod,从而保障控制器仍处于运行状态。

一个简单的 Leader Election 的例子

备注:该例子取自项目文档

启动一个 leader-elector 的 Pod

  1. 创建一个 leader-elector 的 Deployment,其中的 Pod 会进行 Leader Election 的过程

    1
    
     $ kubectl run leader-elector --image=k8s.gcr.io/leader-elector:0.5 --replicas=3 -- --election=example --http=0.0.0.0:4040

    副本数为 3,即将生成 3 个 Pod,如果运行成功,可观察到:

    1
    2
    3
    4
    5
    
     $ kubectl get po
     NAME                              READY   STATUS    RESTARTS   AGE
     leader-elector-68dcb58d55-7dhdz   1/1     Running   0          2m36s
     leader-elector-68dcb58d55-g5zp8   1/1     Running   0          2m36s
     leader-elector-68dcb58d55-q45pd   1/1     Running   0          2m36s
  2. 查看哪个 Pod 成为 Leader

    可以逐个查看 Pod 的日志:

    1
    
     $ kubectl logs -f ${pod_name}

    如果是 Leader 的话,将会有如下的日志:

    1
    2
    3
    4
    5
    6
    7
    
     $ kubectl logs leader-elector-68dcb58d55-g5zp8
     leader-elector-9577494c7-l64lp is the leader
     I0122 03:24:31.779331       8 leaderelection.go:296] lock is held by leader-elector-9577494c7-l64lp and has not yet expired
     I0122 03:24:36.101800       8 leaderelection.go:296] lock is held by leader-elector-9577494c7-l64lp and has not yet expired
     I0122 03:24:41.426387       8 leaderelection.go:296] lock is held by leader-elector-9577494c7-l64lp and has not yet expired
     I0122 03:24:45.947321       8 leaderelection.go:215] sucessfully acquired lease default/example
    leader-elector-68dcb58d55-g5zp8 is the leader

    更通用的方式是查看资源锁的身份标识信息:

    1
    
     $ kubectl get ep example -o yaml

    通过查看 annotations 中的 control-plane.alpha.kubernetes.io/leader 字段来获得 Leader 的信息;

  3. 使用 leader-elector 的 HTTP 接口查看 Leader

    leader-elector 实现了一个简单的 HTTP 接口(:4040)来查看当前 Leader:

    1
    2
    
     curl http://localhost:8001/api/v1/namespaces/default/pods/leader-elector-5d77ccc44d-gwsgg:4040/proxy/
     {"name":"leader-elector-5d77ccc44d-7tmgm"}

用 Sidecar 模式使用 leader-elector

如果自己的项目中需要用到 Leader Election 的逻辑,可以有两种方式:

  • 将调用 leaderelection 库的逻辑内嵌到自己项目中

  • 使用 Sidecar 的方式将 leader-elector 容器组合在 Pod 中,通过调用 HTTP 接口来始终获得 Leader 的信息

文档中以 Node.js 的方式举了一个简单例子,大家可以参考,此处不展开了。

Leader Election 的实现

Leader Election 的过程本质上就是一个竞争分布式锁的过程。在 Kubernetes 中,这个分布式锁是以创建 Endpoint 或者 ConfigMap 资源的形式进行:谁先创建了某种资源,谁就获得锁

按照我们以往的惯例,带着问题去看源码。有这么几个问题:

  • Leader Election 如何竞选

  • Leader 不可用之后如何竞选新的 Leader

不同于 Raft 算法的一致性算法的 Leader 竞选,Pod 之间的 Leader Election 是无状态的,也就是说现在的 Leader 无需同步上一个 Leader 的数据信息,这就把竞选的过程变得非常简单:先到先得

这部分代码在 kubernetes/staging/src/k8s.io/client-go/tools/leaderelection 中,取 1.9.2 版本来分析。

资源锁的实现

Kubernetes 实现了两种资源锁(resourcelock):Endpoint 和 ConfigMap。如果是基于 Endpoint 的资源锁,获取到锁的 Pod 将会在对应 Namespace 下创建对应的 Endpoint 对象,并在其 Annotations 上记录 Pod 的信息

比如 kube-controller-manager:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
$ kubectl get ep -n kube-system | grep kube-controller-manager
kube-controller-manager   <none>   41d

$ kubectl describe ep kube-controller-manager -n kube-system
Name:         kube-controller-manager
Namespace:    kube-system
Labels:       <none>
Annotations:  control-plane.alpha.kubernetes.io/leader:
                {"holderIdentity":"szdc-k8sm-0-5","leaseDurationSeconds":15,"acquireTime":"2018-12-11T0...
Subsets:
Events:  <none>

发现在 kube-system 中创建了同名的 Endpoint(kube-controller-manager),并在 Annotations 中以设置了 key 为 control-plane.alpha.kubernetes.io/leader,value 为对应 Leader 信息的 JSON 数据。同理,如果采用 ConfigMap 作为资源锁也是类似的实现模式。

resourcelock 是以 interface 的形式对外暴露,在创建过程(New())通过相应的参数来控制具体实例化的过程:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
// leaderelection/resourcelock/interface.go
type Interface interface {
	// Get returns the LeaderElectionRecord
	Get() (*LeaderElectionRecord, error)

	// Create attempts to create a LeaderElectionRecord
	Create(ler LeaderElectionRecord) error

	// Update will update and existing LeaderElectionRecord
	Update(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
}

其中 Get()Create()Update() 本质上就是对 LeaderElectionRecord 的读写操作。LeaderElectionRecord 定义如下:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
type LeaderElectionRecord struct {
	// 标示当前资源锁的所有权的信息
	HolderIdentity string `json:"holderIdentity"`

	// 资源锁租约时间是多长
	LeaseDurationSeconds int `json:"leaseDurationSeconds"`

	// 锁获得的时间
	AcquireTime metav1.Time `json:"acquireTime"`

	// 续租的时间
	RenewTime metav1.Time `json:"renewTime"`

	// Leader 进行切换的次数
	LeaderTransitions int `json:"leaderTransitions"`
}

理论上,LeaderElectionRecord 是保存在资源锁的 Annotations 中,可以是任意的字符串,此处是将 JSON 序列化为字符串来进行存储。

leaderelection/resourcelock/configmaplock.goleaderelection/resourcelock/endpointslock.go 分别是基于 Endpoint 和 ConfigMap 对上面接口的实现。拿 endpointslock.go 来看,对这几个接口的实现实际上就是对 Endpoint 资源中 Annotations 的增删查改罢了,比较简单,就不详细展开。

竞争锁的过程

完整的 Leader Election 过程在 leaderelection/leaderelection.go 中。

整个过程可以简单描述为:

  1. 每个 Pod 在启动的时候都会创建 LeaderElector 对象,然后执行 LeaderElector.Run() 循环

  2. 在循环中,Pod 会定期(RetryPeriod)去不断尝试创建资源,如果创建成功,就在对应资源的字段中记录 Pod 相关的 Id(比如节点的 hostname)

  3. 在循环周期中,Leader 会不断 Update 资源锁的对应时间信息,从节点则会不断检查资源锁是否过期,如果过期则尝试更新资源,标记资源所有权。这样一来,一旦 Leader 不可用,则对应的资源锁将得不到更新,过期之后其他从节点会再次创建新的资源锁成为 Leader

其中,LeaderElector.Run() 的源码为:

1
2
3
4
5
6
7
8
func (le *LeaderElector) Run() {
    ...
    // 尝试创建锁
    le.acquire()
    // Leader 更新资源锁的租约
    le.renew()
    ...
}

acquire() 会周期性地创建锁或探查锁有没有过期:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
func (le *LeaderElector) acquire() {
    ...
    wait.JitterUntil(func() {
        // 尝试创建或者续约资源锁
        succeeded := le.tryAcquireOrRenew()
        // leader 可能发生了改变,执行相应的 OnNewLeader() 回调函数
        le.maybeReportTransition()
        // 不成功说明创建资源失败,当前 Leader 是其他 Pod
        if !succeeded {
            ...
            return
        }
        ...
    }, le.config.RetryPeriod, JitterFactor, true, stop)
}

执行的周期为 RetryPeriod

我们重点关注 tryAcquireOrRenew() 的逻辑:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
func (le *leaderElector) tryAcquireOrRenew() bool {
    now := metav1.Now()
    leaderElectionRecord := rl.LeaderElectionRecord{
        HolderIdentity:       le.config.Lock.Identity(),
        LeaseDurationSeconds: int(le.config.LeaseDuration) / time.Second),
        // 将租约改成 now
        RenewTime:            now,
        AcquireTime:          now,
    }
    
    // 获取当前的资源锁
    oldLeaderElectionRecord, err := le.config.Lock.Get()
    if err != nil {
        ...
        // 执行到这里说明找不到资源锁,执行资源锁的创建动作
        // 由于资源锁对应的底层 Kubernetes 资源 Endpoint 或 ConfigMap 是不可重复创建的,所以此处创建是安全的
        if err = le.config.Lock.Create(leaderElectionRecord); err != nil {
            ...
        }
        ...
    }
    
    // 如果当前已经有 Leader,进行 Update 操作
    // 如果当前是 Leader:Update 操作就是续租动作,即将对应字段的时间改成当前时间
    // 如果是非 Leader 节点且可运行 Update 操作,则是一个抢夺资源锁的过程,谁先更新成功谁就抢到资源
    ...
    // 如果还没有过期且当前不是 Leader,直接返回
    // 只有 Leader 才进行续租操作且此时其他节点无须抢夺资源锁
    if le.observedTime.Add(le.config.LeaseDuration).After(now.Time) &&
       oldLeaderElectionRecord.HolderIdentity != le.config.Lock.Identity() {
           ...
           return false
    }
    ...
    // 更新资源
    // 对于 Leader 来说,这是一个续租的过程
    // 对于非 Leader 节点(仅在上一个资源锁已经过期),这是一个更新锁所有权的过程
    if err = le.config.Lock.Update(leaderElectionRecord); err != nil {
        ...
    }
}

由上可以看出,tryAcquireOrRenew() 就是一个不断尝试 Update 操作的过程。

如果执行逻辑从 le.acquire() 跳出,往下执行 le.renew(),这说明当前 Pod 已经成功抢到资源锁成为 Leader,必须定期续租:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
func (le *LeaderElector) renew() {
    stop := make(chan struct{})
    // period 为 0 说明会一直执行
    wait.Until(func() {
        // 每间隔 RetryPeriod 就执行 tryAcquireOrRenew()
        // 如果 tryAcquireOrRenew() 返回 false 跳出 Poll()
        // tryAcquireOrRenew() 返回 false 说明续租失败
        err := wait.Poll(le.config.RetryPeriod, le.config.RenewDeadline, func() (bool, error) {
            return le.tryAcquireOrRenew(), nil
        })
        
        // 续租失败,说明已经不是 Leader
        ...
    }, 0, stop)
}

如何使用 leaderelection

让我们来关注一下 election 的实现。

主要的逻辑位于 election/lib/election.go

1
2
3
func RunElection(e *leaderelection.LeaderElector) {
    wait.Forever(e.Run, 0)
}

主体逻辑很简单,就是不断执行 Run()。而 Run() 的实现就是上文中 leaderelectionRun()

上层应用只需要创建(NewElection())创建 LeaderElector 对象,然后在一个 loop 中调用 Run() 即可。

综上所述,Kubernetes 中 Pod 的选举过程本质上还是为了服务的高可用。希望大家研究得愉快

参考文档

  1. kube-controller-manager
  2. Simple Leader Election with Kubernetes and Docker