cluster reconcile: set incoming replicasets weight to 1 (#8)

Author: vasiliy-t

pkg/controller/cluster/cluster_controller.go

@@ -321,6 +321,50 @@ func (r *ReconcileCluster) Reconcile(request reconcile.Request) (reconcile.Resul
 		}
 	}
 
+	for _, sts := range stsList.Items {
+		stsAnnotations := sts.GetAnnotations()
+		weight, ok := stsAnnotations["tarantool.io/replicaset-weight"]
+		if ok {
+			continue
+		}
+
+		if weight == "1" {
+			continue
+		}
+
+		for i := 0; i < int(*sts.Spec.Replicas); i++ {
+			pod := &corev1.Pod{}
+			name := types.NamespacedName{
+				Namespace: request.Namespace,
+				Name:      fmt.Sprintf("%s-%d", sts.GetName(), i),
+			}
+			if err := r.client.Get(context.TODO(), name, pod); err != nil {
+				if errors.IsNotFound(err) {
+					return reconcile.Result{RequeueAfter: time.Duration(5 * time.Second)}, nil
+				}
+
+				return reconcile.Result{RequeueAfter: time.Duration(5 * time.Second)}, err
+			}
+			if tarantool.IsJoined(pod) == false {
+				reqLogger.Info("Not all instances joined, skip weight change", "StatefulSet.Name", sts.GetName())
+				return reconcile.Result{RequeueAfter: time.Duration(5 * time.Second)}, nil
+			}
+		}
+
+		if err := topologyClient.SetWeight(sts.GetLabels()["tarantool.io/replicaset-uuid"]); err != nil {
+			return reconcile.Result{RequeueAfter: time.Duration(5 * time.Second)}, err
+		}
+		if stsAnnotations == nil {
+			stsAnnotations = make(map[string]string)
+		}
+
+		stsAnnotations["tarantool.io/replicaset-weight"] = "1"
+		sts.SetAnnotations(stsAnnotations)
+		if err := r.client.Update(context.TODO(), &sts); err != nil {
+			return reconcile.Result{RequeueAfter: time.Duration(5 * time.Second)}, err
+		}
+	}
+
 	if err := topologyClient.BootstrapVshard(); err != nil {
 		if topology.IsAlreadyBootstrapped(err) {
 			cluster.Status.State = "Ready"

pkg/controller/role/role_controller.go

@@ -155,6 +155,7 @@ func (r *ReconcileRole) Reconcile(request reconcile.Request) (reconcile.Result,
 		return reconcile.Result{}, err
 	}
 
+	// ensure num of statefulsets matches user expectations
 	if len(stsList.Items) > int(*role.Spec.Replicas) {
 		reqLogger.Info("Role", "more instances", *role.Spec.Replicas)
 		for i := len(stsList.Items); i > int(*role.Spec.Replicas); i-- {

pkg/topology/builtin.go

@@ -17,7 +17,7 @@ type ResponseError struct {
 	Message string `json:"message"`
 }
 type JoinResponseData struct {
-	JoinInstance bool `json:"join_instance"`
+	JoinInstance bool `json:"joinInstanceResponse"`
 }
 type JoinResponse struct {
 	Errors []*ResponseError  `json:"errors,omitempty"`
@@ -44,14 +44,21 @@ type BuiltInTopologyService struct {
 	serviceHost string
 }
 
+type EditReplicasetResponse struct {
+	Response bool `json:"editReplicasetResponse"`
+}
+
 var (
 	topologyIsDown      = errors.New("topology service is down")
 	alreadyJoined       = errors.New("already joined")
 	alreadyBootstrapped = errors.New("already bootstrapped")
 )
 
 var join_mutation = `mutation do_join_server($uri: String!, $instance_uuid: String!, $replicaset_uuid: String!, $roles: [String!]) {
-	join_instance: join_server(uri: $uri, instance_uuid: $instance_uuid, replicaset_uuid: $replicaset_uuid, roles: $roles, timeout: 10)
+	joinInstanceResponse: join_server(uri: $uri, instance_uuid: $instance_uuid, replicaset_uuid: $replicaset_uuid, roles: $roles, timeout: 10)
+}`
+var edit_rs_mutation = `mutation editReplicaset($uuid: String!, $weight: Float) {
+	editReplicasetResponse: edit_replicaset(uuid: $uuid, weight: $weight)
 }`
 
 func (s *BuiltInTopologyService) Join(pod *corev1.Pod) error {
@@ -113,12 +120,30 @@ func (s *BuiltInTopologyService) Expel(pod *corev1.Pod) error {
 	}
 
 	if resp.Data.ExpelInstance == false && (resp.Errors == nil || len(resp.Errors) == 0) {
-		return errors.New("Shit happened")
+		return errors.New("something really bad happened")
 	}
 
 	return nil
 }
 
+func (s *BuiltInTopologyService) SetWeight(replicasetUUID string) error {
+	client := graphql.NewClient(s.serviceHost, graphql.WithHTTPClient(&http.Client{Timeout: time.Duration(time.Second * 5)}))
+	req := graphql.NewRequest(edit_rs_mutation)
+
+	req.Var("uuid", replicasetUUID)
+	req.Var("weight", 1)
+
+	resp := &EditReplicasetResponse{}
+	if err := client.Run(context.TODO(), req, resp); err != nil {
+		return err
+	}
+	if resp.Response == true {
+		return nil
+	}
+
+	return errors.New("something really bad happened")
+}
+
 func (s *BuiltInTopologyService) BootstrapVshard() error {
 	req := fmt.Sprint("mutation bootstrap {bootstrapVshardResponse: bootstrap_vshard}")
 	j := fmt.Sprintf("{\"query\": \"%s\"}", req)