ericpuwang

高可用模式下的选主机制

选举原因

k8s为了保证kube-scheduler和kube-manager-controller组件的高可用性，每个组件本身通过多实例副本方式部署。但是由于多个实例同时使用informer去监听处理必然带来一致性或重复执行问题，故采用分布式锁(选举)机制规避该问题

k8s是通过leaderelection来实现选举，只有一个实例能拿到锁，只有拿到锁的实例可以运行，其他实例会不断重试(轮询)方式获取锁

选举配置

• LeaseDuration: 锁的时长，通过该值判断是否超时，超时则认定丢失锁
• RenewDeadline: 续约时长，每次lease ttl的时长
• RetryPeriod: 周期重试获取锁的间隔
• Callbacks: 自定义方法
  • OnStartedLeading func(context.Context): 作为leader启动时进行回调
  • OnStoppedLeading func(): 关闭时进行回调
  • OnNewLeader func(identity string): 转换为leader时进行回调

package main

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

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	clientset "k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/leaderelection"
	"k8s.io/client-go/tools/leaderelection/resourcelock"
	"k8s.io/klog"
)

var client *clientset.Clientset

func newLeaseLock(lockname, podname, namespace string) *resourcelock.LeaseLock {
	return &resourcelock.LeaseLock{
		LeaseMeta: metav1.ObjectMeta{
			Name:      lockname,
			Namespace: namespace,
		},
		Client: client.CoordinationV1(),
		LockConfig: resourcelock.ResourceLockConfig{
			Identity: podname,
		},
	}
}

func run(lock *resourcelock.LeaseLock, ctx context.Context, id string) {
	leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
		Lock:            lock,
		ReleaseOnCancel: true,
		LeaseDuration:   15 * time.Second,
		RenewDeadline:   10 * time.Second,
		RetryPeriod:     2 * time.Second,
		Callbacks: leaderelection.LeaderCallbacks{
			OnStartedLeading: func(c context.Context) {
				klog.Info("get identity", "leader: ", lock.LeaseMeta.Name)
				select {}
			},
			OnStoppedLeading: func() {
				klog.Info("no longer the leader, staying inactive.")
			},
			OnNewLeader: func(current_id string) {
				if current_id == id {
					klog.Info("still the leader!")
					return
				}
				klog.Info("new leader is %s", current_id)
			},
		},
	})
}

func main() {
	var (
		leaseLockName      string
		leaseLockNamespace string
		podName            = os.Getenv("POD_NAME")
	)
	flag.StringVar(&leaseLockName, "lease-name", "", "Name of lease lock")
	flag.StringVar(&leaseLockNamespace, "lease-namespace", "default", "Name of lease lock namespace")
	flag.Parse()

	if leaseLockName == "" {
		klog.Fatal("missing lease-name flag")
	}
	if leaseLockNamespace == "" {
		klog.Fatal("missing lease-namespace flag")
	}

	config, err := rest.InClusterConfig()
	client = clientset.NewForConfigOrDie(config)

	if err != nil {
		klog.Fatalf("failed to get kubeconfig")
	}

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

	lock := newLeaseLock(leaseLockName, podName, leaseLockNamespace)
	run(lock, ctx, podName)
}

选举逻辑

启动选举

资源锁列表

• leases(recommend)
• endpointsleases
• configmapsleases
• configmaps
• endpoints

创建资源锁

• 使用resourcelock.New创建

  // Manufacture will create a lock of a given type according to the input parameters
  func New(lockType string, ns string, name string, coreClient corev1.CoreV1Interface, coordinationClient coordinationv1.CoordinationV1Interface, rlc ResourceLockConfig) (Interface, error) {
    leaseLock := &LeaseLock{
        LeaseMeta: metav1.ObjectMeta{
            Namespace: ns,
            Name:      name,
        },
        Client:     coordinationClient,
        LockConfig: rlc,
    }
    switch lockType {
    case endpointsResourceLock:
        return nil, fmt.Errorf("endpoints lock is removed, migrate to %s", LeasesResourceLock)
    case configMapsResourceLock:
        return nil, fmt.Errorf("configmaps lock is removed, migrate to %s", LeasesResourceLock)
    case LeasesResourceLock:
        return leaseLock, nil
    case endpointsLeasesResourceLock:
        return nil, fmt.Errorf("endpointsleases lock is removed, migrate to %s", LeasesResourceLock)
    case configMapsLeasesResourceLock:
        return nil, fmt.Errorf("configmapsleases lock is removed, migrated to %s", LeasesResourceLock)
    default:
        return nil, fmt.Errorf("Invalid lock-type %s", lockType)
    }
  }
  

• 使用resourcelock.NewFromKubeconfig创建

  // NewFromKubeconfig will create a lock of a given type according to the input parameters.
  // Timeout set for a client used to contact to Kubernetes should be lower than
  // RenewDeadline to keep a single hung request from forcing a leader loss.
  // Setting it to max(time.Second, RenewDeadline/2) as a reasonable heuristic.
  func NewFromKubeconfig(lockType string, ns string, name string, rlc ResourceLockConfig, kubeconfig *restclient.Config, renewDeadline time.Duration) (Interface, error) {
    // shallow copy, do not modify the kubeconfig
    config := *kubeconfig
    timeout := renewDeadline / 2
    if timeout < time.Second {
        timeout = time.Second
    }
    config.Timeout = timeout
    leaderElectionClient := clientset.NewForConfigOrDie(restclient.AddUserAgent(&config, "leader-election"))
    return New(lockType, ns, name, leaderElectionClient.CoreV1(), leaderElectionClient.CoordinationV1(), rlc)
  }
  

选举

准备好资源锁后，调用方法leaderelection.RunOrDie开始选举。

leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
	Lock:            lock,
	ReleaseOnCancel: true,
	LeaseDuration:   15 * time.Second,
	RenewDeadline:   10 * time.Second,
	RetryPeriod:     2 * time.Second,
	Callbacks: leaderelection.LeaderCallbacks{
		OnStartedLeading: func(c context.Context) {
			klog.Info("get identity", "leader: ", lock.LeaseMeta.Name)
			select {}
		},
		OnStoppedLeading: func() {
			klog.Info("no longer the leader, staying inactive.")
		},
		OnNewLeader: func(current_id string) {
			if current_id == id {
				klog.Info("still the leader!")
				return
			}
			klog.Info("new leader is %s", current_id)
		},
	},
})

选举主流程

启动选举后，RunOrDie方法会调用le.Run(ctx)方法开始真正的选举流程，该方法除非在以下情况下才会返回：

• ctx被取消（外部要求中止选举流程）
• 当选了 Leader 后，任期结束（网络或某种原因导致续期失败）

当未曾竞选Leader时，则会卡在le.acquire方法中持续竞选

// Run starts the leader election loop. Run will not return
// before leader election loop is stopped by ctx or it has
// stopped holding the leader lease
func (le *LeaderElector) Run(ctx context.Context) {
	defer runtime.HandleCrash()
	defer 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)
}

// RunOrDie starts a client with the provided config or panics if the config
// fails to validate. RunOrDie blocks until leader election loop is
// stopped by ctx or it has stopped holding the leader lease
func RunOrDie(ctx context.Context, lec LeaderElectionConfig) {
	le, err := NewLeaderElector(lec)
	if err != nil {
		panic(err)
	}
	if lec.WatchDog != nil {
		lec.WatchDog.SetLeaderElection(le)
	}
	le.Run(ctx)
}

竞选

竞选逻辑

// 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 := false
	desc := le.config.Lock.Describe()
	klog.Infof("attempting to acquire leader lease %v...", desc)
	wait.JitterUntil(func() {
		if !le.config.Coordinated {
			succeeded = le.tryAcquireOrRenew(ctx)
		} else {
			succeeded = le.tryCoordinatedRenew(ctx)
		}
		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
}

返回值情况：

• succeeded 默认值是false，在竞选循环中，外部通过ctx中止竞选时会中止wait.JitterUntil循环并返回false
• 如果竞选成功，则会改写succeeded=true并手动调用cancel()中止wait.JitterUntil循环

故acquire函数只会有两种情况会返回：

• true：当选 leader
• false：外部中止竞选

未当选的竞选者会在wait.JitterUntil循环中持续尝试

抢锁和续约

抢锁操作和续约操作逻辑都在函数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(ctx context.Context) bool {
	now := metav1.NewTime(le.clock.Now())
	leaderElectionRecord := rl.LeaderElectionRecord{
		HolderIdentity:       le.config.Lock.Identity(),
		LeaseDurationSeconds: int(le.config.LeaseDuration / time.Second),
		RenewTime:            now,
		AcquireTime:          now,
	}

	// 1. fast path for the leader to update optimistically assuming that the record observed
	// last time is the current version.
	if le.IsLeader() && le.isLeaseValid(now.Time) {
		oldObservedRecord := le.getObservedRecord()
		leaderElectionRecord.AcquireTime = oldObservedRecord.AcquireTime
		leaderElectionRecord.LeaderTransitions = oldObservedRecord.LeaderTransitions

		err := le.config.Lock.Update(ctx, leaderElectionRecord)
		if err == nil {
			le.setObservedRecord(&leaderElectionRecord)
			return true
		}
		klog.Errorf("Failed to update lock optimistically: %v, falling back to slow path", err)
	}

	// 2. obtain or create the ElectionRecord
	oldLeaderElectionRecord, oldLeaderElectionRawRecord, err := le.config.Lock.Get(ctx)
	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(ctx, leaderElectionRecord); err != nil {
			klog.Errorf("error initially creating leader election record: %v", err)
			return false
		}

		le.setObservedRecord(&leaderElectionRecord)

		return true
	}

	// 3. Record obtained, check the Identity & Time
	if !bytes.Equal(le.observedRawRecord, oldLeaderElectionRawRecord) {
		le.setObservedRecord(oldLeaderElectionRecord)

		le.observedRawRecord = oldLeaderElectionRawRecord
	}
	if len(oldLeaderElectionRecord.HolderIdentity) > 0 && le.isLeaseValid(now.Time) && !le.IsLeader() {
		klog.V(4).Infof("lock is held by %v and has not yet expired", oldLeaderElectionRecord.HolderIdentity)
		return false
	}

	// 4. We're going to try to update. The leaderElectionRecord is set to it's default
	// here. Let's correct it before updating.
	if le.IsLeader() {
		leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTime
		leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions
		le.metrics.slowpathExercised(le.config.Name)
	} else {
		leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + 1
	}

	// update the lock itself
	if err = le.config.Lock.Update(ctx, leaderElectionRecord); err != nil {
		klog.Errorf("Failed to update lock: %v", err)
		return false
	}

	le.setObservedRecord(&leaderElectionRecord)
	return true
}

原理分析

任期原理

基于le.observedTime判断当前主结点是否有效，le.observedTime是在主结点发生变化时调用le.setObservedRecord方法更新。更新时机：

• 锁不存在，创建锁成功时更新
• 获取到锁记录和缓存的不同，说明上次尝试获取锁到现在的间隔内主结点变更，更新缓存
• 主结点超期没续期且当前节点抢锁成功，更新缓存

所以le.observedTime变化的时候都是监测到主结点变化的时间，所以le.observedTime + LeaseDuration的时间就是主结点的结束时间

抢锁原理

基于 kubernetes 的资源乐观锁实现的(通常是主结点自己去更新锁信息)

• 获取锁方法le.config.Lock.Get(ctx)会取得当前锁的最新resourceVersion并保存
• 更新锁时提供保存的resourceVersion
• Kubernetes对比resourceVersion和最新值，如果相等则允许更新，返回成功，否则更新失败