*: Support to manage the increase/deletion of the raft node via Operator. #221

Author: runkecheng

internal/xenon_executor.go

@@ -33,7 +33,10 @@ type XenonExecutor interface {
 	GetRootPassword() string
 	SetRootPassword(rootPassword string)
 	RaftStatus(host string) (*apiv1alpha1.RaftStatus, error)
+	XenonPing(host string) error
 	RaftTryToLeader(host string) error
+	ClusterAdd(host string, toAdd string) error
+	ClusterRemove(host string, toRemove string) error
 }
 
 func NewXenonExecutor() XenonExecutor {
@@ -86,3 +89,41 @@ func (executor *xenonExecutor) RaftTryToLeader(host string) error {
 	}
 	return nil
 }
+
+func (executor *xenonExecutor) XenonPing(host string) error {
+	req, err := NewXenonHttpRequest(NewRequestConfig(host, executor.GetRootPassword(), utils.XenonPing, nil))
+	if err != nil {
+		return err
+	}
+	_, err = executor.httpExecutor.Execute(req)
+	if err != nil {
+		return fmt.Errorf("failed to ping host[%s], err: %s", req.Req.URL, err)
+	}
+	return nil
+}
+
+func (executor *xenonExecutor) ClusterAdd(host string, toAdd string) error {
+	addHost := fmt.Sprintf("{\"address\": \"%s\"}", toAdd)
+	req, err := NewXenonHttpRequest(NewRequestConfig(host, executor.GetRootPassword(), utils.ClusterAdd, addHost))
+	if err != nil {
+		return err
+	}
+	_, err = executor.httpExecutor.Execute(req)
+	if err != nil {
+		return fmt.Errorf("failed to add host[%s] to host[%s], err: %s", addHost, req.Req.URL, err)
+	}
+	return nil
+}
+
+func (executor *xenonExecutor) ClusterRemove(host string, toRemove string) error {
+	removeHost := fmt.Sprintf("{\"address\": \"%s\"}", toRemove)
+	req, err := NewXenonHttpRequest(NewRequestConfig(host, executor.GetRootPassword(), utils.ClusterRemove, removeHost))
+	if err != nil {
+		return err
+	}
+	_, err = executor.httpExecutor.Execute(req)
+	if err != nil {
+		return fmt.Errorf("failed to remove host[%s] from host[%s], err: %s", removeHost, req.Req.URL, err)
+	}
+	return nil
+}

mysqlcluster/syncer/status.go

@@ -119,8 +119,13 @@ func (s *StatusSyncer) Sync(ctx context.Context) (syncer.SyncResult, error) {
 
 	s.Status.ReadyNodes = len(readyNodes)
 	if s.Status.ReadyNodes == int(*s.Spec.Replicas) && int(*s.Spec.Replicas) != 0 {
-		s.Status.State = apiv1alpha1.ClusterReadyState
-		clusterCondition.Type = apiv1alpha1.ConditionReady
+		if err := s.reconcileXenon(s.Status.ReadyNodes); err != nil {
+			clusterCondition.Message = fmt.Sprintf("%s", err)
+			clusterCondition.Type = apiv1alpha1.ConditionError
+		} else {
+			s.Status.State = apiv1alpha1.ClusterReadyState
+			clusterCondition.Type = apiv1alpha1.ConditionReady
+		}
 	}
 
 	if len(s.Status.Conditions) == 0 {
@@ -163,9 +168,13 @@ func (s *StatusSyncer) updateClusterStatus() apiv1alpha1.ClusterCondition {
 	// When the cluster is ready or closed, the number of replicas changes,
 	// indicating that the cluster is updating nodes.
 	if oldState == apiv1alpha1.ClusterReadyState || oldState == apiv1alpha1.ClusterCloseState {
-		if int(*s.Spec.Replicas) != s.Status.ReadyNodes {
-			clusterCondition.Type = apiv1alpha1.ConditionUpdate
-			s.Status.State = apiv1alpha1.ClusterUpdateState
+		if int(*s.Spec.Replicas) > s.Status.ReadyNodes {
+			clusterCondition.Type = apiv1alpha1.ConditionScaleOut
+			s.Status.State = apiv1alpha1.ClusterScaleOutState
+			return clusterCondition
+		} else if int(*s.Spec.Replicas) < s.Status.ReadyNodes {
+			clusterCondition.Type = apiv1alpha1.ConditionScaleIn
+			s.Status.State = apiv1alpha1.ClusterScaleInState
 			return clusterCondition
 		}
 	}
@@ -302,6 +311,53 @@ func (s *StatusSyncer) updateNodeRaftStatus(node *apiv1alpha1.NodeStatus) error
 	return nil
 }
 
+func (s *StatusSyncer) reconcileXenon(readyNodes int) error {
+	expectXenonNodes := s.getExpectXenonNodes(readyNodes)
+	for _, nodeStatus := range s.Status.Nodes {
+		toRemove := utils.StringDiffIn(nodeStatus.RaftStatus.Nodes, expectXenonNodes)
+		if err := s.removeNodesFromXenon(nodeStatus.Name, toRemove); err != nil {
+			return err
+		}
+		toAdd := utils.StringDiffIn(expectXenonNodes, nodeStatus.RaftStatus.Nodes)
+		if err := s.addNodesInXenon(nodeStatus.Name, toAdd); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *StatusSyncer) getExpectXenonNodes(readyNodes int) []string {
+	expectXenonNodes := []string{}
+	for i := 0; i < readyNodes; i++ {
+		expectXenonNodes = append(expectXenonNodes, fmt.Sprintf("%s:%d", s.GetPodHostName(i), utils.XenonPort))
+	}
+	return expectXenonNodes
+}
+
+func (s *StatusSyncer) removeNodesFromXenon(host string, toRemove []string) error {
+	if err := s.XenonExecutor.XenonPing(host); err != nil {
+		return err
+	}
+	for _, removeHost := range toRemove {
+		if err := s.XenonExecutor.ClusterRemove(host, removeHost); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *StatusSyncer) addNodesInXenon(host string, toAdd []string) error {
+	if err := s.XenonExecutor.XenonPing(host); err != nil {
+		return err
+	}
+	for _, addHost := range toAdd {
+		if err := s.XenonExecutor.ClusterAdd(host, addHost); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // setPodHealthy set the pod lable healthy.
 func (s *StatusSyncer) setPodHealthy(ctx context.Context, pod *corev1.Pod, node *apiv1alpha1.NodeStatus) error {
 	healthy := "no"

utils/common.go

@@ -110,7 +110,6 @@ func StringDiffIn(actual, desired []string) []string {
 			diff = append(diff, aStr)
 		}
 	}
-
 	return diff
 }
 
@@ -120,7 +119,6 @@ func stringIn(str string, strs []string) (int, bool) {
 			return i, true
 		}
 	}
-
 	return 0, false
 }
 
@@ -133,6 +131,5 @@ func UnmarshalJSON(in io.Reader, obj interface{}) error {
 	if err = json.Unmarshal(body, obj); err != nil {
 		return fmt.Errorf("error unmarshal data, error: %s, body: %s", err, string(body))
 	}
-
 	return nil
 }

utils/constants.go

@@ -41,6 +41,9 @@ var (
 	XenonHttpUrls = map[XenonHttpUrl]string{
 		RaftStatus:      http.MethodGet,
 		RaftTryToLeader: http.MethodPost,
+		XenonPing:       http.MethodGet,
+		ClusterAdd:      http.MethodPost,
+		ClusterRemove:   http.MethodPost,
 	}
 )
 
@@ -161,5 +164,8 @@ type XenonHttpUrl string
 
 const (
 	RaftStatus      XenonHttpUrl = "/v1/raft/status"
+	XenonPing       XenonHttpUrl = "/v1/xenon/ping"
+	ClusterAdd      XenonHttpUrl = "/v1/cluster/add"
+	ClusterRemove   XenonHttpUrl = "/v1/cluster/remove"
 	RaftTryToLeader XenonHttpUrl = "/v1/raft/trytoleader"
 )