[-] add compacted revision handler, fixes #208

Author: pashagolub

checker/etcd_leader_checker.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"fmt"
 	"log"
 	"os"
 	"time"
 
 	"github.com/cybertec-postgresql/vip-manager/vipconfig"
+	v3rpc "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
@@ -69,8 +71,8 @@ func getTransport(conf *vipconfig.Config) (*tls.Config, error) {
 	return tlsClientConfig, nil
 }
 
-// init gets the current leader from etcd
-func (elc *EtcdLeaderChecker) init(ctx context.Context, out chan<- bool) {
+// get gets the current leader from etcd
+func (elc *EtcdLeaderChecker) get(ctx context.Context, out chan<- bool) {
 	resp, err := elc.Get(ctx, elc.Key)
 	if err != nil {
 		log.Printf("etcd error: %s", err)
@@ -93,8 +95,12 @@ func (elc *EtcdLeaderChecker) watch(ctx context.Context, out chan<- bool) error
 			return ctx.Err()
 		case watchResp := <-watchChan:
 			if err := watchResp.Err(); err != nil {
-				log.Printf("etcd watcher returned error: %s", err)
-				out <- false
+				if errors.Is(err, v3rpc.ErrCompacted) { // revision is compacted, try direct get key
+					elc.get(ctx, out)
+				} else {
+					log.Printf("etcd watcher returned error: %s", err)
+					out <- false
+				}
 				continue
 			}
 			for _, event := range watchResp.Events {
@@ -108,7 +114,7 @@ func (elc *EtcdLeaderChecker) watch(ctx context.Context, out chan<- bool) error
 // GetChangeNotificationStream monitors the leader in etcd
 func (elc *EtcdLeaderChecker) GetChangeNotificationStream(ctx context.Context, out chan<- bool) error {
 	defer elc.Close()
-	go elc.init(ctx, out)
+	go elc.get(ctx, out)
 	wctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	return elc.watch(wctx, out)

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/mdlayher/arp v0.0.0-20220512170110-6706a2966875
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.18.2
+	go.etcd.io/etcd/api/v3 v3.5.13
 	go.etcd.io/etcd/client/v3 v3.5.13
 	golang.org/x/sys v0.19.0
 )
@@ -44,7 +45,6 @@ require (
 	github.com/spf13/afero v1.11.0 // indirect
 	github.com/spf13/cast v1.6.0 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
-	go.etcd.io/etcd/api/v3 v3.5.13 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.13 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect