Kubernetes 如何实现组件高可用

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

 // 2. Record obtained, check the Identity & Time
 if !bytes.Equal(le.observedRawRecord, oldLeaderElectionRawRecord) {
  le.setObservedRecord(oldLeaderElectionRecord)
  le.observedRawRecord = oldLeaderElectionRawRecord
 }
 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. 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
 } 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
}

$ kubectl get lease demo -oyaml
apiVersion: coordination.k8s.io/v1
kind: Lease
metadata:
  ...
spec:
  acquireTime: "2022-07-23T14:28:41.381108Z"
  holderIdentity: "1"
  leaseDurationSeconds: 60
  leaseTransitions: 0
  renewTime: "2022-07-23T14:28:41.397199Z"

func (le *LeaderElector) release() bool {
 if !le.IsLeader() {
  return true
 }
 now := metav1.Now()
 leaderElectionRecord := rl.LeaderElectionRecord{
  LeaderTransitions:    le.observedRecord.LeaderTransitions,
  LeaseDurationSeconds: 1,
  RenewTime:            now,
  AcquireTime:          now,
 }
 if err := le.config.Lock.Update(context.TODO(), leaderElectionRecord); err != nil {
  klog.Errorf("Failed to release lock: %v", err)
  return false
 }

 le.setObservedRecord(&leaderElectionRecord)
 return true
}

$ kubectl get lease demo -oyaml
apiVersion: coordination.k8s.io/v1
kind: Lease
metadata:
  ...
spec:
  acquireTime: "2022-07-23T14:29:26.557658Z"
  holderIdentity: ""
  leaseDurationSeconds: 1
  leaseTransitions: 0
  renewTime: "2022-07-23T14:29:26.557658Z"

func NewResourceLock(config *rest.Config, recorderProvider recorder.Provider, options Options) (resourcelock.Interface, error) {
 if options.LeaderElectionResourceLock == "" {
  options.LeaderElectionResourceLock = resourcelock.LeasesResourceLock
 }

 // LeaderElectionID must be provided to prevent clashes
 if options.LeaderElectionID == "" {
  return nil, errors.New("LeaderElectionID must be configured")
 }

 // Default the namespace (if running in cluster)
 if options.LeaderElectionNamespace == "" {
  var err error
  options.LeaderElectionNamespace, err = getInClusterNamespace()
  if err != nil {
   return nil, fmt.Errorf("unable to find leader election namespace: %w", err)
  }
 }

 // Leader id, needs to be unique
 id, err := os.Hostname()
 if err != nil {
  return nil, err
 }
 id = id + "_" + string(uuid.NewUUID())

 // Construct clients for leader election
 rest.AddUserAgent(config, "leader-election")
 corev1Client, err := corev1client.NewForConfig(config)
 if err != nil {
  return nil, err
 }

 coordinationClient, err := coordinationv1client.NewForConfig(config)
 if err != nil {
  return nil, err
 }

 return resourcelock.New(options.LeaderElectionResourceLock,
  options.LeaderElectionNamespace,
  options.LeaderElectionID,
  corev1Client,
  coordinationClient,
  resourcelock.ResourceLockConfig{
   Identity:      id,
   EventRecorder: recorderProvider.GetEventRecorderFor(id),
  })
}

func (cm *controllerManager) startLeaderElection(ctx context.Context) (err error) {
 l, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
  Lock:          cm.resourceLock,
  LeaseDuration: cm.leaseDuration,
  RenewDeadline: cm.renewDeadline,
  RetryPeriod:   cm.retryPeriod,
  Callbacks: leaderelection.LeaderCallbacks{
   OnStartedLeading: func(_ context.Context) {
    if err := cm.startLeaderElectionRunnables(); err != nil {
     cm.errChan <- err
     return
    }
    close(cm.elected)
   },
   OnStoppedLeading: func() {
    if cm.onStoppedLeading != nil {
     cm.onStoppedLeading()
    }
    cm.gracefulShutdownTimeout = time.Duration(0)
    cm.errChan <- errors.New("leader election lost")
   },
  },
  ReleaseOnCancel: cm.leaderElectionReleaseOnCancel,
 })
 if err != nil {
  return err
 }

 // Start the leader elector process
 go func() {
  l.Run(ctx)
  <-ctx.Done()
  close(cm.leaderElectionStopped)
 }()
 return nil
}

 scheme := runtime.NewScheme()
 _ = corev1.AddToScheme(scheme)
 // 1. init Manager
 mgr, _ := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
  Scheme: scheme,
  Port:   9443,
  LeaderElection:     true,
  LeaderElectionID:   "demo.xxx",
 })
 // 2. init Reconciler（Controller）
 _ = ctrl.NewControllerManagedBy(mgr).
  For(&corev1.Pod{}).
  Complete(&ApplicationReconciler{})
...