use uint64 for replication lag from Patroni's member endpoint (#1893)

* use int64 for replication lag from Patroni's member endpoint

Author: FxKu

pkg/cluster/pod.go

@@ -522,13 +522,13 @@ func (c *Cluster) getSwitchoverCandidate(master *v1.Pod) (spec.NamespacedName, e
 	// if sync_standby replicas were found assume synchronous_mode is enabled and ignore other candidates list
 	if len(syncCandidates) > 0 {
 		sort.Slice(syncCandidates, func(i, j int) bool {
-			return util.IntFromIntStr(syncCandidates[i].Lag) < util.IntFromIntStr(syncCandidates[j].Lag)
+			return syncCandidates[i].Lag < syncCandidates[j].Lag
 		})
 		return spec.NamespacedName{Namespace: master.Namespace, Name: syncCandidates[0].Name}, nil
 	}
 	if len(candidates) > 0 {
 		sort.Slice(candidates, func(i, j int) bool {
-			return util.IntFromIntStr(candidates[i].Lag) < util.IntFromIntStr(candidates[j].Lag)
+			return candidates[i].Lag < candidates[j].Lag
 		})
 		return spec.NamespacedName{Namespace: master.Namespace, Name: candidates[0].Name}, nil
 	}

pkg/util/patroni/patroni.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"math"
 	"net"
 	"net/http"
 	"strconv"
@@ -16,7 +17,6 @@ import (
 	"github.com/sirupsen/logrus"
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
 const (
@@ -185,11 +185,27 @@ type ClusterMembers struct {
 
 // ClusterMember cluster member data from Patroni API
 type ClusterMember struct {
-	Name     string             `json:"name"`
-	Role     string             `json:"role"`
-	State    string             `json:"state"`
-	Timeline int                `json:"timeline"`
-	Lag      intstr.IntOrString `json:"lag,omitempty"`
+	Name     string         `json:"name"`
+	Role     string         `json:"role"`
+	State    string         `json:"state"`
+	Timeline int            `json:"timeline"`
+	Lag      ReplicationLag `json:"lag,omitempty"`
+}
+
+type ReplicationLag uint64
+
+// UnmarshalJSON converts member lag (can be int or string) into uint64
+func (rl *ReplicationLag) UnmarshalJSON(data []byte) error {
+	var lagUInt64 uint64
+	if data[0] == '"' {
+		*rl = math.MaxUint64
+		return nil
+	}
+	if err := json.Unmarshal(data, &lagUInt64); err != nil {
+		return err
+	}
+	*rl = ReplicationLag(lagUInt64)
+	return nil
 }
 
 // MemberDataPatroni child element

pkg/util/patroni/patroni_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io/ioutil"
+	"math"
 	"net/http"
 	"reflect"
 	"testing"
@@ -15,7 +16,6 @@ import (
 
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
 var logger = logrus.New().WithField("test", "patroni")
@@ -101,16 +101,27 @@ func TestGetClusterMembers(t *testing.T) {
 			Role:     "sync_standby",
 			State:    "running",
 			Timeline: 1,
-			Lag:      intstr.IntOrString{IntVal: 0},
+			Lag:      0,
 		}, {
 			Name:     "acid-test-cluster-2",
 			Role:     "replica",
 			State:    "running",
 			Timeline: 1,
-			Lag:      intstr.IntOrString{Type: 1, StrVal: "unknown"},
+			Lag:      math.MaxUint64,
+		}, {
+			Name:     "acid-test-cluster-3",
+			Role:     "replica",
+			State:    "running",
+			Timeline: 1,
+			Lag:      3000000000,
 		}}
 
-	json := `{"members": [{"name": "acid-test-cluster-0", "role": "leader", "state": "running", "api_url": "http://192.168.100.1:8008/patroni", "host": "192.168.100.1", "port": 5432, "timeline": 1}, {"name": "acid-test-cluster-1", "role": "sync_standby", "state": "running", "api_url": "http://192.168.100.2:8008/patroni", "host": "192.168.100.2", "port": 5432, "timeline": 1, "lag": 0}, {"name": "acid-test-cluster-2", "role": "replica", "state": "running", "api_url": "http://192.168.100.3:8008/patroni", "host": "192.168.100.3", "port": 5432, "timeline": 1, "lag": "unknown"}]}`
+	json := `{"members": [
+		{"name": "acid-test-cluster-0", "role": "leader", "state": "running", "api_url": "http://192.168.100.1:8008/patroni", "host": "192.168.100.1", "port": 5432, "timeline": 1},
+		{"name": "acid-test-cluster-1", "role": "sync_standby", "state": "running", "api_url": "http://192.168.100.2:8008/patroni", "host": "192.168.100.2", "port": 5432, "timeline": 1, "lag": 0},
+		{"name": "acid-test-cluster-2", "role": "replica", "state": "running", "api_url": "http://192.168.100.3:8008/patroni", "host": "192.168.100.3", "port": 5432, "timeline": 1, "lag": "unknown"},
+		{"name": "acid-test-cluster-3", "role": "replica", "state": "running", "api_url": "http://192.168.100.3:8008/patroni", "host": "192.168.100.3", "port": 5432, "timeline": 1, "lag": 3000000000}
+		]}`
 	r := ioutil.NopCloser(bytes.NewReader([]byte(json)))
 
 	response := http.Response{

pkg/util/util.go

@@ -8,7 +8,6 @@ import (
 	"encoding/base64"
 	"encoding/hex"
 	"fmt"
-	"math"
 	"math/big"
 	"math/rand"
 	"reflect"
@@ -338,14 +337,6 @@ func Bool2Int(flag bool) int {
 	return 0
 }
 
-// Get int from IntOrString and return max int if string
-func IntFromIntStr(intOrStr intstr.IntOrString) int {
-	if intOrStr.Type == 1 {
-		return math.MaxInt
-	}
-	return intOrStr.IntValue()
-}
-
 // MaxInt32 : Return maximum of two integers provided via pointers. If one value
 // is not defined, return the other one. If both are not defined, result is also
 // undefined, caller needs to check for that.