feat: add rebalance controller of v2alpha2

emqx · Jul 20, 2023 · f6b4efa · f6b4efa
1 parent 832d7e6
commit f6b4efa
Show file tree

Hide file tree

Showing 12 changed files with 1,241 additions and 8 deletions.
diff --git a/apis/apps/v1beta4/emqxbroker_types.go b/apis/apps/v1beta4/emqxbroker_types.go
@@ -135,6 +135,20 @@ func (s *EmqxBrokerStatus) GetConditions() []Condition {
 	return s.Conditions
 }
 
+func (s *EmqxBrokerStatus) GetCondition(t ConditionType) (int, *Condition) {
+	for i, c := range s.GetConditions() {
+		if t == c.Type {
+			return i, &c
+		}
+	}
+	return -1, nil
+}
+
+func (s *EmqxBrokerStatus) IsConditionTrue(t ConditionType) bool {
+	_, c := s.GetCondition(t)
+	return c != nil && c.Status == corev1.ConditionTrue
+}
+
 func (s *EmqxBrokerStatus) AddCondition(condType ConditionType, status corev1.ConditionStatus, reason, message string) {
 	s.Conditions = addCondition(s.Conditions, Condition{
 		Type:    condType,

diff --git a/apis/apps/v1beta4/emqxenterprise_types.go b/apis/apps/v1beta4/emqxenterprise_types.go
@@ -173,6 +173,20 @@ func (s *EmqxEnterpriseStatus) GetConditions() []Condition {
 	return s.Conditions
 }
 
+func (s *EmqxEnterpriseStatus) GetCondition(t ConditionType) (int, *Condition) {
+	for i, c := range s.GetConditions() {
+		if t == c.Type {
+			return i, &c
+		}
+	}
+	return -1, nil
+}
+
+func (s *EmqxEnterpriseStatus) IsConditionTrue(t ConditionType) bool {
+	_, c := s.GetCondition(t)
+	return c != nil && c.Status == corev1.ConditionTrue
+}
+
 func (s *EmqxEnterpriseStatus) AddCondition(condType ConditionType, status corev1.ConditionStatus, reason, message string) {
 	s.Conditions = addCondition(s.Conditions, Condition{
 		Type:    condType,

diff --git a/apis/apps/v1beta4/status_types.go b/apis/apps/v1beta4/status_types.go
@@ -49,6 +49,8 @@ type EmqxStatus interface {
 	GetCurrentStatefulSetVersion() string
 	SetCurrentStatefulSetVersion(version string)
 	GetConditions() []Condition
+	GetCondition(t ConditionType) (int, *Condition)
+	IsConditionTrue(t ConditionType) bool
 	AddCondition(condType ConditionType, status corev1.ConditionStatus, reason, message string)
 }
 

diff --git a/controllers/apps/v1beta4/emqx_controller.go b/controllers/apps/v1beta4/emqx_controller.go
@@ -133,6 +133,19 @@ func (r *EmqxReconciler) processResult(subResult subResult, instance appsv1beta4
 	return subResult.result, subResult.err
 }
 
+func NewRequesterByPod(client client.Client, instance appsv1beta4.Emqx, pod *corev1.Pod) (innerReq.RequesterInterface, error) {
+	username, password, err := getBootstrapUser(context.Background(), client, instance)
+	if err != nil {
+		return nil, err
+	}
+
+	return &innerReq.Requester{
+		Host:     fmt.Sprintf("%s:8081", pod.Status.PodIP),
+		Username: username,
+		Password: password,
+	}, nil
+}
+
 func newRequesterBySvc(client client.Client, instance appsv1beta4.Emqx) (innerReq.RequesterInterface, error) {
 	username, password, err := getBootstrapUser(context.Background(), client, instance)
 	if err != nil {