Kubernetes中leaderelection实现组件高可用

/*
例子来源于client-go中的example包中，
*/
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"
)

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

	cfg, err := rest.InClusterConfig()
	if err != nil {
		returnnil, 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)
	}
	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)
	gofunc() {
		<-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
	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,//租约时间
		RenewDeadline:   15 * time.Second,//更新租约的
		RetryPeriod:     5 * time.Second,//非leader节点重试时间
		Callbacks: leaderelection.LeaderCallbacks{
			OnStartedLeading: func(ctx context.Context) {
                //变为leader执行的业务代码
				// we're notified when we start - this is where you would
				// usually put your code
				run(ctx)
			},
			OnStoppedLeading: func() {
                 // 进程退出
				// we can do cleanup here
				klog.Infof("leader lost: %s", id)
				os.Exit(0)
			},
			OnNewLeader: func(identity string) {
                //当产生新的leader后执行的方法
				// we're notified when new leader elected
				if identity == id {
					// I just got the lock
					return
				}
				klog.Infof("new leader elected: %s", identity)
			},
		},
	})
}

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)
}

type LeaderElectionConfig struct {
	// Lock 的类型
	Lock rl.Interface
	//持有锁的时间
	LeaseDuration time.Duration
	//在更新租约的超时时间
	RenewDeadline time.Duration
    //竞争获取锁的时间
	RetryPeriod time.Duration
    //状态变化时执行的函数，支持三种：
    //1、OnStartedLeading 启动是执行的业务代码
    //2、OnStoppedLeading leader停止执行的方法
    //3、OnNewLeader 当产生新的leader后执行的方法
	Callbacks LeaderCallbacks

    //进行监控检查
	// WatchDog is the associated health checker
	// WatchDog may be null if its not needed/configured.
	WatchDog *HealthzAdaptor
    //leader退出时，是否执行release方法
	ReleaseOnCancel bool
    
	// Name is the name of the resource lock for debugging
	Name string
}

switch lockType {
	case EndpointsResourceLock://保存在endpoints
		return endpointsLock, nil
	case ConfigMapsResourceLock://保存在configmaps
		return configmapLock, nil
	case LeasesResourceLock://保存在leases
		return leaseLock, nil
	case EndpointsLeasesResourceLock://优先尝试保存在endpoint失败时保存在lease
		return &MultiLock{
			Primary:   endpointsLock,
			Secondary: leaseLock,
		}, nil
	case ConfigMapsLeasesResourceLock://优先尝试保存在configmap，失败时保存在lease
		return &MultiLock{
			Primary:   configmapLock,
			Secondary: leaseLock,
		}, nil
	default:
		returnnil, fmt.Errorf("Invalid lock-type %s", lockType)
	}

type LeaderElectionRecord struct {
	HolderIdentity       string`json:"holderIdentity"`//持有锁进程的标识，一般可以利用主机名
	LeaseDurationSeconds int`json:"leaseDurationSeconds"`//  lock的租约
	AcquireTime          metav1.Time `json:"acquireTime"`//持有锁的时间
	RenewTime            metav1.Time `json:"renewTime"`//更新时间
	LeaderTransitions    int`json:"leaderTransitions"`//leader更换的次数
}

// Run starts the leader election loop
func (le *LeaderElector) Run(ctx context.Context) {
	deferfunc() {
        //进行退出执行
		runtime.HandleCrash()
        //停止时执行回调方法
		le.config.Callbacks.OnStoppedLeading()
	}()
    //不断的进行获得锁，如果获得锁成功则执行后面的方法，否则不断的进行重试
	if !le.acquire(ctx) {
		return// ctx signalled done
	}
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
    //获取锁成功，当前进程变为leader，执行回调函数中的业务代码
	go le.config.Callbacks.OnStartedLeading(ctx)
    //不断的循环进行进行租约的更新，保证锁一直被当前进行持有
	le.renew(ctx)
}

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. obtain or create the ElectionRecord
    // 第一步：从k8s资源中获取原有的锁
	oldLeaderElectionRecord, 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)
			returnfalse
		}
        //资源对象不存在，进行锁资源创建
		if err = le.config.Lock.Create(leaderElectionRecord); err != nil {
			klog.Errorf("error initially creating leader election record: %v", err)
			returnfalse
		}
		le.observedRecord = leaderElectionRecord
		le.observedTime = le.clock.Now()
		returntrue
	}

	// 2. Record obtained, check the Identity & Time
    // 第二步，对比存储在k8s中的锁资源与上一次获取的锁资源是否一致
	if !bytes.Equal(le.observedRawRecord, oldLeaderElectionRawRecord) {
		le.observedRecord = *oldLeaderElectionRecord
		le.observedRawRecord = oldLeaderElectionRawRecord
		le.observedTime = le.clock.Now()
	}
    //判断持有的锁是否到期以及是否被自己持有
	iflen(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)
		returnfalse
	}

	// 3. We're going to try to update. The leaderElectionRecord is set to it's default
	// here. Let's correct it before updating.
    //第三步：自己现在是leader，但是分两组情况，上一次也是leader和首次变为leader
	if le.IsLeader() {
        //自己本身就是leader则不需要更新AcquireTime和LeaderTransitions
		leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTime
		leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions
	} else {
        //首次自己变为leader则更新leader的更换次数
		leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + 1
	}

    //更新锁资源，这里如果在 Get 和 Update 之间有变化，将会更新失败
	// update the lock itself
	if err = le.config.Lock.Update(leaderElectionRecord); err != nil {
		klog.Errorf("Failed to update lock: %v", err)
		returnfalse
	}

	le.observedRecord = leaderElectionRecord
	le.observedTime = le.clock.Now()
	returntrue
}