本文主要是介绍k8s中基于资源锁的选主分析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
简介
k8s中为了实现高可用，需要部署多个副本，例如多个apiserver、scheduler、controller-manager等，其中apiserver是无状态的每个组件都可以工作，而scheduler与controller-manager是有状态的，同一时刻只能存在一个活跃的，需要进行选主。
k8s使用了资源锁（endpoints/configmap/lease）的方式来实现选主，多个副本去创建资源，创建成功则获得锁成为leader，leader在租约内去刷新锁，其他副本则通过比对锁的更新时间判断是否成为新的leader。
k8s采用了资源版本号的乐观锁方式来实现选主，对比etcd选主，效率更高，并发性更好。
源码分析
k8s选主实现在client-go中，包k8s.io/client-go/tools/leaderelection
结构定义
锁结构定义如下：
// k8s.io/client-go/tools/leaderelection/resourcelock/interface.go
type LeaderElectionRecord struct {// leader 标识，通常为 hostnameHolderIdentity string `json:"holderIdentity"`// 同启动参数 --leader-elect-lease-durationLeaseDurationSeconds int `json:"leaseDurationSeconds"`// Leader 第一次成功获得租约时的时间戳AcquireTime unversioned.Time `json:"acquireTime"`// leader 定时 renew 的时间戳RenewTime unversioned.Time `json:"renewTime"`LeaderTransitions int `json:"leaderTransitions"`
}
k8s中的选举锁需实现resourcelock.Interface接口，基本上实现CRU，将leader信息存在在annotation中
// k8s.io/client-go/tools/leaderelection/resourcelock/interface.go
type Interface interface {// Get returns the LeaderElectionRecordGet() (*LeaderElectionRecord, []byte, error)// Create attempts to create a LeaderElectionRecordCreate(ler LeaderElectionRecord) error// Update will update and existing LeaderElectionRecordUpdate(ler LeaderElectionRecord) error// RecordEvent 记录锁切换事件RecordEvent(string)// Identity will return the locks IdentityIdentity() string// Describe is used to convert details on current resource lock// into a stringDescribe() string
}
创建资源锁
锁类型包括：configmaps， endpoints, lease, 以及 multiLock
// k8s.io/client-go/tools/leaderelection/resourcelock/interface.go
func New(lockType string, ns string, name string, coreClient corev1.CoreV1Interface, coordinationClient coordinationv1.CoordinationV1Interface, rlc ResourceLockConfig) (Interface, error) {endpointsLock := &EndpointsLock{EndpointsMeta: metav1.ObjectMeta{Namespace: ns,Name: name,},Client: coreClient,LockConfig: rlc,}configmapLock := &ConfigMapLock{ConfigMapMeta: metav1.ObjectMeta{Namespace: ns,Name: name,},Client: coreClient,LockConfig: rlc,}leaseLock := &LeaseLock{LeaseMeta: metav1.ObjectMeta{Namespace: ns,Name: name,},Client: coordinationClient,LockConfig: rlc,}switch lockType {case EndpointsResourceLock:return endpointsLock, nilcase ConfigMapsResourceLock:return configmapLock, nilcase LeasesResourceLock:return leaseLock, nilcase EndpointsLeasesResourceLock:return &MultiLock{Primary: endpointsLock,Secondary: leaseLock,}, nilcase ConfigMapsLeasesResourceLock:return &MultiLock{Primary: configmapLock,Secondary: leaseLock,}, nildefault:return nil, fmt.Errorf("Invalid lock-type %s", lockType)}
}
使用者首先通过new()函数创建资源锁，需要提供锁类型、namespace、name、唯一标示等。
进行选举
创建选举配置，通常如下：
// start the leader election code loopleaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{// 资源锁类型Lock: lock,// 租约时长，非主候选者用来判断资源锁是否过期LeaseDuration: 60 * time.Second,// leader刷新资源锁超时时间 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)},},})
创建选举对象后，执行Run函数开始选主
// k8s.io/client-go/tools/leaderelection/leaderelection.go
// Run starts the leader election loop
func (le *LeaderElector) Run(ctx context.Context) {defer func() {runtime.HandleCrash()// 锁丢失时执行OnStoppedLeading回调函数le.config.Callbacks.OnStoppedLeading()}()// 尝试获得锁if !le.acquire(ctx) {return // ctx signalled done}ctx, cancel := context.WithCancel(ctx)defer cancel()// 获得锁后执行OnStartedLeading回调函数go le.config.Callbacks.OnStartedLeading(ctx)// 定期刷新锁le.renew(ctx)
}
acruire方法：
// k8s.io/client-go/tools/leaderelection/leaderelection.go
// acquire loops calling tryAcquireOrRenew and returns true immediately when tryAcquireOrRenew succeeds.
// Returns false if ctx signals done.
func (le *LeaderElector) acquire(ctx context.Context) bool {ctx, cancel := context.WithCancel(ctx)defer cancel()succeeded := falsedesc := le.config.Lock.Describe()klog.Infof("attempting to acquire leader lease %v...", desc)// 调用 JitterUntil 函数，以 RetryPeriod 为间隔去刷新资源锁，直到获取锁wait.JitterUntil(func() {// tryAcquireOrRenew 方法去调度资源更新接口，判断是否刷新成功succeeded = le.tryAcquireOrRenew()le.maybeReportTransition()if !succeeded {klog.V(4).Infof("failed to acquire lease %v", desc)return}le.config.Lock.RecordEvent("became leader")le.metrics.leaderOn(le.config.Name)klog.Infof("successfully acquired lease %v", desc)cancel()}, le.config.RetryPeriod, JitterFactor, true, ctx.Done())return succeeded
}
renew方法，只有在获取锁之后才会调用，它会通过持续更新资源锁的数据，来确保继续持有已获得的锁，保持自己的leader 状态。
// renew loops calling tryAcquireOrRenew and returns immediately when tryAcquireOrRenew fails or ctx signals done.
func (le *LeaderElector) renew(ctx context.Context) {ctx, cancel := context.WithCancel(ctx)defer cancel()wait.Until(func() {timeoutCtx, timeoutCancel := context.WithTimeout(ctx, le.config.RenewDeadline)defer timeoutCancel()// err := wait.PollImmediateUntil(le.config.RetryPeriod, func() (bool, error) {done := make(chan bool, 1)go func() {defer close(done)done <- le.tryAcquireOrRenew()}()// 超时返回error, 否则返回更新结果select {case <-timeoutCtx.Done():return false, fmt.Errorf("failed to tryAcquireOrRenew %s", timeoutCtx.Err())case result := <-done:return result, nil}}, timeoutCtx.Done())le.maybeReportTransition()desc := le.config.Lock.Describe()if err == nil {klog.V(5).Infof("successfully renewed lease %v", desc)return}le.config.Lock.RecordEvent("stopped leading")le.metrics.leaderOff(le.config.Name)klog.Infof("failed to renew lease %v: %v", desc, err)cancel()}, le.config.RetryPeriod, ctx.Done())// if we hold the lease, give it upif le.config.ReleaseOnCancel {le.release()}
}
这里使用了wait包，wait.Until会不断的调用wait.PollImmediateUntil方法，前者是进行无限循环操作，直到 stop chan被关闭，wait.PollImmediateUntil则不断的对某一条件进行检查，以RetryPeriod为间隔，直到该条件返回true、error或者超时。这一条件是一个需要满足 func() (bool, error) 签名的方法，比如这个例子只是调用了 le.tryAcquireOrRenew()。
最后看下tryAcquireOrRenew方法：
// tryAcquireOrRenew tries to acquire a leader lease if it is not already acquired,
// else it tries to renew the lease if it has already been acquired. Returns true
// on success else returns false.
func (le *LeaderElector) tryAcquireOrRenew() bool {now := metav1.Now()// 这个 leaderElectionRecord 就是保存在 endpoint/configmap 的 annotation 中的值。// 每个节点都将 HolderIdentity 设置为自己，以及关于获取和更新锁的时间。后面会对时间进行修正，才会更新到 API serverleaderElectionRecord := rl.LeaderElectionRecord{HolderIdentity: le.config.Lock.Identity(),LeaseDurationSeconds: int(le.config.LeaseDuration / time.Second),RenewTime: now,AcquireTime: now,}// 1. 获取或者创建 ElectionRecordoldLeaderElectionRecord, oldLeaderElectionRawRecord, err := le.config.Lock.Get()if err != nil {// 记录不存在的话，则创建一条新的记录if !errors.IsNotFound(err) {klog.Errorf("error retrieving resource lock %v: %v", le.config.Lock.Describe(), err)return false}if err = le.config.Lock.Create(leaderElectionRecord); err != nil {klog.Errorf("error initially creating leader election record: %v", err)return false}// 创建记录成功，同时表示获得了锁，返回truele.observedRecord = leaderElectionRecordle.observedTime = le.clock.Now()return true}// 2. 正常获取了锁资源的记录，检查锁持有者和更新时间。if !bytes.Equal(le.observedRawRecord, oldLeaderElectionRawRecord) {// 记录之前的锁持有者le.observedRecord = *oldLeaderElectionRecordle.observedRawRecord = oldLeaderElectionRawRecordle.observedTime = le.clock.Now()}// 在满足以下所有的条件下，认为锁由他人持有，并且还没有过期，返回 false// a. 当前锁持有者的并非自己// b. 上一次观察时间 + 观测检查间隔大于现在时间，即距离上次观测的间隔，小于 `LeaseDuration` 的设置值。if len(oldLeaderElectionRecord.HolderIdentity) > 0 &&le.observedTime.Add(le.config.LeaseDuration).After(now.Time) &&!le.IsLeader() {klog.V(4).Infof("lock is held by %v and has not yet expired", oldLeaderElectionRecord.HolderIdentity)return false}// 3. 更新资源的资源锁if le.IsLeader() {leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTimeleaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions} else {leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + 1}// 调用资源锁更新接口if err = le.config.Lock.Update(leaderElectionRecord); err != nil {klog.Errorf("Failed to update lock: %v", err)return false}le.observedRecord = leaderElectionRecordle.observedTime = le.clock.Now()return true
}总结
当应用在k8s上部署时，使用k8s的资源锁，可方便的实现高可用，但需要注意以下几点：
- 推荐使用
configmap作为资源锁，原因是某些组件如kube-proxy会去监听endpoints来更新节点iptables规则，当有大量资源锁时，势必会对性能有影响。 - 当选举结束时调用
OnStoppedLeading需要退出程序(例如os.Exit(0))，若不退出程序，所有副本选举结束不会去竞争资源锁，就没有leader，造成服务不可用而这时程序并没有异常。需要执行退出逻辑，让Daemon程序k8s/systemd等重启服务来重新参与选主。
这篇关于k8s中基于资源锁的选主分析的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
