[-] fix Patroni resolver, fixes #962 (#973)

Add additional check for the length of the key and if it contains
"members" part. Check both `primary` and `master` roles for backward
compatibility. Add enhanced testing with real Patroni cluster data

Author: pashagolub

internal/sources/resolver.go

@@ -57,6 +57,10 @@ type PatroniClusterMember struct {
 	Role    string
 }
 
+func (pcm PatroniClusterMember) IsPrimary() bool {
+	return pcm.Role == "primary" || pcm.Role == "master"
+}
+
 var logger log.Logger = log.FallbackLogger
 
 var lastFoundClusterMembers = make(map[string][]PatroniClusterMember) // needed for cases where DCS is temporarily down
@@ -171,8 +175,8 @@ func getEtcdClusterMembers(s Source, hc HostConfig) ([]PatroniClusterMember, err
 		}
 		// remove leading slash and split by "/"
 		parts := strings.Split(strings.TrimPrefix(string(node.Key), "/"), "/")
-		if len(parts) < 3 {
-			return nil, errors.New("invalid ETCD key format")
+		if len(parts) < 4 || parts[2] != "members" {
+			continue // skip non-member keys
 		}
 		role := nodeData["role"]
 		connURL := nodeData["conn_url"]
@@ -281,12 +285,12 @@ func ResolveDatabasesFromPatroni(source Source) (SourceConns, error) {
 
 	for _, patroniMember := range clusterMembers {
 		logger.Info("Processing Patroni cluster member: ", patroniMember.Name)
-		if source.OnlyIfMaster && patroniMember.Role != "master" {
+		if source.OnlyIfMaster && !patroniMember.IsPrimary() {
 			continue
 		}
 		src := *source.Clone()
 		src.ConnStr = patroniMember.ConnURL
-		if hostConfig.IsScopeSpecified() {
+		if !hostConfig.IsScopeSpecified() {
 			src.Name += "_" + patroniMember.Scope
 		}
 		src.Name += "_" + patroniMember.Name

internal/sources/resolver_test.go

@@ -3,7 +3,6 @@ package sources_test
 import (
 	"context"
 	"errors"
-	"fmt"
 	"strings"
 	"testing"
 	"time"
@@ -54,6 +53,7 @@ func TestMonitoredDatabase_ResolveDatabasesFromPostgres(t *testing.T) {
 }
 
 func TestMonitoredDatabase_ResolveDatabasesFromPatroni(t *testing.T) {
+
 	etcdContainer, err := etcd.Run(ctx, "gcr.io/etcd-development/etcd:v3.5.14",
 		testcontainers.WithWaitStrategy(wait.ForLog("ready to serve client requests").
 			WithStartupTimeout(15*time.Second)))
@@ -82,18 +82,24 @@ func TestMonitoredDatabase_ResolveDatabasesFromPatroni(t *testing.T) {
 	)
 	require.NoError(t, err)
 	defer func() { assert.NoError(t, pgContainer.Terminate(ctx)) }()
-
-	// Put values to etcd server
-	cancelCtx, cancel := context.WithTimeout(context.Background(), time.Second)
-	defer cancel()
-	connStr, err := pgContainer.ConnectionString(cancelCtx, "sslmode=disable")
-	require.NoError(t, err)
-	_, err = cli.Put(cancelCtx, "/service/batman/members/pg1",
-		fmt.Sprintf(`{"role":"master","conn_url":"%s"}`, connStr))
-	require.NoError(t, err)
-	_, err = cli.Put(cancelCtx, "/service/batman/members/pg2",
-		`{"role":"standby","conn_url":"must_be_skipped"}`)
+	pgConnStr, err := pgContainer.ConnectionString(ctx, "sslmode=disable")
 	require.NoError(t, err)
+	// Put values to etcd server
+	kv := map[string]string{
+		`/service/demo/config`:           `{"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":1048576,"postgresql":{"use_pg_rewind":true,"pg_hba":["local all all trust","host replication replicator all md5","host all all all md5"],"parameters":{"max_connections":100}}}`,
+		`/service/demo/initialize`:       `7553211779477532695`,
+		`/service/demo/leader`:           `patroni3`,
+		`/service/demo/members/patroni1`: `{"conn_url":"postgres://172.18.0.8:5432/postgres","api_url":"http://172.18.0.8:8008/patroni","state":"running","role":"replica","version":"4.0.7","xlog_location":67108960,"replay_lsn":67108960,"receive_lsn":67108960,"replication_state":"streaming","timeline":1}`,
+		`/service/demo/members/patroni2`: `{"conn_url":"postgres://172.18.0.4:5432/postgres","api_url":"http://172.18.0.4:8008/patroni","state":"running","role":"replica","version":"4.0.7","xlog_location":67108960,"replay_lsn":67108960,"receive_lsn":67108960,"replication_state":"streaming","timeline":1}`,
+		`/service/demo/members/patroni3`: `{"conn_url":"` + pgConnStr + `","api_url":"http://172.18.0.3:8008/patroni","state":"running","role":"primary","version":"4.0.7","xlog_location":67108960,"timeline":1}`,
+		`/service/demo/status`:           `{"optime":67108960,"slots":{"patroni1":67108960,"patroni2":67108960,"patroni3":67108960},"retain_slots":["patroni1","patroni2","patroni3"]}}`}
+
+	cancelCtx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	for k, v := range kv {
+		_, err = cli.Put(cancelCtx, k, v)
+		require.NoError(t, err, "failed to put key %s to etcd", k)
+	}
+	cancel()
 
 	md := sources.Source{}
 	md.Name = "continuous"
@@ -103,7 +109,7 @@ func TestMonitoredDatabase_ResolveDatabasesFromPatroni(t *testing.T) {
 		md.Kind = sources.SourcePatroni
 		md.ConnStr = "etcd://" + strings.TrimPrefix(endpoint, "http://")
 		md.ConnStr += "/service"
-		md.ConnStr += "/batman"
+		md.ConnStr += "/demo"
 
 		// Run ResolveDatabasesFromPatroni
 		dbs, err := md.ResolveDatabases()
@@ -117,7 +123,7 @@ func TestMonitoredDatabase_ResolveDatabasesFromPatroni(t *testing.T) {
 		e := strings.TrimPrefix(endpoint, "http://")
 		md.ConnStr = "etcd://" + strings.Join([]string{e, e, e}, ",")
 		md.ConnStr += "/service"
-		md.ConnStr += "/batman"
+		md.ConnStr += "/demo"
 
 		// Run ResolveDatabasesFromPatroni
 		dbs, err := md.ResolveDatabases()