Ability to trigger auto rebalance on imbalance feature

api/src/main/java/io/strimzi/api/kafka/model/kafka/cruisecontrol/KafkaAutoRebalanceMode.java

@@ -12,7 +12,8 @@
  */
 public enum KafkaAutoRebalanceMode {
     ADD_BROKERS("add-brokers"),
-    REMOVE_BROKERS("remove-brokers");
+    REMOVE_BROKERS("remove-brokers"),
+    IMBALANCE("imbalance");
 
     private final String name;
 
@@ -27,6 +28,8 @@ public static KafkaAutoRebalanceMode forValue(String value) {
                 return ADD_BROKERS;
             case "remove-brokers":
                 return REMOVE_BROKERS;
+            case "imbalance":
+                return IMBALANCE;
             default:
                 return null;
         }

api/src/main/java/io/strimzi/api/kafka/model/kafka/cruisecontrol/KafkaAutoRebalanceState.java

@@ -50,5 +50,26 @@ public enum KafkaAutoRebalanceState {
      *     <dt>Idle</dt><dd>if a scale up operation was requested, it was executed and completed successfully or failed.</dd>
      * </dl>
      */
-    RebalanceOnScaleUp
+    RebalanceOnScaleUp,
+
+    /**
+     * A rebalancing triggered by goal violations is running.
+     * Transitions to:
+     * <dl>
+     *     <dt>RebalanceOnImbalance</dt>
+     *     <dd>if a rebalancing on imbalance is still running.</dd>
+     *     <dt>RebalanceOnScaleDown</dt>
+     *     <dd>
+     *          if a scale down operation was requested while imbalance rebalancing is active, so the current rebalancing
+     *          is stopped and a new rebalancing scale down is started (scale operations take precedence).
+     *     </dd>
+     *     <dt>RebalanceOnScaleUp</dt>
+     *     <dd>
+     *          if a scale up operation was requested while imbalance rebalancing is active, so the current rebalancing
+     *          is stopped and a new rebalancing scale up is started (scale operations take precedence).
+     *     </dd>
+     *     <dt>Idle</dt><dd>if the rebalancing on imbalance was executed and completed successfully or failed.</dd>
+     * </dl>
+     */
+    RebalanceOnImbalance
 }

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/CruiseControl.java

@@ -62,14 +62,12 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletionStage;
 
 import static io.strimzi.operator.cluster.model.cruisecontrol.CruiseControlConfiguration.CRUISE_CONTROL_DEFAULT_ANOMALY_DETECTION_GOALS;
-import static io.strimzi.operator.cluster.model.cruisecontrol.CruiseControlConfiguration.CRUISE_CONTROL_GOALS;
 import static java.lang.String.format;
 
 /**
@@ -283,22 +281,9 @@ public void checkGoals(CruiseControlConfiguration configuration) {
             }
         }
 
-        // If no anomaly detection goals have been defined by the user, the defaults defined in Cruise Control will be used.
-        String anomalyGoalsString = configuration.getConfigOption(CruiseControlConfigurationParameters.ANOMALY_DETECTION_CONFIG_KEY.toString(), CRUISE_CONTROL_DEFAULT_ANOMALY_DETECTION_GOALS);
-        Set<String> anomalyDetectionGoals = new HashSet<>(Arrays.asList(anomalyGoalsString.split("\\s*,\\s*")));
-
-        String defaultGoalsString = configuration.getConfigOption(CruiseControlConfigurationParameters.DEFAULT_GOALS_CONFIG_KEY.toString(), CRUISE_CONTROL_GOALS);
-        Set<String> defaultGoals = new HashSet<>(Arrays.asList(defaultGoalsString.split("\\s*,\\s*")));
-
-        // Remove all the goals which are present in the default goals set from the anomaly detection goals
-        anomalyDetectionGoals.removeAll(defaultGoals);
-
-        if (!anomalyDetectionGoals.isEmpty()) {
-            // If the anomaly detection goals contain goals which are not in the default goals then the CC startup
-            // checks will fail, so we make the anomaly goals match the default goals
-            configuration.setConfigOption(CruiseControlConfigurationParameters.ANOMALY_DETECTION_CONFIG_KEY.toString(), defaultGoalsString);
-            LOGGER.warnCr(reconciliation, "Anomaly goals contained goals which are not in the configured default goals. Anomaly goals have " +
-                    "been changed to match the specified default goals.");
+        // If the user has not explicitly configured anomaly detection goals, default them to the default anomaly detection goals
+        if (configuration.getConfigOption(CruiseControlConfigurationParameters.ANOMALY_DETECTION_CONFIG_KEY.toString()) == null) {
+            configuration.setConfigOption(CruiseControlConfigurationParameters.ANOMALY_DETECTION_CONFIG_KEY.toString(), CRUISE_CONTROL_DEFAULT_ANOMALY_DETECTION_GOALS);
         }
     }
 

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaAssemblyOperator.java

@@ -281,7 +281,7 @@ class ReconciliationState {
         private Map<String, Storage> kafkaBrokerStorage;
         private Map<String, ResourceRequirements> kafkaBrokerResources;
         // needed to take information for the auto-rebalancing on scaling via Cruise Control
-        private Set<Integer> scalingDownBlockedNodes;
+        Set<Integer> scalingDownBlockedNodes;
         private KafkaClusterSecurityContext securityContext;
 
         /* test */ KafkaStatus kafkaStatus = new KafkaStatus();
@@ -660,7 +660,7 @@ EntityOperatorReconciler entityOperatorReconciler()   {
          * @return Kafka auto-rebalancing reconciler
          */
         KafkaAutoRebalancingReconciler kafkaAutoRebalancingReconciler() {
-            return new KafkaAutoRebalancingReconciler(reconciliation, kafkaAssembly, supplier, scalingDownBlockedNodes);
+            return new KafkaAutoRebalancingReconciler(reconciliation, kafkaAssembly, supplier, scalingDownBlockedNodes, metrics);
         }
 
         /**

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaAssemblyOperatorMetricsHolder.java

@@ -4,11 +4,13 @@
  */
 package io.strimzi.operator.cluster.operator.assembly;
 
+import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.Tags;
 import io.strimzi.api.kafka.model.nodepool.KafkaNodePool;
 import io.strimzi.operator.common.MetricsProvider;
 import io.strimzi.operator.common.config.ConfigParameter;
+import io.strimzi.operator.common.metrics.AnomalyMetricKey;
 import io.strimzi.operator.common.metrics.CertificateMetricKey;
 import io.strimzi.operator.common.metrics.MetricKey;
 import io.strimzi.operator.common.metrics.MetricsUtils;
@@ -32,9 +34,14 @@ public class KafkaAssemblyOperatorMetricsHolder extends OperatorMetricsHolder {
      * Metric name for certificate expiration timestamp in ms.
      */
     public static final String METRICS_CERTIFICATE_EXPIRATION_MS = METRICS_PREFIX + "certificate.expiration.timestamp.ms";
+    /**
+     * Metric name for anomalies detected counter.
+     */
+    public static final String METRICS_ANOMALIES_DETECTED = METRICS_PREFIX + "auto.rebalance.anomalies.detected.total";
 
     protected final Map<MetricKey, AtomicLong> certificateExpirationMap = new ConcurrentHashMap<>(1);
     protected final Map<MetricKey, AtomicInteger> nodePoolResourceCounterMap = new ConcurrentHashMap<>(1);
+    protected final Map<MetricKey, Counter> anomaliesDetectedCounterMap = new ConcurrentHashMap<>(1);
 
     /**
      * Constructs the operator metrics holder
@@ -128,4 +135,23 @@ public void resetNodePoolCounters(String namespace) {
             nodePoolResourceCounter(namespace).set(0);
         }
     }
+
+    /**
+     * Counter metric for anomalies detected by Cruise Control.
+     *
+     * @param namespace     Namespace of the resources being reconciled
+     * @param fixability    Fixability classification of the detected anomaly
+     *
+     * @return  Metrics counter
+     */
+    public Counter anomaliesDetectedCounter(String namespace, String fixability) {
+        return getCounter(
+                new AnomalyMetricKey(kind, namespace, fixability),
+                METRICS_ANOMALIES_DETECTED,
+                "Total number of anomalies detected by Cruise Control",
+                Optional.of(getLabelSelectorValues()),
+                anomaliesDetectedCounterMap,
+                Tag.of("type", "goal_violation"),
+                Tag.of("fixability", fixability));
+    }
 }