好好学K8S：K8S中的Leader Election机制

1. 需求描述

假设有一个基于client-go的程序，叫做watcher，会监听k8s集群中pod被删除的消息，当pod被删除时，会触发执行一个动作。
当watcher只有一个副本时，程序运行符合预期。但是当watcher有多个副本时，多个watcher副本都监听到pod被删除的消息，都会触发执行一个动作。而这个动作，我们希望只执行一次。
有什么办法，可以让多个watcher具备多个副本，但是当监听到pod被删除时，只会触发一次执行动作？

2. 实现思路

要实现这个需求，有三个思路：

思路一：基于K8S ValidatingAdmissionWebhook机制。实现一个ValidatingAdmissionWebhook，限制watcher服务只能有一个副本。watcher服务只有一个副本，自然就只会执行一次动作，但是，与需求不相符，不能高可用。
思路二：基于分布式锁。常见的实现方式包括使用 redis、zookeeper 或 etcd 等工具。这种实现方式简单，但是需要额外维护一个中间件，不够优雅。
思路三：基于K8S Leader Election机制。K8S 提供了 Leader Election 机制，可以通过 client-go 库实现，只有被选为 Leader 的 watcher 才会处理 pod 删除事件。

本文中选择思路三（K8S Leader Election机制）来实现需求：服务多个副本但是只执行一次动作。

参考文档：

3.1. Leader Election原理

Leader Election 是一种分布式系统中的机制，旨在确保在多个候选者中选出一个“领导者”进程，以负责执行特定的操作。

1、候选者识别：在 Leader Election 中，首先需要识别出一组候选者，这些候选者可能是运行在同一集群中的多个实例（如 Pods）。这些候选者会竞争成为领导者。

2、竞选过程：候选者通过某种方式（如心跳信号）宣告自己为领导者。通常，所有候选者会尝试同时声明自己为领导者。其中一个候选者成功地获得领导权，而其他候选者则进入待命状态，准备在当前领导者失效时进行新的竞选。

3、心跳机制：一旦某个实例成为领导者，它会定期发送心跳信号以维持其领导地位。如果领导者未能在预定时间内发送心跳信号，其他候选者将启动新的竞选过程，以确保始终有一个活跃的领导者。

4、故障恢复：当当前领导者发生故障或被终止时，其他候选者会迅速重新进行竞选，以确定新的领导者。这种机制保证了系统的高可用性。

4. K8S 中的 Leader Election 实现

Kubernetes 提供了一种简化的方式来实现 Leader Election，相关概念包括资源锁和Lease API。

资源锁：Kubernetes 使用 ConfigMap 或 Lease 等资源作为锁来管理领导权。每个候选者尝试更新这个资源以声明自己为领导者。例如，通过更新 ConfigMap 中的某个字段来表示当前的领导者。
Lease API：Kubernetes 从 v1.14 开始引入了 coordination.k8s.io API，允许更高效地管理领导权。使用 Lease 对象可以减少对 API 的调用频率，并避免过多的事件通知。

选举过程：
1、候选者创建或获取 Lease 对象，并尝试更新其内容以表明其身份。
2、只有第一个成功更新 Lease 的实例能够获得领导权。
3、其他实例在发现 Lease 被更新后，将停止尝试并进入待命状态。

选举机制保证了控制器的高可用，同时只有一个控制器为主，其他为从，防止同个事件被多次重复监听，重复执行相关的业务逻辑。

5. Leader Election示例代码

示例代码地址：examples - leader-election

1、创建示例项目

1 2	mkdir leader-election-demo && cd leader-election-demo go mod init leader-election-demo

2、粘贴示例代码
创建文件 main.go ，并且把示例代码粘贴进去。

















package main

import (
    "context"
    "flag"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/google/uuid"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    clientset "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/rest"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/client-go/tools/leaderelection"
    "k8s.io/client-go/tools/leaderelection/resourcelock"
    "k8s.io/klog/v2"
)

func buildConfig(kubeconfig string) (*rest.Config, error) {
    if kubeconfig != "" {
        cfg, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
        if err != nil {
            return nil, err
        }
        return cfg, nil
    }

    cfg, err := rest.InClusterConfig()
    if err != nil {
        return nil, err
    }
    return cfg, nil
}

func main() {
    klog.InitFlags(nil)

    var kubeconfig string
    var leaseLockName string
    var leaseLockNamespace string
    var id string

    flag.StringVar(&kubeconfig, "kubeconfig", "", "absolute path to the kubeconfig file")
    flag.StringVar(&id, "id", uuid.New().String(), "the holder identity name")
    flag.StringVar(&leaseLockName, "lease-lock-name", "", "the lease lock resource name")
    flag.StringVar(&leaseLockNamespace, "lease-lock-namespace", "", "the lease lock resource namespace")
    flag.Parse()

    if leaseLockName == "" {
        klog.Fatal("unable to get lease lock resource name (missing lease-lock-name flag).")
    }
    if leaseLockNamespace == "" {
        klog.Fatal("unable to get lease lock resource namespace (missing lease-lock-namespace flag).")
    }

    
    
    
    
    
    config, err := buildConfig(kubeconfig)
    if err != nil {
        klog.Fatal(err)
    }
    client := clientset.NewForConfigOrDie(config)

    run := func(ctx context.Context) {
        
        klog.Info("Controller loop...")

        select {}
    }

    
    
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    
    
    
    ch := make(chan os.Signal, 1)
    signal.Notify(ch, os.Interrupt, syscall.SIGTERM)
    go func() {
        <-ch
        klog.Info("Received termination, signaling shutdown")
        cancel()
    }()

    
    
    lock := &resourcelock.LeaseLock{
        LeaseMeta: metav1.ObjectMeta{
            Name:      leaseLockName,
            Namespace: leaseLockNamespace,
        },
        Client: client.CoordinationV1(),
        LockConfig: resourcelock.ResourceLockConfig{
            Identity: id,
        },
    }

    
    leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
        Lock: lock,
        
        
        
        
        
        
        ReleaseOnCancel: true,
        LeaseDuration:   60 * time.Second,
        RenewDeadline:   15 * time.Second,
        RetryPeriod:     5 * time.Second,
        Callbacks: leaderelection.LeaderCallbacks{
            OnStartedLeading: func(ctx context.Context) {
                
                
                run(ctx)
            },
            OnStoppedLeading: func() {
                
                klog.Infof("leader lost: %s", id)
                os.Exit(0)
            },
            OnNewLeader: func(identity string) {
                
                if identity == id {
                    
                    return
                }
                klog.Infof("new leader elected: %s", identity)
            },
        },
    })
}

3、安装依赖

1	go mod tidy

4、运行代码


go run main.go -kubeconfig=/path/to/kubeconfig -logtostderr=true -lease-lock-name=example -lease-lock-namespace=default -id=1


go run main.go -kubeconfig=/path/to/kubeconfig -logtostderr=true -lease-lock-name=example -lease-lock-namespace=default -id=2


go run main.go -kubeconfig=/path/to/kubeconfig -logtostderr=true -lease-lock-name=example -lease-lock-namespace=default -id=3

5、查看lease

1	kubectl get lease

看到如下内容：

1 2	NAME HOLDER AGE example 1 104s

6、结束1号进程，再次查看lease
看到如下内容：

1 2	NAME HOLDER AGE example 3 3m12s

上面的示例，可以证明Leader Election机制已经生效了。当1号进程不可用时，另外的两个进程其中之一会成为领导者。