• 欢迎访问V小白网站,推荐使用最新版火狐浏览器和Chrome浏览器访问本网站,欢迎加入V小白 暂无QQ群
  • 欢迎访问V小白网站,这是第二个滚动条

kubernets 中组件高可用的实现方式

其他分类 admin 2019年6月27日 10:46 62 0 个评论

生产环境中为了保障业务的稳定性,集群都需要高可用部署,k8s 中 apiserver 是无状态的,可以横向扩容保证其高可用,kube-controller-manager 和 kube-scheduler 两个组件通过 leader 选举保障高可用,即正常情况下 kube-scheduler 或 kube-manager-controller 组件的多个副本只有一个是处于业务逻辑运行状态,其它副本则不断的尝试去获取锁,去竞争 leader,直到自己成为leader。如果正在运行的 leader 因某种原因导致当前进程退出,或者锁丢失,则由其它副本去竞争新的 leader,获取 leader 继而执行业务逻辑。

kubernetes 中查看组件 leader 的方法:

$ kubectl get endpoints kube-controller-manager --namespace=kube-system -o yaml && 
  kubectl get endpoints kube-scheduler --namespace=kube-system -o yaml

Leader Election 的实现

Leader Election 的过程本质上是一个竞争分布式锁的过程。在 Kubernetes 中,这个分布式锁是以创建 Endpoint 资源的形式进行,谁先创建了该资源,谁就先获得锁,之后会对该资源不断更新以保持锁的拥有权。

  • 1、初始化资源锁,kubernetes 中默认的资源锁使用 endpoints,也就是 c.ComponentConfig.Generic.LeaderElection.ResourceLock 的值为 "endpoints",在代码中我并没有找到对 ResourceLock 初始化的地方,只看到了对该参数的说明以及日志中配置的默认值:

​在初始化资源锁的时候还传入了 EventRecorder,其作用是当 leader 发生变化的时候会将对应的 events 发送到 apiserver。

func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {
    controllers := map[string]InitFunc{}
    controllers["endpoint"] = startEndpointController
    controllers["replicationcontroller"] = startReplicationController
    controllers["podgc"] = startPodGCController
    controllers["resourcequota"] = startResourceQuotaController
    controllers["namespace"] = startNamespaceController
    controllers["serviceaccount"] = startServiceAccountController
    controllers["garbagecollector"] = startGarbageCollectorController
    controllers["daemonset"] = startDaemonSetController
    controllers["job"] = startJobController
    controllers["deployment"] = startDeploymentController
    controllers["replicaset"] = startReplicaSetController
    controllers["horizontalpodautoscaling"] = startHPAController
    controllers["disruption"] = startDisruptionController
    controllers["statefulset"] = startStatefulSetController
    controllers["cronjob"] = startCronJobController
    controllers["csrsigning"] = startCSRSigningController
    controllers["csrapproving"] = startCSRApprovingController
    controllers["csrcleaner"] = startCSRCleanerController
    controllers["ttl"] = startTTLController
    controllers["bootstrapsigner"] = startBootstrapSignerController
    controllers["tokencleaner"] = startTokenCleanerController
    controllers["nodeipam"] = startNodeIpamController
    if loopMode == IncludeCloudLoops {
        controllers["service"] = startServiceController
        controllers["route"] = startRouteController
    }
    controllers["nodelifecycle"] = startNodeLifecycleController
    controllers["persistentvolume-binder"] = startPersistentVolumeBinderController
    controllers["attachdetach"] = startAttachDetachController
    controllers["persistentvolume-expander"] = startVolumeExpandController
    controllers["clusterrole-aggregation"] = startClusterRoleAggregrationController
    controllers["pvc-protection"] = startPVCProtectionController
    controllers["pv-protection"] = startPVProtectionController
    controllers["ttl-after-finished"] = startTTLAfterFinishedController

    return controllers
}

在 RunOrDie 中首先调用 NewLeaderElector 初始化了一个 LeaderElector 对象,然后执行 LeaderElector 的 run 方法进行选举。

func (le *LeaderElector) Run(ctx context.Context) {
    defer func() {
        runtime.HandleCrash()
        le.config.Callbacks.OnStoppedLeading()
    }()
    if !le.acquire(ctx) {
        return // ctx signalled done
    }
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()
    go le.config.Callbacks.OnStartedLeading(ctx)
    le.renew(ctx)
}

在 acquire 中首先初始化了一个 ctx,通过 wait.JitterUntil 周期性的去调用 le.tryAcquireOrRenew 方法来获取资源锁,直到获取为止。如果获取不到锁,则会以 RetryPeriod 为间隔不断尝试。如果获取到锁,就会关闭 ctx 通知 wait.JitterUntil 停止尝试,tryAcquireOrRenew 是最核心的方法。

func (le *LeaderElector) tryAcquireOrRenew() 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, err := le.config.Lock.Get()
    if err != nil {
        if !errors.IsNotFound(err) {
            glog.Errorf("error retrieving resource lock %v: %v", le.config.Lock.Describe(), err)
            return false
        }
        // 没有获取到资源锁,开始创建资源锁,若创建成功则成为 leader 
        if err = le.config.Lock.Create(leaderElectionRecord); err != nil {
            glog.Errorf("error initially creating leader election record: %v", err)
            return false
        }
        le.observedRecord = leaderElectionRecord
        le.observedTime = le.clock.Now()
        return true
    }

    // 2、获取资源锁后检查当前 id 是不是 leader
    if !reflect.DeepEqual(le.observedRecord, *oldLeaderElectionRecord) {
        le.observedRecord = *oldLeaderElectionRecord
        le.observedTime = le.clock.Now()
    }
    // 如果资源锁没有过期且当前 id 不是 Leader,直接返回
    if le.observedTime.Add(le.config.LeaseDuration).After(now.Time) &&
        !le.IsLeader() {
        glog.V(4).Infof("lock is held by %v and has not yet expired", oldLeaderElectionRecord.HolderIdentity)
        return false
    }

    // 3、如果当前 id 是 Leader,将对应字段的时间改成当前时间,准备续租
    // 如果是非 Leader 节点则抢夺资源锁
    if le.IsLeader() {
        leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTime
        leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions
    } else {
        leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + 1
    }

    // 更新资源
    // 对于 Leader 来说,这是一个续租的过程
    // 对于非 Leader 节点(仅在上一个资源锁已经过期),这是一个更新锁所有权的过程
    if err = le.config.Lock.Update(leaderElectionRecord); err != nil {
        glog.Errorf("Failed to update lock: %v", err)
        return false
    }
    le.observedRecord = leaderElectionRecord
    le.observedTime = le.clock.Now()
    return true
}
  • 1、获取 ElectionRecord 记录,如果没有则创建一条新的 ElectionRecord 记录,创建成功则表示获取到锁并成为 leader 了。

  • 2、当获取到资源锁后开始检查其中的信息,比较当前 id 是不是 leader 以及资源锁有没有过期,如果资源锁没有过期且当前 id 不是 Leader,则直接返回。

  • 3、如果当前 id 是 Leader,将对应字段的时间改成当前时间,更新资源锁进行续租。

  • 4、如果当前 id 不是 Leader 但是资源锁已经过期了,则抢夺资源锁,抢夺成功则成为 leader 否则返回。

最后是 renew 方法:

以下是一个 demo,使用 k8s 中 k8s.io/client-go/tools/leaderelection 进行一个演示:

package main

import (
    "context"
    "flag"
    "fmt"
    "os"
    "time"

    "github.com/golang/glog"
    "k8s.io/api/core/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/kubernetes/scheme"
    v1core "k8s.io/client-go/kubernetes/typed/core/v1"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/client-go/tools/leaderelection"
    "k8s.io/client-go/tools/leaderelection/resourcelock"
    "k8s.io/client-go/tools/record"
)

var (
    masterURL  string
    kubeconfig string
)

func init() {
    flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
    flag.StringVar(&masterURL, "master", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")

    flag.Set("logtostderr", "true")
}

func main() {
    flag.Parse()
    defer glog.Flush()

    id, err := os.Hostname()
    if err != nil {
        panic(err)
    }

    // 加载 kubeconfig 配置
    cfg, err := clientcmd.BuildConfigFromFlags(masterURL, kubeconfig)
    if err != nil {
        glog.Fatalf("Error building kubeconfig: %s", err.Error())
    }

    // 创建 kubeclient
    kubeClient, err := kubernetes.NewForConfig(cfg)
    if err != nil {
        glog.Fatalf("Error building kubernetes clientset: %s", err.Error())
    }

    // 初始化 eventRecorder
    eventBroadcaster := record.NewBroadcaster()
    eventRecorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "test-1"})
    eventBroadcaster.StartLogging(glog.Infof)
    eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})

    run := func(ctx context.Context) {
        fmt.Println("run.........")
        select {}
    }

    id = id + "_" + "1"
    rl, err := resourcelock.New("endpoints",
        "kube-system",
        "test",
        kubeClient.CoreV1(),
        resourcelock.ResourceLockConfig{
            Identity:      id,
            EventRecorder: eventRecorder,
        })
    if err != nil {
        glog.Fatalf("error creating lock: %v", err)
    }

    leaderelection.RunOrDie(context.TODO(), leaderelection.LeaderElectionConfig{
        Lock:          rl,
        LeaseDuration: 15 * time.Second,
        RenewDeadline: 10 * time.Second,
        RetryPeriod:   2 * time.Second,
        Callbacks: leaderelection.LeaderCallbacks{
            OnStartedLeading: run,
            OnStoppedLeading: func() {
                glog.Info("leaderelection lost")
            },
        },
        Name: "test-1",
    })
}

总结

本文讲述了 kube-controller-manager 使用 HA 的方式启动后 leader 选举过程的实现说明,k8s 中通过创建 endpoints 资源以及对该资源的持续更新来实现资源锁轮转的过程。但是相对于其他分布式锁的实现,普遍是直接基于现有的中间件实现,比如 redis、zookeeper、etcd 等,其所有对锁的操作都是原子性的,那 k8s 选举过程中的原子操作是如何实现的?k8s 中的原子操作最终也是通过 etcd 实现的,其在做 update 更新锁的操作时采用的是乐观锁,通过对比 resourceVersion 实现的,详细的实现下节再讲。

参考文档:
API OVERVIEW
Simple leader election with Kubernetes and Docker


V小白 , 版权所有丨如未注明 , 均为原创丨本网站采用BY-NC-SA协议进行授权
转载请注明原文链接:kubernets 中组件高可用的实现方式
喜欢 (0)
发表我的评论
取消评论
表情 贴图 加粗 删除线 居中 斜体 签到
0个小伙伴在吐槽