使用etcd選舉sdk實踐master/slave故障轉移

本次將記錄[利用etcd選主sdk實踐master/slave高可用]， 並利用etcdctl原生腳本驗證選主sdk的工作原理。

master/slave高可用集群

本文目標

在異地多機房部署節點，slave作為備用實例啟動，但不接受業務流量， 監測到master宕機，slave節點自動提升為master並接管業務流量。

基本思路

各節點向etcd注冊帶租約的節點信息， 並各自維持心跳保活，選主sdk根據目前存活的、最早創建的節點信息鍵值對 來決定leader， 並通過watch機制通知業務代碼leader變更。

講道理，每個節點只需要知道兩個信息就能各司其職

• 誰是leader > 當前節點是什么角色=> 當前節點該做什么事情
• 感知集群leader變更的能力 ===》當前節點現在要不要改變行為

除了官方etcd客戶端go.etcd.io/etcd/client/v3， 還依賴go.etcd.io/etcd/client/v3/concurrency package：實現了基於etcd的分布式鎖、屏障、選舉

選主過程	實質	api
競選前先查詢leader了解現場	查詢當前存活的，最早創建的kv值	*concurrency.Election.Leader()
初始化時，各節點向etcd阻塞式競選	各節點向etcd注冊帶租約的鍵值對	*concurrency.Election.compaign
建立master/slave集群，還能及時收到變更通知	通過chan傳遞最新的leader value	*concurrency.Election.Observe()

重點解讀

1.初始化etcd go客戶端

注意：etcd客戶端和服務端是通過grpc來通信，目前新版本的etcd客戶端默認使用非阻塞式連接， 也就是說v3.New函數僅表示從指定配置創建etcd客戶端。

為快速確定etcd選舉的可用性，本實踐使用阻塞式創建客戶端：

cli, err := v3.New(v3.Config{
		Endpoints:   addr,
		DialTimeout: time.Second * 5,
		DialOptions: []grpc.DialOption{grpc.WithBlock()},
	})
	if err != nil {
		log.WithField("instance", Id).Errorln(err)
		return nil, err
	}

2. 競選

使用阻塞式命令compaign競選之前，應先查詢當前leader

// 將id：ip：port作為競選時寫入etcd的value
func (c *Client) Election(id string, notify chan<- bool) error {
	//競選前先試圖去了解情況
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	resp, err := c.election.Leader(ctx)
	if err != nil {
		if err != concurrency.ErrElectionNoLeader {
			return err
		}
	} else { // 已經有leader了
		c.Leader = string(resp.Kvs[0].Value)
		notify <- (c.Leader == id)
	}

	if err = c.election.Campaign(context.TODO(), id); err != nil {
		log.WithError(err).WithField("id", id).Error("Campaign error")
		return err
	} else {
		log.Infoln("Campaign success!!!")
		c.Leader = id
		notify <- true
	}
	c.election.Key()
	return nil
}

競選：election.Campaign(val) 的實質是將K:V（節點id）添加到etcd， 並給於10s的TTL保活

過程	輸出	備注
etcdctl lease grant 10	lease 694d813d3a3ccbfb granted with TTL(10s)
etcdctl lease keep-alive 694d813d3a3ccbfb	lease 694d813d3a3ccbfb keepalived with TTL(20)......	NewSession(ttl:10s) 默認創建帶keep-alive的lease
etcdctl put --lease=694d813d3a3ccbfb /merc/694d813d3a3ccbfb id	ok	使用綁定到節點保活會話的 leaseId作為參選的節點標識

etcd v3 從邏輯上是一個扁平的二進制鍵空間，此處產生的key以keyprefix 為前綴，我們需要在該keyprefix下全局定位幾個參與競選的節點，這里使用綁定到參選節點的leaseId。
etcd v2 從邏輯上是一個帶階級的樹形結構（有文件夾）

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

當選： 當前存活的、最早創建的key是leader ， 也就是說master/slave故障轉移並不是隨機的。

3. watch leader變更

golang使用信道完成goroutine通信，

本例聲明信道： notify = make(chan bool, 1)

一石二鳥：標記集群leader是否發生變化；信道內傳值表示當前節點是否是leader

func (c *Client) Watchloop(id string, notify chan<- bool) error {
	ch := c.election.Observe(context.TODO()) // 觀察leader變更
	tick := time.NewTicker(c.askTime)

	defer tick.Stop()
	for {
		var leader string

		select {
		case _ = <-c.sessionCh:
			log.Warning("Recv session event")
			return fmt.Errorf("session Done") // 一次續約不穩，立馬退出程序
		case e := <-ch:
			log.WithField("event", e).Info("watch leader event")
			leader = string(e.Kvs[0].Value)
			ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
			defer cancel()
			resp, err := c.election.Leader(ctx)
			if err != nil {
				if err != concurrency.ErrElectionNoLeader {
					return err
				} else { // 目前沒leader，開始競選了
					if err = c.election.Campaign(context.TODO(), id); err != nil {
						log.WithError(err).WithField("id", id).Error("Campaign error")
						return err
					} else { // 競選成功
						leader = id
					}
				}
			} else {
				leader = string(resp.Kvs[0].Value)
			}
		}
		if leader != c.Leader {
			log.WithField("before", c.Leader).WithField("after", leader == id).Info("leader changed")
			notify <- (leader == id)
		}
		c.Leader = leader
	}
}

c.election.Observe(context.TODO()) 返回最新的leader信息，配合select case控制結構能夠及時拿到leader變更信息。

如題：通過Leader字段和chan <- bool， 掌控了整個選舉集群的狀態， 可根據這兩個信息去完成業務上的master/slave故障轉移。

使用etcdctl確定leader

election.Leader的源碼證明了[當前存活的，最早創建的kv為leader]

// 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
	} else if len(resp.Kvs) == 0 {
		// no leader currently elected
		return nil, ErrElectionNoLeader
	}
	return resp, nil
}

等價於./etcdctl --endpoints=127.0.0.1:2379 get /merc --prefix --sort-by=CREATE --order=ASCEND --limit=1

--prefix 正如上面所言:使用綁定到節點保活會話的leaseId作為參選的節點標識， 這里我們指定--prefix /merc篩選key
--sort-by ：以x標准(創建時間)檢索數據
-- order ： 以升降序對已檢出的數據排序
-- limit： 從已檢出的數據中取x條數據顯示

這里留一個疑問：當前存活的，最早創建的節點: 如何定義最早創建的節點？