Listen to leadership change, instead of polling

[paulm17]

Hi there,

I've come up with a poc (full code below).

Right now I'm polling to determine if the current node is the leader.

Is there a way to listen for this instead? Thanks

polling mechanism:

for {
		select {
		case <-ticker.C:
			// Get the cluster status
			resp, err := cli.Status(context.Background(), cli.Endpoints()[0])
			if err != nil {
				log.Printf("Error getting cluster status: %v", err)
				continue
			}

			// Check if the current node is the leader
			isLeader := resp.Leader == resp.Header.MemberId
			fmt.Printf("Is the current node the leader? %v\n", isLeader)
		}
	}

full code:

package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"net/url"
	"os"
	"strings"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/server/v3/embed"
)

func genURLArray(urlStr string) []url.URL {
	if !strings.Contains(urlStr, "http://") {
		urlStr = "http://" + urlStr
	}

	urls := []url.URL{}

	newURL, err := url.Parse(urlStr)
	if err != nil {
		log.Fatal(err)
	}

	urls = append(urls, *newURL)

	return urls
}

func formatAddresses(input string) string {
	// Split the input string by commas
	addresses := strings.Split(input, ",")

	// Initialize a slice to hold the formatted addresses
	formattedAddresses := make([]string, len(addresses))

	// Iterate over the addresses
	for i, address := range addresses {
		// Prepend each address with "http://" and append with a node identifier
		formattedAddresses[i] = fmt.Sprintf("node%d=http://%s", i+1, address)
	}

	// Join the formatted addresses back into a single string with commas
	return strings.Join(formattedAddresses, ",")
}

func main() {
	// Define flags for configuration
	var (
		dirPath             = flag.String("dir", "./data", "Directory path for etcd data")
		name                = flag.String("name", "node1", "Name of the etcd node")
		initialCluster      = flag.String("cluster", "localhost:2380,localhost:2382,localhost:2384", "Cluster configuration")
		initialClusterToken = flag.String("cluster-token", "etcd-cluster", "Cluster token")
		peer                = flag.String("peer", "localhost:2380", "Listen peer URLs")
		client              = flag.String("client", "localhost:2379", "Listen client URLs")
	)

	flag.Parse()

	// Create directory if it doesn't exist
	err := os.MkdirAll(*dirPath, 0777)
	if err != nil {
		log.Fatal(err)
	}

	// Configuration for the first node
	cfg := embed.NewConfig()
	cfg.Name = *name
	cfg.Dir = fmt.Sprintf("%s/%s/%s.etcd", *dirPath, *name, *name)
	cfg.InitialCluster = formatAddresses(*initialCluster)
	cfg.InitialClusterToken = *initialClusterToken
	cfg.ListenPeerUrls, cfg.AdvertisePeerUrls = genURLArray(*peer), genURLArray(*peer)
	cfg.ListenClientUrls, cfg.AdvertiseClientUrls = genURLArray(*client), genURLArray(*client)

	// Start the first node
	e, err := embed.StartEtcd(cfg)
	if err != nil {
		log.Fatal(err)
	}
	defer e.Close()

	// Wait for the first node to be ready
	select {
	case <-e.Server.ReadyNotify():
		log.Printf("%v is ready!", *name)
	case <-time.After(60 * time.Second):
		e.Server.Stop()
		log.Printf("%v took too long to start!", *name)
	}

	// Create an etcd client
	cli, err := clientv3.New(clientv3.Config{
		Endpoints: []string{*client},
	})
	if err != nil {
		log.Fatal(err)
	}
	defer cli.Close()

	// Create a ticker that triggers every second
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	// Continuously check the leader status
	for {
		select {
		case <-ticker.C:
			// Get the cluster status
			resp, err := cli.Status(context.Background(), cli.Endpoints()[0])
			if err != nil {
				log.Printf("Error getting cluster status: %v", err)
				continue
			}

			// Check if the current node is the leader
			isLeader := resp.Leader == resp.Header.MemberId
			fmt.Printf("Is the current node the leader? %v\n", isLeader)
		}
	}

	// Wait for errors
	// log.Fatal(<-e.Err())
}