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/v2"
)
func buildConfig(kubeconfig string) (*rest.Config, error) {
  if kubeconfig != "" {
    cfg, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
      return nil, err
    }
    return cfg, nil
  }
  cfg, err := rest.InClusterConfig()
  if err != nil {
    return nil, err
  }
  return cfg, nil
}
func main() {
  klog.InitFlags(nil)
  var kubeconfig string
  var leaseLockName string
  var leaseLockNamespace string
  var id string
  // kubeconfig 指定了kubernetes集群的配置文文件路徑
  flag.StringVar(&kubeconfig, "kubeconfig", "", "absolute path to the kubeconfig file")
  // 鎖的擁有者的ID，如果沒有傳參數(shù)進(jìn)來，就隨機生成一個
  flag.StringVar(&id, "id", uuid.New().String(), "the holder identity name")
  // 鎖的ID，對應(yīng)kubernetes中資源的name
  flag.StringVar(&leaseLockName, "lease-lock-name", "", "the lease lock resource name")
  // 鎖的命名空間
  flag.StringVar(&leaseLockNamespace, "lease-lock-namespace", "", "the lease lock resource namespace")
  // 解析命令行參數(shù)
  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)
  }
  // 獲取kubernetes集群的客戶端，如果獲取不到，就拋異常退出
  client := clientset.NewForConfigOrDie(config)
  // 模擬Controller的邏輯代碼
  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
  // 處理系統(tǒng)的系統(tǒng)，收到SIGTERM信號后，會退出進(jìn)程
  ch := make(chan os.Signal, 1)
  signal.Notify(ch, os.Interrupt, syscall.SIGTERM)
  go func() {
    <-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".
  
  // 根據(jù)參數(shù)，生成鎖。這里使用的Lease這種類型資源作為鎖
  lock := &resourcelock.LeaseLock{
    LeaseMeta: metav1.ObjectMeta{
      Name:      leaseLockName,
      Namespace: leaseLockNamespace,
    },
    // 跟kubernetes集群關(guān)聯(lián)起來
    Client: client.CoordinationV1(),
    LockConfig: resourcelock.ResourceLockConfig{
      Identity: id,
    },
  }
  // start the leader election code loop
  
  // 注意，選舉邏輯啟動時候，會傳入ctx參數(shù)，如果ctx對應(yīng)的cancel函數(shù)被調(diào)用，那么選舉也會結(jié)束
  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.
    //主動放棄leader，當(dāng)ctx canceled的時候
    ReleaseOnCancel: true,
    LeaseDuration:   60 * time.Second,  // 選舉的任期，60s一個任期，如果在60s后沒有renew，那么leader就會釋放鎖，重新選舉
    RenewDeadline:   15 * time.Second,  // renew的請求的超時時間
    RetryPeriod:     5 * time.Second, // leader獲取到鎖后，renew leadership的間隔。非leader，搶鎖成為leader的間隔（有1.2的jitter因子，詳細(xì)看代碼）
    
    // 回調(diào)函數(shù)的注冊
    Callbacks: leaderelection.LeaderCallbacks{
      
      // 成為leader的回調(diào)
      OnStartedLeading: func(ctx context.Context) {
        // we're notified when we start - this is where you would
        // usually put your code
        // 運行controller的邏輯
        run(ctx)
      },
      OnStoppedLeading: func() {
        // we can do cleanup here
        // 退出leader的
        klog.Infof("leader lost: %s", id)
        os.Exit(0)
      },
      OnNewLeader: func(identity string) {
        // 有新的leader當(dāng)選
        // we're notified when new leader elected
        if identity == id {
          // I just got the lock
          return
        }
        klog.Infof("new leader elected: %s", identity)
      },
    },
  })
}

type Interface interface {
  // Get returns the LeaderElectionRecord
  Get(ctx context.Context) (*LeaderElectionRecord, []byte, error)
  // Create attempts to create a LeaderElectionRecord
  Create(ctx context.Context, ler LeaderElectionRecord) error
  // Update will update and existing LeaderElectionRecord
  Update(ctx context.Context, ler LeaderElectionRecord) error
  // RecordEvent is used to record events
  RecordEvent(string)
  // Identity will return the locks Identity
  Identity() string
  // Describe is used to convert details on current resource lock
  // into a string
  Describe() string
}

func (le *LeaderElector) Run(ctx context.Context) {
  defer runtime.HandleCrash()
  defer func() {
    le.config.Callbacks.OnStoppedLeading()
  }()
  // 獲取鎖，如果沒有獲取到，就一直等待
  if !le.acquire(ctx) {
    return // ctx signalled done
  }
  ctx, cancel := context.WithCancel(ctx)
  defer cancel()
  // 獲取到鎖后，需要調(diào)用回調(diào)函數(shù)中的OnStartedLeading，運行controller的代碼
  go le.config.Callbacks.OnStartedLeading(ctx)
  
  // 獲取到鎖后，需要不斷地進(jìn)行renew操作
  le.renew(ctx)
}

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)
  // 此接口會阻塞，利用定時的機制，獲取鎖，如果獲取不到一直循環(huán)，除非ctx被取消。
  wait.JitterUntil(func() {
    // 獲取鎖
    succeeded = le.tryAcquireOrRenew(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
}
// 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()
  // 循環(huán)renew機制，renew成功，不會返回true，導(dǎo)致Until會不斷循環(huán)
  wait.Until(func() {
    //RenewDeadline的實現(xiàn)在這里，如果renew超過了RenewDeadline，會導(dǎo)致renew失敗，主退出
    timeoutCtx, timeoutCancel := context.WithTimeout(ctx, le.config.RenewDeadline)
    defer timeoutCancel()
    err := wait.PollImmediateUntil(le.config.RetryPeriod, func() (bool, error) {
      // renew鎖
      return le.tryAcquireOrRenew(timeoutCtx), nil
    }, timeoutCtx.Done())
    le.maybeReportTransition()
    desc := le.config.Lock.Describe()
    if err == nil {
      klog.V(5).Infof("successfully renewed lease %v", desc)
      // renew成功
      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 up
  if le.config.ReleaseOnCancel {
    le.release()
  }
}

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 {
    // 沒有鎖的資源，就創(chuàng)建一個
    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
    }
    //對外宣稱自己成為了leader
    le.setObservedRecord(&leaderElectionRecord)
    return true
  }
  // 2. Record obtained, check the Identity & Time
  if !bytes.Equal(le.observedRawRecord, oldLeaderElectionRawRecord) {
    // 這個機制很重要，會如果leader會不斷正常renew這個鎖，oldLeaderElectionRawRecord會一直發(fā)生變化，發(fā)生變化會更新le.observedTime
    le.setObservedRecord(oldLeaderElectionRecord)
    le.observedRawRecord = oldLeaderElectionRawRecord
  }
  // 如果還沒超時并且此實例不是leader（leader是其他實例），那么就直接退出
  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.
  // 如果是leader，就更新時間RenewTime，保證其他實例（非主）可以觀察到：主還活著
  if le.IsLeader() {
    leaderElectionRecord.AcquireTime = oldLeaderElectionRecord.AcquireTime
    leaderElectionRecord.LeaderTransitions = oldLeaderElectionRecord.LeaderTransitions
  } else {
  // 不是leader，那么鎖就發(fā)生了轉(zhuǎn)移
    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
}