Merge pull request #61 from banzaicloud/remove_least_partition

In case of Downgrade remove broker with the least partition

Author: matyix

charts/kafka-operator/values.yaml

@@ -7,7 +7,7 @@ replicaCount: 1
 operator:
   image:
     repository: banzaicloud/kafka-operator
-    tag: 0.3.0
+    tag: 0.3.1
     pullPolicy: IfNotPresent
   resources:
     limits:

internal/alertmanager/currentalert/process.go

@@ -17,6 +17,7 @@ package currentalert
 import (
 	banzaicloudv1alpha1 "github.com/banzaicloud/kafka-operator/pkg/apis/banzaicloud/v1alpha1"
 	"github.com/banzaicloud/kafka-operator/pkg/k8sutil"
+	"github.com/banzaicloud/kafka-operator/pkg/scale"
 	"github.com/banzaicloud/kafka-operator/pkg/util"
 	"github.com/prometheus/common/model"
 	corev1 "k8s.io/api/core/v1"
@@ -52,7 +53,11 @@ func addPVC(labels model.LabelSet, annotations model.LabelSet, client client.Cli
 }
 
 func downScale(labels model.LabelSet, client client.Client) error {
-	err := k8sutil.RemoveBrokerFromCr(string(labels["brokerId"]), string(labels["kafka_cr"]), string(labels["kubernetes_namespace"]), client)
+	brokerId, err := scale.GetBrokerIDWithLeastPartition(string(labels["kubernetes_namespace"]))
+	if err != nil {
+		return err
+	}
+	err = k8sutil.RemoveBrokerFromCr(brokerId, string(labels["kafka_cr"]), string(labels["kubernetes_namespace"]), client)
 	if err != nil {
 		return err
 	}

internal/alertmanager/receiver/http_handler.go

@@ -58,7 +58,7 @@ func (a *HTTPController) reciveAlert(w http.ResponseWriter, r *http.Request) {
 		}
 		err = alertReciever(a.Logger, alert, a.Client)
 		if err != nil {
-			http.Error(w, "alert reciever error", http.StatusBadRequest)
+			http.Error(w, "alert receiver error", http.StatusBadRequest)
 			return
 		}
 		w.WriteHeader(http.StatusAccepted)

pkg/scale/scale.go

@@ -161,6 +161,56 @@ func isKafkaBrokerReady(brokerId, namespace string) (bool, error) {
 	return running, nil
 }
 
+// GetBrokerIDWithLeastPartition returns
+func GetBrokerIDWithLeastPartition(namespace string) (string, error) {
+
+	brokerWithLeastPartition := ""
+
+	err := getCruiseControlStatus(namespace)
+	if err != nil {
+		return brokerWithLeastPartition, err
+	}
+
+	options := map[string]string{
+		"json": "true",
+	}
+
+	rsp, err := getCruiseControl(kafkaClusterStateAction, namespace, options)
+	if err != nil {
+		log.Error(err, "can't work with cruise-control because it is not ready")
+		return brokerWithLeastPartition, err
+	}
+
+	body, err := ioutil.ReadAll(rsp.Body)
+	if err != nil {
+		return brokerWithLeastPartition, err
+	}
+
+	err = rsp.Body.Close()
+	if err != nil {
+		return brokerWithLeastPartition, err
+	}
+
+	var response map[string]interface{}
+
+	err = json.Unmarshal(body, &response)
+	if err != nil {
+		return brokerWithLeastPartition, err
+	}
+
+	replicaCountByBroker := response["KafkaBrokerState"].(map[string]interface{})["ReplicaCountByBrokerId"].(map[string]interface{})
+	replicaCount := float64(99999)
+
+	for brokerID, replica := range replicaCountByBroker {
+		if replicaCount > replica.(float64) {
+			replicaCount = replica.(float64)
+			brokerWithLeastPartition = brokerID
+		}
+	}
+	return brokerWithLeastPartition, nil
+
+}
+
 // UpScaleCluster upscales Kafka cluster
 func UpScaleCluster(brokerId, namespace string) error {