前言
Kubernetes:kube-scheduler 源码分析 介绍了 kube-scheduler 调度 Pod 的逻辑。文中有一点未提的是，在 Kubernetes 集群中，kube-scheduler 组件是多副本，单实例运行。仅有一个副本作为 leader 运行，当发生故障时，其它副本会抢占为 leader 继续运行。这种机制通过 leader election 实现，本文将从源码角度分析 leader election 选举的实现。并且，在此基础上，在回头看 kube-scheduler 是怎么结合 leader elction 实现多副本，单实例运行的。
leader election
leader election 通过多个副本抢占资源锁的方式实现单实例运行。在 Kubernetes 中，[Configmap|Lease|Endpoint] 可以作为资源锁的实现。
示例
直接看 leader election 代码容易晕，这里从示例入手，看 leader election 是怎么运行的。
1// lease.go
2package main
3
4func buildConfig(kubeconfig string) (*rest.Config, error) {
5 ...
6}
7
8func main() {
9 klog.InitFlags(nil)
10 var kubeconfig string
11 var leaseLockName string
12 var leaseLockNamespace string
13 var id string
14
15 flag.StringVar(&kubeconfig, "kubeconfig", "", "absolute path to the kubeconfig file")
16 flag.StringVar(&id, "id", uuid.New().String(), "the holder identity name")
17 flag.StringVar(&leaseLockName, "lease-lock-name", "kube-scheduler", "the lease lock resource name")
18 flag.StringVar(&leaseLockNamespace, "lease-lock-namespace", "kube-system", "the lease lock resource namespace")
19 flag.Parse()
20
21 if leaseLockName == "" {
22 klog.Fatal("unable to get lease lock resource name (missing lease-lock-name flag).")
23 }
24 if leaseLockNamespace == "" {
25 klog.Fatal("unable to get lease lock resource namespace (missing lease-lock-namespace flag).")
26 }
27
28 config, err := buildConfig(kubeconfig)
29 if err != nil {
30 klog.Fatal(err)
31 }
32 client := clientset.NewForConfigOrDie(config)
33
34 run := func(ctx context.Context) {
35 klog.Info("Controller loop...")
36 }
37
38 ctx, cancel := context.WithCancel(context.Background())
39 defer cancel()
40
41 ch := make(chan os.Signal, 1)
42 signal.Notify(ch, os.Interrupt, syscall.SIGTERM)
43 go func() {
44 <-ch
45 klog.Info("Received termination, signaling shutdown")
46 cancel()
47 }()
48
49 lock := &resourcelock.LeaseLock{
50 LeaseMeta: metav1.ObjectMeta{
51 Name: leaseLockName,
52 Namespace: leaseLockNamespace,
53 },
54 Client: client.CoordinationV1(),
55 LockConfig: resourcelock.ResourceLockConfig{
56 Identity: id,
57 },
58 }
59
60 leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
61 Lock: lock,
62 ReleaseOnCancel: true,
63 LeaseDuration: 60 * time.Second,
64 RenewDeadline: 15 * time.Second,
65 RetryPeriod: 5 * time.Second,
66 Callbacks: leaderelection.LeaderCallbacks{
67 OnStartedLeading: func(ctx context.Context) {
68 // we're notified when we start - this is where you would
69 // usually put your code
70 run(ctx)
71 },
72 OnStoppedLeading: func() {
73 // we can do cleanup here
74 klog.Infof("leader lost: %s", id)
75 os.Exit(0)
76 },
77 OnNewLeader: func(identity string) {
78 // we're notified when new leader elected
79 if identity == id {
80 // I just got the lock
81 return
82 }
83 klog.Infof("new leader elected: %s", identity)
84 },
85 },
86 })
87}
详细代码戳 这里
运行 id 为 1 的进程抢占 lease：
1# go run lease.go --kubeconfig=/root/.kube/config -id=1 -lease-lock-name=example -lease-lock-namespace=default
2I0222 09:44:40.389093 4054932 leaderelection.go:250] attempting to acquire leader lease default/example...
3I0222 09:44:40.400607 4054932 leaderelection.go:260] successfully acquired lease default/example
4I0222 09:44:40.402998 4054932 lease.go:87] Controller loop...
可以看到，id 为 1 的进程（简称进程 1）抢占到 lease，接着进入进程的运行逻辑。
继续运行 id 为 2 的进程（简称进程 2）抢占 lease：
1# go run lease.go --kubeconfig=/root/.kube/config -id=2 -lease-lock-name=example -lease-lock-namespace=default
2I0222 10:26:53.162990 4070458 leaderelection.go:250] attempting to acquire leader lease default/example...
3I0222 10:26:53.170046 4070458 lease.go:151] new leader elected: 1
进程 2 抢占不到 lease，因为此时进程 1 是 leader，且 lease 未被释放或者过期。
kill 掉进程 1（注：此 id 不是进程的 PID），查看进程 2 能否抢占到 lease：
1# go run lease.go --kubeconfig=/root/.kube/config -id=2 -lease-lock-name=example -lease-lock-namespace=default
2I0222 10:26:53.162990 4070458 leaderelection.go:250] attempting to acquire leader lease default/example...
3I0222 10:26:53.170046 4070458 lease.go:151] new leader elected: 1
4I0222 10:31:54.704714 4070458 leaderelection.go:260] successfully acquired lease default/example
5I0222 10:31:54.704931 4070458 lease.go:87] Controller loop...
可以看到，进程 2 成功抢占 lease。
源码分析
lease 资源
lease 作为 leader election 的资源锁存在，多副本抢占 lease 实际是对 lease 进行 [Get|Create|Update] 操作的过程。我们看 lease 在 Kubernetes 的资源表示：
1# kubectl describe lease example -n default
2Name: example
3Namespace: default
4Labels: <none>
5Annotations: <none>
6API Version: coordination.k8s.io/v1
7Kind: Lease
8Metadata:
9 ...
10Spec:
11 Acquire Time: 2024-02-22T08:31:54.696415Z
12 Holder Identity: 2
13 Lease Duration Seconds: 60
14 Lease Transitions: 5
15 Renew Time: 2024-02-22T08:51:47.060020Z
16Events: <none>
在 Spec 域中有五个字段，分别介绍如下：
- Acquire Time: 首次获取 lease 的时间。
- Holder Identity: 当前持有 lease 的用户身份。
- Lease Duration Seconds: lease 过期时间。
- Lease Transitions: lease 被多少 Holder 持有过。
- Renew Time: Holder 刷新 lease 的时间。lease 的持有者需要 renew lease，否则其它副本将抢占过期的 lease。
多副本抢占 lease 的信息将被记录到这几个字段中。lease 存储在 etcd 中，通过 Raft 算法实现 lease 资源的唯一性和原子性，以保证多副本仅可抢占唯一的 lease。
抢占状态图
多副本抢占 lease 的状态图如下，结合状态图看源码会更加清晰。
创建 lease
我们从创建 lease 开始分析。
首先，删掉 lease 资源。
1# kubectl get lease
2NAME HOLDER AGE
3example 2 23h
4# kubectl delete lease example
5lease.coordination.k8s.io "example" deleted
然后，调式模式下运行 lease.go，进入 leaderelection.RunOrDie 查看函数做了什么。
1// k8s.io/client-go/tools/leaderelection/leaderelection.go
2func RunOrDie(ctx context.Context, lec LeaderElectionConfig) {
3 // 创建 leader elector 实例 le
4 le, err := NewLeaderElector(lec)
5 if err != nil {
6 panic(err)
7 }
8 ...
9
10 // 运行 leader elector 实例
11 le.Run(ctx)
12}
进入 LeaderElector.Run：
1// k8s.io/client-go/tools/leaderelection/leaderelection.go
2func (le *LeaderElector) Run(ctx context.Context) {
3 ...
4
5 if !le.acquire(ctx) {
6 return // ctx signalled done
7 }
8
9 ...
10
11 le.renew(ctx)
12}
LeaderElector.Run 中有两个重要的函数 LeaderElector.acquire 和 LeaderElector.renew。LeaderElector.acquire 负责获取（抢占）锁资源。LeaderElector.renew 负责更新锁资源，从而一直持有锁。
进入 LeaderElector.acquire：
1// k8s.io/client-go/tools/leaderelection/leaderelection.go
2func (le *LeaderElector) acquire(ctx context.Context) bool {
3 ...
4 succeeded := false
5 desc := le.config.Lock.Describe()
6 klog.Infof("attempting to acquire leader lease %v...", desc)
7
8 wait.JitterUntil(func() {
9 // 重点是这里，tryAcquireOrRenew 负责获取锁，或者更新已经获取锁的信息
10 succeeded = le.tryAcquireOrRenew(ctx)
11 le.maybeReportTransition()
12 if !succeeded {
13 klog.V(4).Infof("failed to acquire lease %v", desc)
14 return
15 }
16 le.config.Lock.RecordEvent("became leader")
17 le.metrics.leaderOn(le.config.Name)
18 klog.Infof("successfully acquired lease %v", desc)
19 cancel()
20 }, le.config.RetryPeriod, JitterFactor, true, ctx.Done())
21 return succeeded
22}
LeaderElector.acquire 函数的重点是 LeaderElector.tryAcquireOrRenew：
1// k8s.io/client-go/tools/leaderelection/leaderelection.go
2func (le *LeaderElector) tryAcquireOrRenew(ctx context.Context) bool {
3 now := metav1.NewTime(le.clock.Now())
4 leaderElectionRecord := rl.LeaderElectionRecord{
5 HolderIdentity: le.config.Lock.Identity(),
6 LeaseDurationSeconds: int(le.config.LeaseDuration / time.Second),
7 RenewTime: now,
8 AcquireTime: now,
9 }
10
11 // 1. obtain or create the ElectionRecord
12 oldLeaderElectionRecord, oldLeaderElectionRawRecord, err := le.config.Lock.Get(ctx)
13 if err != nil {
14 if !errors.IsNotFound(err) {
15 klog.Errorf("error retrieving resource lock %v: %v", le.config.Lock.Describe(), err)
16 return false
17 }
18 if err = le.config.Lock.Create(ctx, leaderElectionRecord); err != nil {
19 klog.Errorf("error initially creating leader election record: %v", err)
20 return false
21 }
22
23 le.setObservedRecord(&leaderElectionRecord)
24
25 return true
26 }
27
28 ...
29 return true
30}
LeaderElector.tryAcquireOrRenew 实现的就是状态转移图中的逻辑，这里我们仅关注创建锁这一流程。
首先，LeaderElector.config.Lock.Get 通过 client-go 获取 etcd 中锁的信息。
1// k8s.io/client-go/tools/leaderelection/resourcelock/leaselock.go
2func (ll *LeaseLock) Get(ctx context.Context) (*LeaderElectionRecord, []byte, error) {
3 // 获取锁（lease）
4 lease, err := ll.Client.Leases(ll.LeaseMeta.Namespace).Get(ctx, ll.LeaseMeta.Name, metav1.GetOptions{})
5 if err != nil {
6 return nil, nil, err
7 }
8
9 ...
10 return record, recordByte, nil
11}
没有找到锁，进入 LeaderElector.config.Lock.Create 创建锁。
1// k8s.io/client-go/tools/leaderelection/resourcelock/leaselock.go
2func (ll *LeaseLock) Create(ctx context.Context, ler LeaderElectionRecord) error {
3 var err error
4 ll.lease, err = ll.Client.Leases(ll.LeaseMeta.Namespace).Create(ctx, &coordinationv1.Lease{
5 ObjectMeta: metav1.ObjectMeta{
6 Name: ll.LeaseMeta.Name,
7 Namespace: ll.LeaseMeta.Namespace,
8 },
9 Spec: LeaderElectionRecordToLeaseSpec(&ler),
10 }, metav1.CreateOptions{})
11 return err
12}
查看创建锁的信息：
1# kubectl describe lease example -n default
2Name: example
3Namespace: default
4Labels: <none>
5Annotations: <none>
6API Version: coordination.k8s.io/v1
7Kind: Lease
8Metadata:
9 ...
10Spec:
11 Acquire Time: 2024-02-23T05:42:07.781552Z
12 Holder Identity: 1
13 Lease Duration Seconds: 60
14 Lease Transitions: 0
15 Renew Time: 2024-02-23T05:42:07.781552Z
更新 lease
接着看状态转移图中的更新 lease 的流程，场景是 lease 资源已存在并过期，holder 是 id 为 1 的进程。
进入 LeaderElector.tryAcquireOrRenew：
1func (le *LeaderElector) tryAcquireOrRenew(ctx context.Context) bool {
2 ...
3 // 1. obtain or create the ElectionRecord
4 oldLeaderElectionRecord, oldLeaderElectionRawRecord, err := le.config.Lock.Get(ctx)
5 if err != nil {
6 ...
7 }
8
9 // 2. Record obtained, check the Identity & Time
10 if !bytes.Equal(le.observedRawRecord, oldLeaderElectionRawRecord) {
11 le.setObservedRecord(oldLeaderElectionRecord)
12
13 le.observedRawRecord = oldLeaderElectionRawRecord
14 }
15 if len(oldLeaderElectionRecord.HolderIdentity) > 0 &&
16 le.observedTime.Add(time.Second*time.Duration(oldLeaderElectionRecord.LeaseDurationSeconds)).After(now.Time) &&
17 !le.IsLeader() {
18 klog.V(4).Infof("lock is held by %v and has not yet expired", oldLeaderElectionRecord.HolderIdentity)
19 return false
20 }
21
22 // 3. We're going to try to update. The leaderElectionRecord is set to it's default
23 // here. Let's correct it before updating.
24 if le.IsLeader() {
25 leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTime
26 leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions
27 } else {
28 leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + 1
29 }
30
31 // update the lock itself
32 if err = le.config.Lock.Update(ctx, leaderElectionRecord); err != nil {
33 klog.Errorf("Failed to update lock: %v", err)
34 return false
35 }
36}
在 LeaderElector.tryAcquireOrRenew 中获取锁，记录锁信息，比较锁 holder 是否是当前请求的 holder。如果是，记录锁的 AcquireTime 和 LeaderTransitions。最后，进入 LeaderElector.config.Lock.Update 更新锁：
1func (ll *LeaseLock) Update(ctx context.Context, ler LeaderElectionRecord) error {
2 ...
3
4 lease, err := ll.Client.Leases(ll.LeaseMeta.Namespace).Update(ctx, ll.lease, metav1.UpdateOptions{})
5 if err != nil {
6 return err
7 }
8
9 ll.lease = lease
10 return nil
11}
查看更新的 lease 信息：
1# kubectl describe lease example -n default
2Name: example
3Namespace: default
4Labels: <none>
5Annotations: <none>
6API Version: coordination.k8s.io/v1
7Kind: Lease
8Metadata:
9 ...
10Spec:
11 Acquire Time: 2024-02-23T05:42:07.781552Z
12 Holder Identity: 1
13 Lease Duration Seconds: 60
14 Lease Transitions: 0
15 Renew Time: 2024-02-23T06:00:52.802923Z
可以看到，holder 更新了 Renew Time，并且这里的 Lease Transitions 未更新。
renew lease
当 holder 拿到 lease 后需要 renew lease，以一直持有 lease。renew 是在获取锁之后，实现如下：
1func (le *LeaderElector) Run(ctx context.Context) {
2 ...
3
4 if !le.acquire(ctx) {
5 return // ctx signalled done
6 }
7
8 ...
9 le.renew(ctx)
10}
11
12func (le *LeaderElector) renew(ctx context.Context) {
13 ...
14 wait.Until(func() {
15 ...
16 err := wait.PollImmediateUntil(le.config.RetryPeriod, func() (bool, error) {
17 // 获取并且 renew lease
18 return le.tryAcquireOrRenew(timeoutCtx), nil
19 }, timeoutCtx.Done())
20
21 le.maybeReportTransition()
22 desc := le.config.Lock.Describe()
23 if err == nil {
24 klog.V(5).Infof("successfully renewed lease %v", desc)
25 return
26 }
27 le.metrics.leaderOff(le.config.Name)
28 klog.Infof("failed to renew lease %v: %v", desc, err)
29 cancel()
30 }, le.config.RetryPeriod, ctx.Done())
31
32 // if we hold the lease, give it up
33 if le.config.ReleaseOnCancel {
34 le.release()
35 }
36}
renew lease 也是通过 LeaderElector.tryAcquireOrRenew 更新 lease 信息实现的，这里就不多赘述了。
kube-scheduler 和 leader election
kube-scheduler 中调度和 leader election 结合的部分在 kube-schduler 的 Run 函数：
1func Run(ctx context.Context, cc *schedulerserverconfig.CompletedConfig, sched *scheduler.Scheduler) error {
2 ...
3 // If leader election is enabled, runCommand via LeaderElector until done and exit.
4 // 配置 leader election
5 if cc.LeaderElection != nil {
6 cc.LeaderElection.Callbacks = leaderelection.LeaderCallbacks{
7 // OnStartedLeading 回调函数，当抢占到 lease 时回调
8 OnStartedLeading: func(ctx context.Context) {
9 ...
10 // 作为 leader 运行调度器的调度逻辑
11 sched.Run(ctx)
12 },
13 OnStoppedLeading: func() {
14 ...
15 }
16 },
17 }
18
19 leaderElector, err := leaderelection.NewLeaderElector(*cc.LeaderElection)
20 if err != nil {
21 return fmt.Errorf("couldn't create leader elector: %v", err)
22 }
23
24 // 运行 leader election 抢占 lease
25 leaderElector.Run(ctx)
26
27 return fmt.Errorf("lost lease")
28 }
29
30 // Leader election is disabled, so runCommand inline until done.
31 close(waitingForLeader)
32 sched.Run(ctx)
33 return fmt.Errorf("finished without leader elect")
34}
小结
本文从源码角度分析 leader election 的实现，并且介绍了 leader election 和 kube-scheduler 结合实现是怎么实现多副本，单实例运行的。