etcd internal state sync

main.go

package main

import (
	"context"
	"log"
	"time"

	"github.com/coreos/etcd/client"
)

var mySpecialSet = map[string]string{}

func main() {
	cfg := client.Config{
		Endpoints: []string{"http://127.0.0.1:2379"},
		Transport: client.DefaultTransport,
		// set timeout per request to fail fast when the target endpoint is unavailable
		HeaderTimeoutPerRequest: time.Second,
	}
	c, err := client.New(cfg)
	if err != nil {
		log.Fatal(err)
	}
	kapi := client.NewKeysAPI(c)

	// get "/foo" key's value
	log.Print("Requesting data dump...")
	resp, err := kapi.Get(context.Background(), "/ips", &client.GetOptions{Recursive: true})
	if err != nil {
		log.Fatal(err)
	}

	watchIndex := resp.Index
	log.Println("Loading dump...")
	for _, node := range resp.Node.Nodes {
		log.Println(node.Key, node.Value)
	}

	watch := kapi.Watcher("/ips", &client.WatcherOptions{AfterIndex: watchIndex, Recursive: true})

	log.Println("Introducing looooonnnng timer")
	time.Sleep(30 * time.Second)

	ctx := context.Background()
	log.Println("Catching up...")
	for {
		select {
		case <-ctx.Done():
			break
		default:
			change, err := watch.Next(ctx)
			if err != nil {
				break
			}
			log.Println(change.Node.Key, change.Node.Value)
		}
	}
}

output.txt

❯ go run main.go
2018/03/07 22:59:42 Requesting data dump...
2018/03/07 22:59:42 Loading dump...
2018/03/07 22:59:42 /ips/1.1.1.2 allow
2018/03/07 22:59:42 /ips/1.1.1.3 allow
2018/03/07 22:59:42 /ips/1.1.1.4 allow
2018/03/07 22:59:42 /ips/1.1.1.1 allow
2018/03/07 22:59:42 Introducing looooonnnng timer
2018/03/07 23:00:12 Catching up...
2018/03/07 23:00:12 /ips/1.1.1.4 block
2018/03/07 23:00:12 /ips/1.1.1.1 block
2018/03/07 23:00:12 /ips/1.1.1.2 block