Repository: banzaicloud/kafka-operator
Pull request #258 (2020-01-22 16:28:44)

Fix: upscale/downscale action executed only even though continuously receiving triggering alert


| Q               | A
| --------------- | ---
| Bug fix?        | yes
| New feature?    | no
| API breaks?     | no
| Deprecations?   | no
| License         | Apache 2.0


### What's in this PR?
<!-- Explain the contents of the PR. Give an overview about the implementation, which decisions were made and why. -->
Do not add alert identifier to the processed alerts cache in order to process the alert multiple times if keeps fired. 

### Why?
<!-- Which problem does the PR fix? (Please remove this section if you linked an issue above) -->
In case the root cause triggering an alert is not rectified by kafka-operator reacting to it the first time the operator did not react to upcoming occurrences of the same alert. The operator should react as long as the alert is present.

### Checklist
<!-- Place an '[x]' (no spaces) in all applicable fields. Please remove unrelated fields. -->

- [x] Implementation tested
- [x] Error handling code meets the [guideline](https://github.com/banzaicloud/pipeline/blob/master/docs/error-handling-guide.md)
- [x] Logging code meets the guideline




Patch:
diff --git a/internal/alertmanager/currentalert/process.go b/internal/alertmanager/currentalert/process.go
index a0959b70..0515faeb 100644
--- a/internal/alertmanager/currentalert/process.go
+++ b/internal/alertmanager/currentalert/process.go
@@ -15,10 +15,14 @@
 package currentalert
 
 import (
+	"strconv"
+
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/k8sutil"
+	"github.com/banzaicloud/kafka-operator/pkg/resources/kafka"
 	"github.com/banzaicloud/kafka-operator/pkg/scale"
 	"github.com/banzaicloud/kafka-operator/pkg/util"
+	"github.com/go-logr/logr"
 	"github.com/prometheus/common/model"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -93,22 +97,26 @@ func (e *examiner) processAlert(ds disableScaling) (bool, error) {
 		}
 	case "downScale":
 		if ds.Down {
-			e.Log.Info("downscaling is skipped due to downscale limit")
+			e.Log.Info("downscale is skipped due to downscale limit")
 			return true, nil
 		}
-		err := downScale(e.Alert.Labels, e.Client)
+		err := downScale(e.Log, e.Alert.Labels, e.Client)
 		if err != nil {
 			return false, err
 		}
+
+		return true, nil
 	case "upScale":
 		if ds.Up {
-			e.Log.Info("upscaling is skipped due to upscale limit")
+			e.Log.Info("upscale is skipped due to upscale limit")
 			return true, nil
 		}
-		err := upScale(e.Alert.Labels, e.Alert.Annotations, e.Client)
+		err := upScale(e.Log, e.Alert.Labels, e.Alert.Annotations, e.Client)
 		if err != nil {
 			return false, err
 		}
+
+		return true, nil
 	}
 	return false, nil
 }
@@ -118,13 +126,25 @@ func addPVC(labels model.LabelSet, annotations model.LabelSet, client client.Cli
 	return nil
 }
 
-func downScale(labels model.LabelSet, client client.Client) error {
+func downScale(log logr.Logger, labels model.LabelSet, client client.Client) error {
 
 	cr, err := k8sutil.GetCr(string(labels["kafka_cr"]), string(labels["namespace"]), client)
 	if err != nil {
 		return err
 	}
 
+	if ids := kafka.GetBrokersWithPendingOrRunningCCTask(cr); len(ids) > 0 {
+		var keyVals []interface{}
+		for _, id := range ids {
+			brokerId := strconv.Itoa(int(id))
+			keyVals = append(keyVals, brokerId, cr.Status.BrokersState[brokerId].GracefulActionState.CruiseControlState)
+		}
+
+		log.Info("downscale is skipped as there are brokers which have tasks pending or running in CC", keyVals)
+
+		return nil
+	}
+
 	brokerId, err := scale.GetBrokerIDWithLeastPartition(string(labels["namespace"]), cr.Spec.CruiseControlConfig.CruiseControlEndpoint, cr.Name)
 	if err != nil {
 		return err
@@ -136,13 +156,25 @@ func downScale(labels model.LabelSet, client client.Client) error {
 	return nil
 }
 
-func upScale(labels model.LabelSet, annotations model.LabelSet, client client.Client) error {
+func upScale(log logr.Logger, labels model.LabelSet, annotations model.LabelSet, client client.Client) error {
 
 	cr, err := k8sutil.GetCr(string(labels["kafka_cr"]), string(labels["namespace"]), client)
 	if err != nil {
 		return err
 	}
 
+	if ids := kafka.GetBrokersWithPendingOrRunningCCTask(cr); len(ids) > 0 {
+		var keyVals []interface{}
+		for _, id := range ids {
+			brokerId := strconv.Itoa(int(id))
+			keyVals = append(keyVals, brokerId, cr.Status.BrokersState[brokerId].GracefulActionState.CruiseControlState)
+		}
+
+		log.Info("upscale is skipped as there are brokers which have tasks pending or running in CC", keyVals)
+
+		return nil
+	}
+
 	biggestId := int32(0)
 	for _, broker := range cr.Spec.Brokers {
 		if broker.Id > biggestId {
@@ -160,6 +192,10 @@ func upScale(labels model.LabelSet, annotations model.LabelSet, client client.Cl
 		broker.Id = biggestId + 1
 
 	} else {
+		var storageClassName *string
+		if annotations["storageClass"] != "" {
+			storageClassName = util.StringPointer(string(annotations["storageClass"]))
+		}
 
 		broker = v1beta1.Broker{
 			Id: biggestId + 1,
@@ -172,7 +208,7 @@ func upScale(labels model.LabelSet, annotations model.LabelSet, client client.Cl
 							AccessModes: []corev1.PersistentVolumeAccessMode{
 								corev1.ReadWriteOnce,
 							},
-							StorageClassName: util.StringPointer(string(annotations["storageClass"])),
+							StorageClassName: storageClassName,
 							Resources: corev1.ResourceRequirements{
 								Requests: corev1.ResourceList{
 									"storage": resource.MustParse(string(annotations["diskSize"])),
diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go
index a88c69d3..40aa3945 100644
--- a/pkg/resources/kafka/kafka.go
+++ b/pkg/resources/kafka/kafka.go
@@ -767,3 +767,17 @@ func (r *Reconciler) reconcileKafkaPVC(log logr.Logger, desiredPVC *corev1.Persi
 	}
 	return nil
 }
+
+// GetBrokersWithPendingOrRunningCCTask returns list of brokers that are either waiting for CC
+// to start executing a broker task (add broker, remove broker, etc) or CC already running a task for it.
+func GetBrokersWithPendingOrRunningCCTask(kafkaCluster *v1beta1.KafkaCluster) []int32 {
+	var brokerIDs []int32
+	for i := range kafkaCluster.Spec.Brokers {
+		if state, ok := kafkaCluster.Status.BrokersState[strconv.Itoa(int(kafkaCluster.Spec.Brokers[i].Id))]; ok {
+			if state.GracefulActionState.CruiseControlState == v1beta1.GracefulUpdateRequired || (state.GracefulActionState.CruiseControlTaskId != "" && state.GracefulActionState.CruiseControlState == v1beta1.GracefulUpdateRunning) {
+				brokerIDs = append(brokerIDs, kafkaCluster.Spec.Brokers[i].Id)
+			}
+		}
+	}
+	return brokerIDs
+}


Test patch:
diff --git a/pkg/resources/kafka/kafka_test.go b/pkg/resources/kafka/kafka_test.go
new file mode 100644
index 00000000..ba56015e
--- /dev/null
+++ b/pkg/resources/kafka/kafka_test.go
@@ -0,0 +1,123 @@
+// Copyright © 2020 Banzai Cloud
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kafka
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
+)
+
+func TestGetBrokersWithPendingOrRunningCCTask(t *testing.T) {
+	testCases := []struct {
+		testName     string
+		kafkaCluster v1beta1.KafkaCluster
+		expectedIDs  []int32
+	}{
+		{
+			testName: "no pending or running CC tasks",
+			kafkaCluster: v1beta1.KafkaCluster{
+				Spec: v1beta1.KafkaClusterSpec{
+					Brokers: []v1beta1.Broker{
+						{
+							Id: 0,
+						},
+						{
+							Id: 1,
+						},
+						{
+							Id: 2,
+						},
+					},
+				},
+				Status: v1beta1.KafkaClusterStatus{
+					BrokersState: map[string]v1beta1.BrokerState{
+						"0": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateSucceeded},
+						},
+						"1": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateFailed},
+						},
+						"2": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateNotRequired},
+						},
+						"3": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateRequired},
+						},
+					},
+				},
+			},
+		},
+		{
+			testName: "pending and running CC tasks",
+			kafkaCluster: v1beta1.KafkaCluster{
+				Spec: v1beta1.KafkaClusterSpec{
+					Brokers: []v1beta1.Broker{
+						{
+							Id: 0,
+						},
+						{
+							Id: 1,
+						},
+						{
+							Id: 2,
+						},
+						{
+							Id: 4,
+						},
+					},
+				},
+				Status: v1beta1.KafkaClusterStatus{
+					BrokersState: map[string]v1beta1.BrokerState{
+						"0": {
+							GracefulActionState: v1beta1.GracefulActionState{
+								CruiseControlTaskId: "cc-task-id-1",
+								CruiseControlState:  v1beta1.GracefulUpdateRunning,
+							},
+						},
+						"1": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateFailed},
+						},
+						"2": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateRequired},
+						},
+						"3": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateRequired},
+						},
+						"4": {
+							GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulUpdateRunning},
+						},
+					},
+				},
+			},
+			expectedIDs: []int32{0, 2},
+		},
+	}
+
+	t.Parallel()
+
+	for _, test := range testCases {
+		test := test
+
+		t.Run(test.testName, func(t *testing.T) {
+			actual := GetBrokersWithPendingOrRunningCCTask(&test.kafkaCluster)
+
+			if !reflect.DeepEqual(actual, test.expectedIDs) {
+				t.Error("Expected:", test.expectedIDs, ", got:", actual)
+			}
+		})
+	}
+}