Merge branch 'pre32' into pre32

Author: jmccormick2001

apiserver/common.go

@@ -24,6 +24,7 @@ import (
 "k8s.io/api/core/v1"
 )
 
+//TODO remove and replace with util.GetSecrets
 func GetSecrets(cluster *crv1.Pgcluster) ([]msgs.ShowUserSecret, error) {
 
 output := make([]msgs.ShowUserSecret, 0)

apiserver/failoverservice/failoverimpl.go

@@ -138,6 +138,8 @@ func QueryFailover(name string) msgs.QueryFailoverResponse {
 log.Debug("found " + dep.Name)
 target := msgs.FailoverTargetSpec{}
 target.Name = dep.Name
+
+target.ReceiveLocation, target.ReplayLocation = util.GetRepStatus(apiserver.RESTClient, apiserver.Clientset, &dep, apiserver.Namespace)
 //get the pod status
 target.ReadyStatus, target.Node = apiserver.GetPodStatus(dep.Name)
 //get the rep status

apiservermsgs/failovermsgs.go

@@ -18,10 +18,12 @@ limitations under the License.
 import ()
 
 type FailoverTargetSpec struct {
-Name string
-ReadyStatus string
-Node string
-RepStatus string
+Name string
+ReadyStatus string
+Node string
+RepStatus string
+ReceiveLocation uint64
+ReplayLocation uint64
 }
 
 // QueryFailoverResponse ...

bin/postgres-operator/runpsql.sh

@@ -21,7 +21,7 @@ FAILURES=0
 MAX_FAILURES=7
 while true; do
  sleep $SLEEP_TIME
- /usr/pgsql-9.6/bin/pg_isready --dbname=postgres --host=$2 --port=5432 --username=postgres
+ /usr/pgsql-10/bin/pg_isready --dbname=postgres --host=$2 --port=5432 --username=postgres
  if [ $? -eq 0 ]
  then
  echo "Successfully reached master @ " `date`

operator/cluster/autofailover.go

@@ -25,6 +25,7 @@ import (
 "github.com/crunchydata/postgres-operator/kubeapi"
 "github.com/crunchydata/postgres-operator/operator"
 "github.com/crunchydata/postgres-operator/util"
+"k8s.io/api/extensions/v1beta1"
 kerrors "k8s.io/apimachinery/pkg/api/errors"
 meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 "k8s.io/client-go/kubernetes"
@@ -256,7 +257,7 @@ func (*AutoFailoverTask) GetEvents(restclient *rest.RESTClient, clusterName, nam
 return "", make(map[string]string)
 }
 
-func getTargetDeployment(clientset *kubernetes.Clientset, clusterName, ns string) (string, error) {
+func getTargetDeployment(restclient *rest.RESTClient, clientset *kubernetes.Clientset, clusterName, ns string) (string, error) {
 
 selector := util.LABEL_PRIMARY + "=false," + util.LABEL_PG_CLUSTER + "=" + clusterName
 
@@ -272,22 +273,44 @@ func getTargetDeployment(clientset *kubernetes.Clientset, clusterName, ns string
 //return a deployment target that has a Ready database
 log.Debugf("deps len %d\n", len(deployments.Items))
 found := false
+readyDeps := make([]v1beta1.Deployment, 0)
 for _, dep := range deployments.Items {
 ready := getPodStatus(clientset, dep.Name, ns)
 if ready {
 log.Debug("autofail: found ready deployment " + dep.Name)
 found = true
-return dep.Name, err
+readyDeps = append(readyDeps, dep)
+//return dep.Name, err
 } else {
 log.Debug("autofail: found not ready deployment " + dep.Name)
 }
 }
+
 if !found {
 log.Error("could not find a ready pod in autofailover for cluster " + clusterName)
 return "", errors.New("could not find a ready pod to failover to")
 }
 
-return "", err
+//at this point readyDeps should hold all the Ready deployments
+//we look for the most up to date and return that name
+
+var value uint64
+value = 0
+var selectedDeployment v1beta1.Deployment
+
+for _, d := range readyDeps {
+target := util.ReplicationInfo{}
+target.ReceiveLocation, target.ReplayLocation = util.GetRepStatus(restclient, clientset, &d, ns)
+log.Debug("autofail receive=%d replay=%d dep=%s\n", target.ReceiveLocation, target.ReplayLocation, d.Name)
+if target.ReceiveLocation > value {
+value = target.ReceiveLocation
+selectedDeployment = d
+}
+
+}
+
+log.Debugf("autofail logic selected deployment is %s receive=%d\n", selectedDeployment.Name, value)
+return selectedDeployment.Name, err
 
 }
 
@@ -314,7 +337,7 @@ func getPodStatus(clientset *kubernetes.Clientset, depname, ns string) bool {
 }
 
 func (s *StateMachine) triggerFailover() {
-targetDeploy, err := getTargetDeployment(s.Clientset, s.ClusterName, s.Namespace)
+targetDeploy, err := getTargetDeployment(s.RESTClient, s.Clientset, s.ClusterName, s.Namespace)
 if targetDeploy == "" || err != nil {
 log.Errorf("could not autofailover with no replicas found for %s\n", s.ClusterName)
 return

pgo/cmd/failover.go

@@ -61,7 +61,7 @@ func init() {
 failoverCmd.Flags().BoolVarP(&Query, "query", "", false, "Prints the list of failover candidates.")
 failoverCmd.Flags().BoolVarP(&NoPrompt, "no-prompt", "n", false, "No command line confirmation.")
 failoverCmd.Flags().StringVarP(&Target, "target", "", "", "The replica target which the failover will occur on.")
-
+
 }
 
 // createFailover ....
@@ -140,7 +140,6 @@ func queryFailover(args []string) {
 fmt.Println("Error: Do: ", err)
 return
 }
-log.Debugf("%v\n", resp)
 StatusCheck(resp)
 
 defer resp.Body.Close()
@@ -158,7 +157,7 @@ func queryFailover(args []string) {
 if len(response.Targets) > 0 {
 fmt.Println("Failover targets include:")
 for i := 0; i < len(response.Targets); i++ {
-fmt.Println("\t" + response.Targets[i].Name + " (" + response.Targets[i].ReadyStatus + ") (" + response.Targets[i].Node + ")")
+printTarget(response.Targets[i])
 }
 }
 for k := range response.Results {
@@ -170,3 +169,7 @@ func queryFailover(args []string) {
 }
 
 }
+
+func printTarget(target msgs.FailoverTargetSpec) {
+fmt.Printf("\t%s (%s) (%s) ReceiveLoc (%d) ReplayLoc (%d)\n", target.Name, target.ReadyStatus, target.Node, target.ReceiveLocation, target.ReplayLocation)
+}

pgo/cmd/scale.go

@@ -172,11 +172,12 @@ func queryCluster(args []string) {
 return
 }
 
+fmt.Println("jeff here we are")
 if response.Status.Code == msgs.Ok {
 if len(response.Targets) > 0 {
 fmt.Println("Replica targets include:")
 for i := 0; i < len(response.Targets); i++ {
-fmt.Println("\t" + response.Targets[i].Name + " (" + response.Targets[i].ReadyStatus + ") (" + response.Targets[i].Node + ")")
+printScaleTarget(response.Targets[i])
 }
 }
 
@@ -232,3 +233,7 @@ func scaleDownCluster(clusterName string) {
 }
 
 }
+
+func printScaleTarget(target msgs.ScaleQueryTargetSpec) {
+fmt.Printf("\t%s (%s) (%s)\n", target.Name, target.ReadyStatus, target.Node)
+}

util/failover.go

@@ -16,14 +16,37 @@ package util
 */
 
 import (
+"database/sql"
 "errors"
+"fmt"
 log "github.com/Sirupsen/logrus"
+crv1 "github.com/crunchydata/postgres-operator/apis/cr/v1"
+msgs "github.com/crunchydata/postgres-operator/apiservermsgs"
 "github.com/crunchydata/postgres-operator/kubeapi"
+_ "github.com/lib/pq"
 "k8s.io/api/core/v1"
 "k8s.io/api/extensions/v1beta1"
 "k8s.io/client-go/kubernetes"
+"k8s.io/client-go/rest"
 )
 
+const (
+replInfoQueryFormat = "SELECT %s(%s(), '0/0')::bigint, %s(%s(), '0/0')::bigint"
+
+recvV9 = "pg_last_xlog_receive_location"
+replayV9 = "pg_last_xlog_replay_location"
+locationDiffV9 = "pg_xlog_location_diff"
+
+recvV10 = "pg_last_wal_receive_lsn"
+replayV10 = "pg_last_wal_replay_lsn"
+locationDiffV10 = "pg_wal_lsn_diff"
+)
+
+type ReplicationInfo struct {
+ReceiveLocation uint64
+ReplayLocation uint64
+}
+
 // GetBestTarget
 func GetBestTarget(clientset *kubernetes.Clientset, clusterName, namespace string) (*v1.Pod, *v1beta1.Deployment, error) {
 
@@ -101,3 +124,139 @@ func GetPod(clientset *kubernetes.Clientset, deploymentName, namespace string) (
 
 return pod, err
 }
+
+func GetRepStatus(restclient *rest.RESTClient, clientset *kubernetes.Clientset, dep *v1beta1.Deployment, namespace string) (uint64, uint64) {
+var receiveLocation, replayLocation uint64
+
+//get the pods for this deployment
+selector := "primary=false,replica-name=" + dep.Name
+podList, err := kubeapi.GetPods(clientset, selector, namespace)
+if err != nil {
+log.Error(err.Error())
+return receiveLocation, replayLocation
+}
+
+if len(podList.Items) != 1 {
+log.Debug("no replicas found for dep " + dep.Name)
+return receiveLocation, replayLocation
+}
+
+pod := podList.Items[0]
+
+//get the crd for this dep
+cluster := crv1.Pgcluster{}
+var clusterfound bool
+clusterfound, err = kubeapi.Getpgcluster(restclient, &cluster, dep.ObjectMeta.Labels[LABEL_PG_CLUSTER], namespace)
+if err != nil || !clusterfound {
+log.Error("Getpgcluster error: " + err.Error())
+return receiveLocation, replayLocation
+}
+
+//get the postgres secret for this dep
+var secretInfo []msgs.ShowUserSecret
+secretInfo, err = GetSecrets(clientset, &cluster, namespace)
+var pgSecret msgs.ShowUserSecret
+var found bool
+for _, si := range secretInfo {
+if si.Username == "postgres" {
+pgSecret = si
+found = true
+log.Debug("postgres secret found")
+}
+}
+
+if !found {
+log.Error("postgres secret not found for " + dep.Name)
+return receiveLocation, replayLocation
+}
+
+port := "5432"
+databaseName := "postgres"
+target := getSQLTarget(&pod, pgSecret.Username, pgSecret.Password, port, databaseName)
+var repInfo *ReplicationInfo
+repInfo, err = GetReplicationInfo(target)
+if err != nil {
+log.Error(err)
+return receiveLocation, replayLocation
+}
+
+receiveLocation = repInfo.ReceiveLocation
+replayLocation = repInfo.ReplayLocation
+return receiveLocation, replayLocation
+}
+
+func getSQLTarget(pod *v1.Pod, username, password, port, db string) string {
+target := fmt.Sprintf(
+"postgresql://%s:%s@%s:%s/%s?sslmode=disable",
+username,
+password,
+pod.Status.PodIP,
+port,
+db,
+)
+return target
+
+}
+func GetReplicationInfo(target string) (*ReplicationInfo, error) {
+conn, err := sql.Open("postgres", target)
+
+if err != nil {
+log.Errorf("Could not connect to: %s", target)
+return nil, err
+}
+
+defer conn.Close()
+
+// Get PG version
+var version int
+
+rows, err := conn.Query("SELECT current_setting('server_version_num')")
+
+if err != nil {
+log.Errorf("Could not perform query for version: %s", target)
+return nil, err
+}
+
+defer rows.Close()
+
+for rows.Next() {
+if err := rows.Scan(&version); err != nil {
+return nil, err
+}
+}
+// Get replication info
+var replicationInfoQuery string
+var recvLocation uint64
+var replayLocation uint64
+
+if version < 100000 {
+replicationInfoQuery = fmt.Sprintf(
+replInfoQueryFormat,
+locationDiffV9, recvV9,
+locationDiffV9, replayV9,
+)
+} else {
+replicationInfoQuery = fmt.Sprintf(
+replInfoQueryFormat,
+locationDiffV10, recvV10,
+locationDiffV10, replayV10,
+)
+}
+
+rows, err = conn.Query(replicationInfoQuery)
+
+if err != nil {
+log.Errorf("Could not perform replication info query: %s", target)
+return nil, err
+}
+
+defer rows.Close()
+
+for rows.Next() {
+if err := rows.Scan(&recvLocation, &replayLocation); err != nil {
+return nil, err
+}
+}
+
+return &ReplicationInfo{recvLocation, replayLocation}, nil
+}