47｜Leader Election：基于Lease的Leader Election源码剖析

孔令飞

你好，我是孔令飞。
上一节课，我详细介绍了 Kubernetes 中 Leader Election  机制的原理，以及 kube-controller-manager 具体是如何通过 Leader Election 来实现多副本容灾的。
本节课，我将详细介绍如何在自己的项目中使用 Kubernetes 的 Leader Election 能力开发竞态资源处理组件，并结合实战示例与剖析源码，讲解基于 Lease 资源的 Leader Election 具体实现。
Leader Election 使用实战Kubernetes client-go 项目仓库中，其实已经有一个优秀的 Leader Election 使用示例：examples/leader-election/main.go，示例代码如下：
/*
该示例来自于：https://github.com/kubernetes/client-go/blob/master/examples/leader-election/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) {
    // 如果指定了 kubeconfig，则使用指定的 kubeconfg 构建客户端
    if kubeconfig != "" {
        cfg, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
        if err != nil {
            return nil, err
        }
        return cfg, nil
    }
    // 如果程序部署在 Kubernetes 集群内，则使用 In-Cluster Config
    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).")
    }
    // leader election uses the Kubernetes API by writing to a
    // lock object, which can be a LeaseLock object (preferred),
    // a ConfigMap, or an Endpoints (deprecated) object.
    // Conflicting writes are detected and each client handles those actions
    // independently.
    config, err := buildConfig(kubeconfig)
    if err != nil {
        klog.Fatal(err)
    }
    // 创建 Kubernetes 客户端实例，用来同 kube-apiserver 进行交互
    client := clientset.NewForConfigOrDie(config)
    // 执行程序的主业务逻辑
    run := func(ctx context.Context) {
        // complete your controller loop here
        klog.Info("Controller loop...")
        select {}
    }
    // use a Go context so we can tell the leaderelection code when we
    // want to step down
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    // listen for interrupts or the Linux SIGTERM signal and cancel
    // our context, which the leader election code will observe and
    // step down
    // 优雅关停实现
    ch := make(chan os.Signal, 1)
    signal.Notify(ch, os.Interrupt, syscall.SIGTERM)
    go func() {
        <-ch
        klog.Info("Received termination, signaling shutdown")
        cancel()
    }()
    // we use the Lease lock type since edits to Leases are less common
    // and fewer objects in the cluster watch "all Leases".
    // 指定锁的资源对象，这里使用了 Lease 资源，还支持 configmap，endpoint，或者 multilock（即多种配合使用）
    lock := &resourcelock.LeaseLock{
        LeaseMeta: metav1.ObjectMeta{
            Name:      leaseLockName,
            Namespace: leaseLockNamespace,
        },
        Client: client.CoordinationV1(),
        LockConfig: resourcelock.ResourceLockConfig{
            Identity: id,
        },
    }
    // start the leader election code loop
    // 开始 Leader Election 循环
    leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
        Lock: lock,
        // IMPORTANT: you MUST ensure that any code you have that
        // is protected by the lease must terminate **before**
        // you call cancel. Otherwise, you could have a background
        // loop still running and another process could
        // get elected before your background loop finished, violating
        // the stated goal of the lease.
        ReleaseOnCancel: true,
        LeaseDuration:   60 * time.Second, // Leader 持有租约的最大时间
        RenewDeadline:   15 * time.Second, // Leader 续租租约的超时时间
        RetryPeriod:     5 * time.Second,  // 候选节点重试时间
        Callbacks: leaderelection.LeaderCallbacks{
            OnStartedLeading: func(ctx context.Context) {
                // we're notified when we start - this is where you would
                // usually put your code
                // 变为 Leader 后需要执行的业务代码
                run(ctx)
            },
            OnStoppedLeading: func() {
                // we can do cleanup here
                // 打印日志，并退出进程
                klog.Infof("leader lost: %s", id)
                os.Exit(0)
            },
            OnNewLeader: func(identity string) {
                // we're notified when new leader elected
                // 当产生新的 Leader 后执行的业务逻辑
                if identity == id {
                    // I just got the lock
                    return
                }
                klog.Infof("new leader elected: %s", identity)
            },
        },
    })
}

公开

同步至部落

取消

完成

0/2000

荧光笔

直线

曲线

笔记

复制

AI

深入了解
翻译
英语
中文简体
法语
德语
日语
韩语
俄语
西班牙语
解释
总结

1. Kubernetes中的Leader Election机制是实现多副本容灾的关键，可以帮助开发者在自己的项目中使用Kubernetes的Leader Election能力开发竞态资源处理组件。 2. 本节课重点介绍了基于Lease资源的Leader Election具体实现，结合实战示例和源码剖析，帮助读者深入理解Leader Election的实际应用和实现原理。 3. 通过详细介绍如何在自己的项目中使用Kubernetes的Leader Election能力开发竞态资源处理组件，读者可以学习到实际的应用技巧和开发方法。 4. 本文帮助读者深入了解Kubernetes的Leader Election机制，为他们在实际项目中应用该机制提供指导和帮助。 5. 通过剖析源码，读者可以更加深入地理解基于Lease资源的Leader Election的具体实现细节，从而更好地应用于自己的项目中。 6. `LeaderElector`的`Run`方法实现了处理Panic、抢锁、执行主业务逻辑和定期更新的功能。 7. `acquire`方法通过调用`tryAcquireOrRenew`方法来抢锁，使用`wait.JitterUntil`函数定期调用`tryAcquireOrRenew`方法抢锁，直到成功为止。 8. `tryAcquireOrRenew`方法实现了核心的抢锁逻辑，包括更新Lease资源、获取或创建ElectionRecord、检查Identity和Time、更新锁资源对象等步骤。 9. 在`Run`方法中，抢锁成功后会在Go协程中调用`OnStartedLeading`方法来运行业务逻辑。 10. `renew`方法通过`wait.Until`方法定期调用`tryAcquireOrRenew`方法续约资源锁，保持自己的Leader位置。

仅可试看部分内容，如需阅读全部内容，请付费购买文章所属专栏
《Kubernetes 源码剖析与实战》，新⼈⾸单¥68

立即购买

登录后留言

精选留言

由作者筛选后的优质留言将会公开显示，欢迎踊跃留言。

收起评论