RFC: Leader Election

[Pivnoy]

Дискуссия посвящена сравнению клиентских API для Leader Election в существующих инструментах.

Для исследования были выбраны инструменты: Etcd, Consul, Zookeeper.

Etcd:

Основные сущности и методы API:

type Election struct {
    session *Session // текущая сессия

	keyPrefix string // префикс

	leaderKey     string // ключ лидера или пустая строка

	leaderRev     int64 // редакция создания лидера

	leaderSession *Session // сессия лидера

	hdr           *pb.ResponseHeader // заголовок ответа от последних успешных выборов
}

func (e *Election) Campaign(ctx context.Context, val string) error // Выставляем значениие, как возможное для выборов
func (e *Election) Proclaim(ctx context.Context, val string) error // Объявление нового значения без выборов
func (e *Election) Resign(ctx context.Context) (err error) // Лидер начинает выборы

Псевдокод для использования:

	s, err := concurrency.NewSession(cli)
	defer s.Close()
	e := concurrency.NewElection(s, "/leader-election/")
	if err := e.Campaign(ctx, "e"); err != nil {
		log.Fatal(err)
	}
	time.Sleep(5 * time.Second)
	if err := e.Resign(ctx); err != nil {
		log.Fatal(err)
	}

Документация:

https://pkg.go.dev/go.etcd.io/etcd@v3.3.27+incompatible/clientv3/concurrency#Election

Consul:

Основные сущности и методы API:

// SessionEntry represents a session in consul
type SessionEntry struct {
	CreateIndex uint64
	ID          string
	Name        string // позволяет задать имя, которое будет использоваться для distributed lock
	Node        string
	LockDelay   time.Duration
	Behavior    string // поведение
	TTL         string // время жизни
	Namespace   string `json:",omitempty"`

	// Deprecated for Consul Enterprise in v1.7.0.
	Checks []string

	// NodeChecks and ServiceChecks are new in Consul 1.7.0.
	// When associating checks with sessions, namespaces can be specified for service checks.
	NodeChecks    []string
	ServiceChecks []ServiceCheck
}

// KV is used to manipulate the K/V API
type KV struct {
    c *Client
}


func (k *KV) Acquire(p *KVPair, q *WriteOptions) (bool, *WriteMeta, error) // Метод, иницирующий попытку захвата блокировки
func (k *KV) Release(p *KVPair, q *WriteOptions) (bool, *WriteMeta, error) // Метод, ответственный за снятие блокировки

Псевдокод для испольвания:

Источник:

https://clivern.com/leader-election-with-consul-and-golang/

	// configs to connect to consul
	client, err := api.NewClient(&api.Config{
		Address: "127.0.0.1:8500",
		Scheme:  "http",
	})

	if err != nil {
		panic(err)
	}

	err = client.Agent().ServiceRegister(&api.AgentServiceRegistration{
		Address: "http://127.0.0.1:8080",
		ID:      "node01_monitoring", // Unique for each node
		Name:    "monitoring",        // Can be service type
		Tags:    []string{"monitoring"},
		Check: &api.AgentServiceCheck{
			HTTP:     "http://127.0.0.1:8080/_health",
			Interval: "10s",
		},
	})

	if err != nil {
		panic(err)
	}

	sessionID, _, err := client.Session().Create(&api.SessionEntry{
		Name:     "service/monitoring/leader", // distributed lock
		Behavior: "delete",
		TTL:      "10s",
	}, nil)

	if err != nil {
		panic(err)
	}

	isLeader, _, err := client.KV().Acquire(&api.KVPair{
		Key:     "service/monitoring/leader", // distributed lock
		Value:   []byte(sessionID),
		Session: sessionID,
	}, nil)

	if err != nil {
		panic(err)
	}

	fmt.Println(isLeader)

	doneChan := make(chan struct{})
	defer close(doneChan)

	go func() {
		// RenewPeriodic is used to periodically invoke Session.Renew on a
		// session until a doneChan is closed. This is meant to be used in a long running
		// goroutine to ensure a session stays valid.
		client.Session().RenewPeriodic(
			"10s",
			sessionID,
			nil,
			doneChan,
		)
	}()

	time.Sleep(90 * time.Second)

Документация:

https://developer.hashicorp.com/consul/tutorials/developer-configuration/application-leader-elections

Zookeeper:

Основные сущности и методы API:

// Election is a structure that represents a new instance of a Election. This instance can then
	// be used to request leadership for a specific resource.
	type Election struct {
		status chan Status // Used to communicate state/ status changes to the client
		// E.g., Client is leader, election is over. It is closed
		// by Election goroutine (i.e., ElectLeader())
		resign chan struct{} // This is a signal channel used by the Client to indicate that
		// it is resigning from the election. It is closed by Client goroutine.
		end chan struct{} // This is a signal channel used by the Client to indicate that
		// the election is over and should be terminated. It is closed by Client goroutine
		zkEvents <-chan zk.Event // Used by Zookeeper to send watch events to interested clients.
		// It will be closed by Zookeeper
		triggerElection chan error // This is used by the ZK watch goroutine to signal that a watched
		// znode was deleted. This can mean that a followed candidate resigned or that the leader exited
		// unexpectedly. It will be closed by the Election goroutine (i.e., ElectLeader()).
		stopZKWatch chan struct{} // This is a signal channel used to indicate that a ZK watch
		// goroutine should exit.
		zkConn           *zk.Conn // This is the Zookeeper connection provided by the client.
		ElectionResource string   // This is the Zookeeper node that represents the overall election. Candidates
		clientName       string   // Client that created the election for instrumentation purposes.
		// will be children of this node.
		candidateID string // This is the ID assigned by Zookeeper that is associated with the ephemeral node
		// representing a candidate
		once sync.Once // Used to protect operations that should only be executed one time (i.e.,
		// closing a channel.
		doneChl <-chan zk.Event // A specific signal that the DONE node, which is created during Delete/EndElection,
		// was created. This event means that the election should be terminated for the receiving client.
	}

func NewElection(zkConn *zk.Conn, electionResource string, clientName string) (*Election, error) // создание сущности выборов, которая сможет принимать запросы на участие в них
func (le *Election) ElectLeader() // объявление начало выборов
func (le *Election) ElectLeader() // делает вызывающего кандидатом на выборы
func (le *Election) EndElection() // окончание выборов
	

Псевдокод для использования:

// Create the ZK connection
zkConn, _, err := zk.Connect([]string{zkURL}, heartBeat)
if err != nil {
log.Printf("Error in zk.Connect (%s): %v",
zkURL, err)
return nil, err
}

// Create the election node in ZooKeeper
_, err = zkConn.Create(electionNode,
[]byte(""), 0, zk.WorldACL(zk.PermAll))
if err != nil {
log.Printf("Error creating the election node (%s): %v",
electionNode, err)
return nil, err
}

candidate, err := leaderelection.NewElection(zkConn, electionNode)

var status Status
var ok bool

for {
    select {
    case status, ok = <-leaderElector.Status():
        if !ok {
            fmt.Println("\t\t\tChannel closed, election is terminated!!!")
            respCh <- ElectionResponse{false, status.CandidateID}
            leaderElector.Resign()
            wg.Done()
            return
        }
        if status.Err != nil {
            fmt.Println("Received election status error <<", status.Err, ">> for candidate <", leaderElector.candidateID, ">.")
            leaderElector.Resign()
            wg.Done()
            return
        }

        fmt.Println("Candidate received status message: <", status, ">.")
        if status.Role == Leader {
            doLeaderStuff(leaderElector, status, respCh, connFailCh, waitFor)
            wg.Done()
            return
        }
    case <-connFailCh:
        fmt.Println("\t\t\tZK connection failed for candidate <", status.CandidateID, ">, exiting")
        respCh <- ElectionResponse{false, status.CandidateID}
        leaderElector.Resign()
        wg.Done()
        return
    case <-time.After(time.Second * 100):
        fmt.Println("\t\t\tERROR!!! Timer expired, stop waiting to become leader for", status.CandidateID)
        leaderElector.Resign()
        respCh <- ElectionResponse{false, status.CandidateID}
        wg.Done()
        return
}

Документация:

Библиотека, реализующая возможность выбора лидера для Zookeeper:

https://github.com/Comcast/go-leaderelection