Prevent to switch failover mode to 'eventual' in case of new replica set has been added into a cluster

Author: Kasen

controllers/cluster_controller.go

@@ -437,16 +437,24 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		if stsAnnotations["tarantool.io/failoverEnabled"] == "1" {
 			reqLogger.Info("failover is enabled, not retrying")
 		} else {
-			if err := topologyClient.SetFailover(true); err != nil {
-				reqLogger.Error(err, "failed to enable cluster failover")
-			} else {
-				reqLogger.Info("enabled failover")
+			enabled, err := topologyClient.GetFailover()
+			if err != nil {
+				reqLogger.Error(err, "failed to get failover status")
+				continue
+			}
 
-				stsAnnotations["tarantool.io/failoverEnabled"] = "1"
-				sts.SetAnnotations(stsAnnotations)
-				if err := r.Update(context.TODO(), &sts); err != nil {
-					reqLogger.Error(err, "failed to set failover enabled annotation")
+			if !enabled {
+				if err := topologyClient.SetFailover(true); err != nil {
+					reqLogger.Error(err, "failed to enable cluster failover")
+					continue
 				}
+				reqLogger.Info("enabled failover")
+			}
+
+			stsAnnotations["tarantool.io/failoverEnabled"] = "1"
+			sts.SetAnnotations(stsAnnotations)
+			if err := r.Update(context.TODO(), &sts); err != nil {
+				reqLogger.Error(err, "failed to set failover enabled annotation")
 			}
 		}
 	}

controllers/topology/builtin.go

@@ -53,14 +53,15 @@ type BootstrapVshardResponse struct {
 	Errors []*ResponseError     `json:"errors,omitempty"`
 }
 
-// FailoverData Structure of data for changing failover status
-type FailoverData struct {
+type FailoverParams struct {
+	Mode string `json:"mode"`
 }
 
-// FailoverResponse type struct for returning on failovers
-type FailoverResponse struct {
-	Data   *FailoverData
-	Errors []*ResponseError
+type ClusterData struct {
+	FailoverParams *FailoverParams `json:"failover_params"`
+}
+type FailoverResponseData struct {
+	Cluster *ClusterData `json:"cluster"`
 }
 
 // BuiltInTopologyService .
@@ -171,6 +172,22 @@ var getServerStatQuery = `query serverList {
 	}
 }`
 
+var getFailoverStateQuery = `query {
+	cluster {
+		failover_params {
+			mode
+		}
+	}
+}`
+
+var setFailoverStateQuery = `mutation setFailoverMode($mode: String) {
+	cluster {
+		failover_params(mode: $mode) {
+		  mode
+		}
+	}
+}`
+
 // An interface describing an object with accessor methods for labels and annotations
 type ObjectWithMeta interface {
 	GetLabels() map[string]string
@@ -290,11 +307,15 @@ func (s *BuiltInTopologyService) Join(pod *corev1.Pod) error {
 // SetFailover enables cluster failover
 func (s *BuiltInTopologyService) SetFailover(enabled bool) error {
 	client := graphql.NewClient(s.serviceHost, graphql.WithHTTPClient(&http.Client{Timeout: time.Duration(time.Second * 5)}))
-	req := graphql.NewRequest(`mutation changeFailover($enabled: Boolean!) { cluster { failover(enabled: $enabled) }}`)
+	req := graphql.NewRequest(setFailoverStateQuery)
 
-	req.Var("enabled", enabled)
+	mode := "eventual"
+	if !enabled {
+		mode = "disabled"
+	}
+	req.Var("mode", mode)
 
-	resp := &FailoverData{}
+	resp := &FailoverResponseData{}
 	if err := client.Run(context.TODO(), req, resp); err != nil {
 		log.Error(err, "failoverError")
 		return errors.New("failed to enable cluster failover")
@@ -303,6 +324,25 @@ func (s *BuiltInTopologyService) SetFailover(enabled bool) error {
 	return nil
 }
 
+func (s *BuiltInTopologyService) GetFailover() (bool, error) {
+	client := graphql.NewClient(s.serviceHost, graphql.WithHTTPClient(&http.Client{Timeout: time.Duration(time.Second * 5)}))
+	req := graphql.NewRequest(getFailoverStateQuery)
+
+	resp := &FailoverResponseData{}
+	if err := client.Run(context.TODO(), req, resp); err != nil {
+		log.Error(err, "failoverError")
+		return false, errors.New("failed to get info about failover")
+	}
+
+	if resp == nil || resp.Cluster == nil || resp.Cluster.FailoverParams == nil || resp.Cluster.FailoverParams.Mode == "" {
+		return false, errors.New("failed to get info about failover: broken gql response")
+	}
+
+	mode := resp.Cluster.FailoverParams.Mode
+
+	return mode != "disabled", nil
+}
+
 // Expel removes an instance from the replicaset
 func (s *BuiltInTopologyService) Expel(pod *corev1.Pod) error {
 	req := fmt.Sprintf("mutation {expel_instance:expel_server(uuid:\\\"%s\\\")}", pod.GetAnnotations()["tarantool.io/instance_uuid"])

controllers/topology/builtin_test.go

@@ -1,6 +1,11 @@
 package topology
 
 import (
+	"encoding/json"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 
@@ -149,3 +154,167 @@ func TestGetRoles_ParseRolesFromAnnotations(t *testing.T) {
 		}
 	}
 }
+
+type FailoverVariables struct {
+	Mode string `json:"mode"`
+}
+
+type FailoverQuery struct {
+	Query     string            `json:"query"`
+	Variables FailoverVariables `json:"variables"`
+}
+
+var setFailoverGQL = `mutation setFailoverMode($mode: String) {
+	cluster {
+		failover_params(mode: $mode) {
+		  mode
+		}
+	}
+}`
+
+func TestSetFailover(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			t.Fatalf("%s", err)
+		}
+		query := FailoverQuery{}
+		if err = json.Unmarshal(b, &query); err != nil {
+			t.Fatalf("Wrong qeury: %s", err)
+		}
+
+		if query.Query != setFailoverGQL {
+			t.Fatalf("Wrong query: %s", query.Query)
+		}
+
+		if query.Variables.Mode != "eventual" {
+			t.Fatalf("Wrong failover type: %s", query.Variables.Mode)
+		}
+
+		_, _ = io.WriteString(w, `{
+			"data": {
+			  "cluster": {
+				"failover_params": {
+				  "mode": "eventual"
+				}
+			  }
+			}
+		}`)
+	}))
+
+	defer srv.Close()
+
+	topology := BuiltInTopologyService{
+		serviceHost: srv.URL,
+		clusterID:   "uuid",
+	}
+
+	err := topology.SetFailover(true)
+	if err != nil {
+		t.Fatalf("%s", err)
+	}
+}
+
+var getFailoverGQL = `query {
+	cluster {
+		failover_params {
+			mode
+		}
+	}
+}`
+
+func TestGetFailover(t *testing.T) {
+	var failoverMode string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			t.Fatalf("%s", err)
+		}
+		query := FailoverQuery{}
+		if err = json.Unmarshal(b, &query); err != nil {
+			t.Fatalf("Wrong qeury: %s", err)
+		}
+
+		if query.Query != getFailoverGQL {
+			t.Fatalf("Wrong query: %s", query.Query)
+		}
+
+		_, _ = io.WriteString(w, failoverMode)
+	}))
+
+	defer srv.Close()
+
+	topology := BuiltInTopologyService{
+		serviceHost: srv.URL,
+		clusterID:   "uuid",
+	}
+
+	failoverMode = `{
+		"data": {
+		  "cluster": {
+			"failover_params": {
+			  "mode": "eventual"
+			}
+		  }
+		}
+	}`
+
+	enabled, err := topology.GetFailover()
+	if err != nil {
+		t.Fatalf("%s", err)
+	}
+
+	if !enabled {
+		t.Fatal("Failover should be enabled")
+	}
+
+	failoverMode = `{
+		"data": {
+		  "cluster": {
+			"failover_params": {
+			  "mode": "stateful"
+			}
+		  }
+		}
+	}`
+
+	enabled, err = topology.GetFailover()
+	if err != nil {
+		t.Fatalf("%s", err)
+	}
+
+	if !enabled {
+		t.Fatal("Failover should be enabled")
+	}
+
+	failoverMode = `{
+		"data": {
+		  "cluster": {
+			"failover_params": {
+			  "mode": "disabled"
+			}
+		  }
+		}
+	}`
+
+	enabled, err = topology.GetFailover()
+	if err != nil {
+		t.Fatalf("%s", err)
+	}
+
+	if enabled {
+		t.Fatal("Failover should be disabled")
+	}
+
+	failoverMode = `{
+		"data": {
+		  "cluster": {
+		  }
+		}
+	}`
+
+	_, err = topology.GetFailover()
+	if err == nil {
+		t.Fatal("Wrong answer format, but error wasn't thrown")
+	}
+}