[WIP] Implement more v3 endpoints

Makefile

@@ -73,9 +73,15 @@ proto:
 		--grpc-gateway_opt logtostderr=true \
 		--grpc-gateway_opt paths=source_relative \
 		--grpc-gateway_out repmanv3 \
-		--openapiv2_out repmanv3 \
+		--openapiv2_out swagger \
 		--openapiv2_opt logtostderr=true \
 		--openapiv2_opt allow_merge=true \
 		--openapiv2_opt merge_file_name=repmanv3 \
 		-orepmanv3/service.desc \
 		${PROTO_DIR}/cluster.proto ${PROTO_DIR}/messages.proto
+
+	@protoc/bin/protoc \
+		-I ${PROTO_DIR} \
+		--go_opt=paths=source_relative \
+		--go_out=utils/gtid \
+		${PROTO_DIR}/gtid.proto

clients/client_cmd.go

@@ -11,10 +11,12 @@ package clients
 
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net/http"
 	"net/url"
@@ -32,6 +34,8 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 	"golang.org/x/crypto/ssh/terminal"
+
+	v3 "github.com/signal18/replication-manager/repmanv3"
 )
 
 var (
@@ -77,6 +81,8 @@ var (
 	cliConfirm                   string
 	cfgGroup                     string
 	memprofile                   string
+
+	v3Config *v3.ClientConfig
 )
 
 type RequetParam struct {
@@ -157,6 +163,20 @@ func cliInit(needcluster bool) {
 		log.WithError(err).Fatal()
 		return
 	}
+
+	// the legacy code has already received all information we need
+	v3Config = &v3.ClientConfig{
+		Address: cliHost + ":" + cliPort,
+		TLS:     true,
+		Auth: &v3.LoginCreds{
+			Username: cliUser,
+			Password: cliPassword,
+		},
+	}
+
+	if cliNoCheckCert {
+		v3Config.InsecureSkipVerify = true
+	}
 }
 
 func cliClusterInServerList() bool {
@@ -286,6 +306,7 @@ func init() {
 
 	rootClientCmd.AddCommand(statusCmd)
 	initStatusFlags(statusCmd)
+	initClusterFlags(statusCmd)
 
 	rootClientCmd.AddCommand(bootstrapCmd)
 	initBootstrapFlags(bootstrapCmd)
@@ -563,35 +584,33 @@ func cliGetMaster() (cluster.ServerMonitor, error) {
 }
 
 func cliGetLogs() ([]string, error) {
-	var r []string
-	urlpost := "https://" + cliHost + ":" + cliPort + "/api/clusters/" + cliClusters[cliClusterIndex] + "/topology/logs"
-	var bearer = "Bearer " + cliToken
-	req, err := http.NewRequest("GET", urlpost, nil)
+	client, err := v3.NewClient(context.Background(), v3Config)
 	if err != nil {
-		return r, err
+		return []string{}, fmt.Errorf("Could not initialize v3 Client: %s", err)
 	}
-	req.Header.Set("Authorization", bearer)
-	resp, err := cliConn.Do(req)
-	if err != nil {
-		log.Println("ERROR on getting logs ", err)
-		return r, err
-	}
-	defer resp.Body.Close()
-	body, err := ioutil.ReadAll(resp.Body)
+
+	stream, err := client.RetrieveLogs(context.Background(), &v3.Cluster{Name: cliClusters[cliClusterIndex]})
 	if err != nil {
-		log.Println("ERROR on getting logs", err)
-		return r, err
+		return []string{}, fmt.Errorf("Error fetching RetrieveLogs: %s", err)
 	}
-	if resp.StatusCode == http.StatusForbidden {
-		return r, errors.New("Wrong credentential")
-	}
-	err = json.Unmarshal(body, &r)
-	if err != nil {
-		log.Println("ERROR on getting logs", err)
-		return r, err
+
+	var logs []string
+
+	// TODO: we should directly stream the data into the termlog
+	for {
+		recv, err := stream.Recv()
+		if err == io.EOF {
+			break
+		}
+
+		if err != nil {
+			log.Fatalf("Error receiving stream: %s", err)
+		}
+
+		logs = append(logs, recv.Value)
 	}
-	resp.Body.Close()
-	return r, nil
+
+	return logs, nil
 }
 
 func cliClusterCmd(command string, params []RequetParam) error {

clients/client_show.go

@@ -9,83 +9,150 @@
 package clients
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
+	"io"
+	"log"
 	"os"
 	"strings"
 
 	"github.com/signal18/replication-manager/cluster"
 	"github.com/signal18/replication-manager/server"
+	"github.com/signal18/replication-manager/utils/state"
 	"github.com/spf13/cobra"
+	"google.golang.org/protobuf/encoding/protojson"
+
+	v3 "github.com/signal18/replication-manager/repmanv3"
 )
 
 var showCmd = &cobra.Command{
 	Use:   "show",
 	Short: "Print json informations",
 	Long:  `To use for support issues`,
 	Run: func(cmd *cobra.Command, args []string) {
-		//cliClusters, err = cliGetClusters()
 		cliInit(false)
-		urlpost := ""
 		type Objects struct {
 			Name     string
 			Settings server.Settings         `json:"settings"`
 			Servers  []cluster.ServerMonitor `json:"servers"`
 			Master   cluster.ServerMonitor   `json:"master"`
 			Slaves   []cluster.ServerMonitor `json:"slaves"`
-			Crashes  []cluster.Crash         `json:"crashes"`
+			Crashes  []v3.Crash              `json:"crashes"`
 			Alerts   cluster.Alerts          `json:"alerts"`
 		}
 		type Report struct {
 			Clusters []Objects `json:"clusters"`
 		}
 		var myReport Report
 
-		for _, cluster := range cliClusters {
+		client, err := v3.NewClient(context.Background(), v3Config)
+		if err != nil {
+			log.Fatal("Could not initialize v3 Client: %s", err)
+		}
+
+		for _, clusterName := range cliClusters {
 
 			var myObjects Objects
-			myObjects.Name = cluster
+			myObjects.Name = clusterName
 			if strings.Contains(cliShowObjects, "settings") {
-				urlpost = "https://" + cliHost + ":" + cliPort + "/api/clusters/" + cluster
-				res, err := cliAPICmd(urlpost, nil)
-				if err == nil {
+				c, err := client.GetCluster(context.Background(), &v3.Cluster{
+					Name: clusterName,
+				})
+				if err != nil {
+					log.Fatal("Error fetching GetCluster: %s", err)
+				}
 
-					json.Unmarshal([]byte(res), &myObjects.Settings)
+				// to get the Cluster response back into the Settings format we need to once
+				// convert it to JSON and unmarshall it again :)
+				buf, err := protojson.Marshal(c)
+				// buf, err := json.Marshal(c)
+				if err == nil {
+					json.Unmarshal(buf, &myObjects.Settings)
 				}
 			}
 			if strings.Contains(cliShowObjects, "servers") {
-				urlpost = "https://" + cliHost + ":" + cliPort + "/api/clusters/" + cluster + "/topology/servers"
-				res, err := cliAPICmd(urlpost, nil)
-				if err == nil {
-					json.Unmarshal([]byte(res), &myObjects.Servers)
+				srv, err := RetrieveServerMonitorsFromTopology(client, clusterName, v3.TopologyRetrieval_SERVERS)
+				if err != nil {
+					log.Fatalf("Could not retrieve ServerMonitors: %s", err)
 				}
+				myObjects.Servers = append(myObjects.Servers, srv...)
 			}
 			if strings.Contains(cliShowObjects, "master") {
-				urlpost = "https://" + cliHost + ":" + cliPort + "/api/clusters/" + cluster + "/topology/master"
-				res, err := cliAPICmd(urlpost, nil)
-				if err == nil {
-					json.Unmarshal([]byte(res), &myObjects.Master)
+				srv, err := RetrieveServerMonitorsFromTopology(client, clusterName, v3.TopologyRetrieval_MASTER)
+				if err != nil {
+					if !strings.Contains(err.Error(), v3.ErrClusterMasterNotSet.Error()) {
+						log.Fatalf("Could not retrieve ServerMonitors: %s", err)
+					}
+				}
+				if srv != nil {
+					myObjects.Master = srv[0]
 				}
 			}
 			if strings.Contains(cliShowObjects, "slaves") {
-				urlpost = "https://" + cliHost + ":" + cliPort + "/api/clusters/" + cluster + "/topology/master"
-				res, err := cliAPICmd(urlpost, nil)
-				if err == nil {
-					json.Unmarshal([]byte(res), &myObjects.Slaves)
+				srv, err := RetrieveServerMonitorsFromTopology(client, clusterName, v3.TopologyRetrieval_SLAVES)
+				if err != nil {
+					log.Fatalf("Could not retrieve ServerMonitors: %s", err)
 				}
+				myObjects.Slaves = append(myObjects.Slaves, srv...)
 			}
 			if strings.Contains(cliShowObjects, "crashes") {
-				urlpost = "https://" + cliHost + ":" + cliPort + "/api/clusters/" + cluster + "/topology/crashes"
-				res, err := cliAPICmd(urlpost, nil)
-				if err == nil {
-					json.Unmarshal([]byte(res), &myObjects.Crashes)
+				stream, err := client.RetrieveCrashes(context.Background(), &v3.Cluster{
+					Name: clusterName,
+				})
+
+				if err != nil {
+					log.Fatal("Error fetching RetrieveCrashes: %s", err)
+				}
+
+				for {
+					recv, err := stream.Recv()
+					if err == io.EOF {
+						break
+					}
+
+					if err != nil {
+						log.Fatalf("Error receiving stream: %s", err)
+					}
+
+					myObjects.Crashes = append(myObjects.Crashes, *recv)
 				}
 			}
+
 			if strings.Contains(cliShowObjects, "alerts") {
-				urlpost = "https://" + cliHost + ":" + cliPort + "/api/clusters/" + cluster + "/topology/alerts"
-				res, err := cliAPICmd(urlpost, nil)
-				if err == nil {
-					json.Unmarshal([]byte(res), &myObjects.Alerts)
+				stream, err := client.RetrieveAlerts(context.Background(), &v3.Cluster{
+					Name: clusterName,
+				})
+
+				if err != nil {
+					log.Fatal("Error fetching RetrieveAlerts: %s", err)
+				}
+
+				for {
+					recv, err := stream.Recv()
+					if err == io.EOF {
+						break
+					}
+
+					if err != nil {
+						log.Fatalf("Error receiving stream: %s", err)
+					}
+
+					if recv.Severity == v3.StateMessage_WARNING {
+						myObjects.Alerts.Warnings = append(myObjects.Alerts.Warnings, state.StateHttp{
+							ErrNumber: recv.Number,
+							ErrFrom:   recv.From,
+							ErrDesc:   recv.Desc,
+						})
+					}
+
+					if recv.Severity == v3.StateMessage_ERROR {
+						myObjects.Alerts.Errors = append(myObjects.Alerts.Errors, state.StateHttp{
+							ErrNumber: recv.Number,
+							ErrFrom:   recv.From,
+							ErrDesc:   recv.Desc,
+						})
+					}
 				}
 			}
 			myReport.Clusters = append(myReport.Clusters, myObjects)
@@ -103,3 +170,44 @@ var showCmd = &cobra.Command{
 
 	},
 }
+
+func RetrieveServerMonitorsFromTopology(client *v3.Client, clusterName string, retrieve v3.TopologyRetrieval_Retrieval) ([]cluster.ServerMonitor, error) {
+	var list []cluster.ServerMonitor
+	stream, err := client.RetrieveFromTopology(context.Background(),
+		&v3.TopologyRetrieval{
+			Cluster: &v3.Cluster{
+				Name: clusterName,
+			},
+			Retrieve: retrieve,
+		})
+
+	if err != nil {
+		return nil, fmt.Errorf("Error fetching RetrieveFromTopology: %s", err)
+	}
+
+	for {
+		st, err := stream.Recv()
+		if err == io.EOF {
+			break
+		}
+
+		if err != nil {
+			return nil, fmt.Errorf("Error retrieving topology: %s", err)
+		}
+
+		buf, err := st.MarshalJSON()
+		if err != nil {
+			return nil, fmt.Errorf("Error marshalling servermonitor: %s", err)
+		}
+
+		var smBuf cluster.ServerMonitor
+		err = json.Unmarshal(buf, &smBuf)
+		if err != nil {
+			return nil, fmt.Errorf("Error unmarshalling servermonitor: %s", err)
+		}
+
+		list = append(list, smBuf)
+	}
+
+	return list, nil
+}