Feature watch etcd

checker/etcd_leader_checker.go

@@ -10,14 +10,15 @@
 	"time"
 
 	"github.com/cybertec-postgresql/vip-manager/vipconfig"
-	client "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 // EtcdLeaderChecker is used to check state of the leader key in Etcd
 type EtcdLeaderChecker struct {
 	key      string
 	nodename string
-	kapi     client.KV
+	client   *clientv3.Client
 }
 
 // naming this c_conf to avoid conflict with conf in etcd_leader_checker.go
@@ -71,58 +72,64 @@
 		return nil, err
 	}
 
-	cfg := client.Config{
-		Endpoints:            eConf.Endpoints,
-		TLS:                  tlsConfig,
+	cfg := clientv3.Config{
+		Endpoints: eConf.Endpoints,
+		TLS:       tlsConfig,
 		DialKeepAliveTimeout: 5 * time.Second,
 		DialKeepAliveTime:    5 * time.Second,
 		Username:             eConf.EtcdUser,
 		Password:             eConf.EtcdPassword,
 	}
-	c, err := client.New(cfg)
+	c, err := clientv3.New(cfg)
 	if err != nil {
 		return nil, err
 	}
-	e.kapi = c.KV
+	e.client = c
 	return e, nil
 }
 
 // GetChangeNotificationStream checks the status in the loop
 func (e *EtcdLeaderChecker) GetChangeNotificationStream(ctx context.Context, out chan<- bool) error {
 	var state bool
-	var alreadyConnected = false
-checkLoop:
-	for {
-		resp, err := e.kapi.Get(ctx, e.key)
 
-		if err != nil {
-			if ctx.Err() != nil {
-				break checkLoop
-			}
-			log.Printf("etcd error: %s", err)
-			out <- false
-			time.Sleep(time.Duration(eConf.Interval) * time.Millisecond)
-			continue
-		}
+	defer e.client.Close()
 
-		if !alreadyConnected {
-			log.Printf("etcd checker started up, found key %s", e.key)
-			alreadyConnected = true
+	// get current state from etcd
+	resp, err := e.client.Get(ctx, e.key)
+	if err != nil {
+		if ctx.Err() != nil {
+			return ctx.Err()
 		}
+		log.Printf("etcd error: %s", err)
+		out <- false
+		time.Sleep(time.Duration(eConf.Interval) * time.Millisecond)
+	}
 
-		for _, kv := range resp.Kvs {
-			if eConf.Verbose {
-				log.Println("Leader from DCS:", string(kv.Value))
-			}
-			state = string(kv.Value) == e.nodename
-		}
+	// process responce
+	for _, kv := range resp.Kvs {
+		log.Println("Current Leader from DCS:", string(kv.Value))
+		state = string(kv.Value) == e.nodename
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case out <- state:
+		time.Sleep(time.Duration(eConf.Interval) * time.Millisecond)
+	}
+
+	watchChan := e.client.Watch(ctx, e.key)
+	fmt.Println("set WATCH on " + e.key)
 
-		select {
-		case <-ctx.Done():
-			break checkLoop
-		case out <- state:
-			time.Sleep(time.Duration(eConf.Interval) * time.Millisecond)
-			continue
+	// process watching changes for key
+	for watchResp := range watchChan {
+		for _, event := range watchResp.Events {
+			// new val on key
+			if event.Type == mvccpb.PUT {
+				state = string(event.Kv.Value) == e.nodename
+				out <- state
+			}
+			fmt.Printf("Event received! %s executed on %q with value %q\n", event.Type, event.Kv.Key, event.Kv.Value)
 		}
 	}
 

main.go

@@ -96,6 +96,7 @@ func main() {
 			log.Fatalf("Leader checker returned the following error: %s", err)
 		}
 		wg.Done()
+		panic("Leader checker process fail")
 	}()
 
 	wg.Add(1)