一种基于etcd实践节点自动故障转移的思路

type Client struct {
 addr               []string
 Leader             string                  // 每个服务节点都知道集群中谁是leader
 cli                  *clientv3.Client
 val                 string                   // 注册到etcd的值， 标记节点
 election          *concurrency.Election
 electSessionDoneCh <-chan struct{}      // 换届的信号
 IsLeader           bool                        //标示当前节点是否是leader
}

    session, err := concurrency.NewSession(cli, concurrency.WithTTL(10))         // 使用etcd的租约机制来实现心跳保活
 if err != nil {
  return nil, err
 }
 ele := concurrency.NewElection(session, LeaderkeyPfx)     

func (c *Client) Election(ctx context.Context, id string) bool {
 c.Leader = c.leader() 
 err := c.election.Campaign(ctx, id)
 if err != nil {
  log.WithError(err).WithField("id", id).Error("Campaign error")
  return false
 }
 c.IsLeader = true
 return true
}

// Leader returns the leader value for the current election.
func (e *Election) Leader(ctx context.Context) (*v3.GetResponse, error) {
 client := e.session.Client()
 resp, err := client.Get(ctx, e.keyPrefix, v3.WithFirstCreate()...)
if err != nil {
return nil, err
 } elseif len(resp.Kvs) == 0 {
  // no leader currently elected
return nil, ErrElectionNoLeader
 }
return resp, nil
}

func (c *Client) leader() string {
 ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
 defer cancel()

 resp, err := c.election.Leader(ctx)
if err != nil {
return""
 }
return string(resp.Kvs[0].Value)
}

func (c *Client) Watchloop(id string, notify chan<- bool) error {
 ch := c.election.Observe(context.TODO()) // 信道传递最新的leader节点， 但是如果底层watcher被一其他方式中段或者超时， 信道会被关闭
 tick := time.NewTicker(time.Minute*5)          // 5min去问下，防止假死
 defer tick.Stop()
for {
  select {
case <-c.electSessionDoneCh: // Done returns a channel that closes when the lease is orphaned, expires, or  is otherwise no longer being refreshed.
   log.Warning("Recv session event")
   return fmt.Errorf("session Done")    // 意味心跳保活失败
case latestLeaderResp, ok := <-ch: // 注意， 从closed（chan） 会持续读到零值， 造成死循环。
   if !ok {
    log.WithField("topic", "watch-loop").Warn("channel closed， something underlying cause error.")
    ch = c.election.Observe(context.TODO())
   } else {
    log.WithField("topic", "watch-loop").Info(latestLeaderResp)
   }
        case <-tick.C:
  }
  ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
  defer cancel()
  resp, err := c.election.Leader(ctx)
  var isLeader bool
if err != nil {
   if err == concurrency.ErrElectionNoLeader {
    ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
    defer cancel()
    isLeader = c.Election(ctx, id)
   } else {
    log.WithError(err).Errorf("watchLoop get leader error")
    isLeader = false
   }
  } elseif string(resp.Kvs[0].Value) != id { //收到leader变化消息，判断发现自己不是leader，停止工作
   c.Leader = string(resp.Kvs[0].Value)
   isLeader = false
  } else {
   c.Leader = string(resp.Kvs[0].Value)
   isLeader = true
  }
if isLeader != c.IsLeader {
   log.WithField("after", isLeader).WithField("before", c.IsLeader).WithField("leader", c.Leader).Info("reElect")
   notify <- isLeader
   c.IsLeader = isLeader
  }
 }
}

 go func() {
  err := eCli.Watchloop(Id, notify) //后台监测与etcd的连通性以及leader节点的变化
  log.WithError(err).Error("watchLoop error")
  notify <- false
 }()

 go func() {
  if eCli.Election(context.TODO(), Id) {
   notify <- true
  }
 }()

    func (e *Election) Campaign(ctx context.Context, val string) error {
     s := e.session
     client := e.session.Client()

     k := fmt.Sprintf("%s%x", e.keyPrefix, s.Lease())
     txn := client.Txn(ctx).If(v3.Compare(v3.CreateRevision(k), "=", 0))
     txn = txn.Then(v3.OpPut(k, val, v3.WithLease(s.Lease())))
     txn = txn.Else(v3.OpGet(k))
     resp, err := txn.Commit()
     if err != nil {
      return err
     }
     e.leaderKey, e.leaderRev, e.leaderSession = k, resp.Header.Revision, s
     if !resp.Succeeded {
      kv := resp.Responses[0].GetResponseRange().Kvs[0]
      e.leaderRev = kv.CreateRevision
      if string(kv.Value) != val {
       if err = e.Proclaim(ctx, val); err != nil {
        e.Resign(ctx)
        return err
       }
      }
     }

func (kv *kv) Do(ctx context.Context, op Op) (OpResponse, error) {
 var err error
 switch op.t {
case tRange:
if op.IsSortOptionValid() {
   var resp *pb.RangeResponse
   resp, err = kv.remote.Range(ctx, op.toRangeRequest(), kv.callOpts...)
   if err == nil {
    return OpResponse{get: (*GetResponse)(resp)}, nil
   }
  } else {
   err = rpctypes.ErrInvalidSortOption
  }
case tPut: