k8s client-go快速入门教程及源代码阅读之选举

func main() {
	var kubeconfig *string
	var leaseLockName string
	var leaseLockNamespace string
	var id string

	if home := homedir.HomeDir(); home != "" {
		kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
	} else {
		kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
	}
	flag.StringVar(&id, "id", uuid.New().String(), "the holder identity name")
	flag.StringVar(&leaseLockName, "lease-lock-name", "client-go-election", "the lease lock resource name")
	flag.StringVar(&leaseLockNamespace, "lease-lock-namespace", "default", "the lease lock resource namespace")
	flag.Parse()

	// 1.
	config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
	client := clientset.NewForConfigOrDie(config)
    // 2.
	run := func(ctx context.Context) {
		klog.Info("现在我是leader了")
		select {}
	}

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

    // 3.
	ch := make(chan os.Signal, 1)
	signal.Notify(ch, os.Interrupt, syscall.SIGTERM)
	go func() {
		<-ch
		klog.Info("Received termination, signaling shutdown")
		cancel()
	}()

    // 4.
	lock := &resourcelock.LeaseLock{
		LeaseMeta: metav1.ObjectMeta{
			Name:      leaseLockName,
			Namespace: leaseLockNamespace,
		},
		Client: client.CoordinationV1(),
		LockConfig: resourcelock.ResourceLockConfig{
			Identity: id,
		},
	}

	// 5.
	leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
		Lock: lock,
		ReleaseOnCancel: true,
		LeaseDuration:   60 * time.Second,
		RenewDeadline:   15 * time.Second,
		RetryPeriod:     5 * time.Second,
        // 6.
		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) {
                // 7. 
				if identity == id {
					return
				}
				klog.Infof("new leader elected: %s", identity)
			},
		},
	})
}

kubectl get leases election
# 输出如下
NAME       HOLDER                                 AGE
election   ea0a3e2a-d8ec-4fa8-8581-02a0507337c9   16s

leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
    // 省略相关配置
}
                        
func RunOrDie(ctx context.Context, lec LeaderElectionConfig) {
    // 1.
	le, err := NewLeaderElector(lec)
    // 2.
	le.Run(ctx)
}
                        
func (le *LeaderElector) Run(ctx context.Context) {
	defer runtime.HandleCrash()
	defer func() {
        // 3. 
		le.config.Callbacks.OnStoppedLeading()
	}()
	// 4.
	if !le.acquire(ctx) {
		return // ctx signalled done
	}
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
    // 5.
	go le.config.Callbacks.OnStartedLeading(ctx)
    // 6.
	le.renew(ctx)
}

func (le *LeaderElector) acquire(ctx context.Context) bool {
    // 1.
    ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	succeeded := false
	wait.JitterUntil(func() {
        // 2.
		succeeded = le.tryAcquireOrRenew(ctx)
        // 3. 
		le.maybeReportTransition()
        // 4.
		if !succeeded {
			klog.V(4).Infof("failed to acquire lease %v", desc)
			return
		}
        // 5. 
        cancel()
	}, le.config.RetryPeriod, JitterFactor, true, ctx.Done())
	return succeeded
}

func (le *LeaderElector) tryAcquireOrRenew(ctx context.Context) 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, oldLeaderElectionRawRecord, err := le.config.Lock.Get(ctx)
	if err != nil {
        // 2.
		if !errors.IsNotFound(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.observedRecord = leaderElectionRecord
		le.observedTime = le.clock.Now()
		return true
	}

	// 3.
	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
	}

	// 4.
	if le.IsLeader() {
		leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTime
		leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions
	} else {
		leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions + 1
	}

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

func (ll *LeaseLock) Get(ctx context.Context) (*LeaderElectionRecord, []byte, error) {
	var err error
    // 6.
	ll.lease, err = ll.Client.Leases(ll.LeaseMeta.Namespace).Get(ctx, ll.LeaseMeta.Name, metav1.GetOptions{})
	record := LeaseSpecToLeaderElectionRecord(&ll.lease.Spec)
	recordByte, err := json.Marshal(*record)
	if err != nil {
		return nil, nil, err
	}
	return record, recordByte, nil
}

func (le *LeaderElector) Run(ctx context.Context) {
	// 省略获取租约等逻辑
	le.renew(ctx)
}

func (le *LeaderElector) renew(ctx context.Context) {
   	// 1.
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	wait.Until(func() {
        // 2.
		timeoutCtx, timeoutCancel := context.WithTimeout(ctx, le.config.RenewDeadline)
		defer timeoutCancel()
        // 3.
		err := wait.PollImmediateUntil(le.config.RetryPeriod, func() (bool, error) {
			return le.tryAcquireOrRenew(timeoutCtx), nil
		}, timeoutCtx.Done())
		// 4.
		le.maybeReportTransition()
		desc := le.config.Lock.Describe()
		if err == nil {
			klog.V(5).Infof("successfully renewed lease %v", desc)
			return
		}
        // 5.
		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())

	// 6.
	if le.config.ReleaseOnCancel {
		le.release()
	}
}