Cruise Control deployment (#2773)

* Added Cruise Control System Tests

    - Added deployment with CC to Kafka ST resources
    - Added pod name and deployment resources
    - Added CC deployment system test

Signed-off-by: Thomas Cooper <tcooper@redhat.com>

* Cruise Control Deployment

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Add broker capacity estimation and configuration

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Add k8s memory parsing to return different byte multiples

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Refactor Capacity constructor

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Regenerate helm charts

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Addressing some comments

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Addressing more comments

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Addressing more comments (refactoring capacity API, checkOwnerReferences)

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Addressing comments ( capacity properties as strings/validate units in schema)

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Tightening capacity regex; fixing docs

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Another doc fix

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Add missing goal; bump CC version to 2.0.100

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Fix logging and typos

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Addressing comments ( Update allowed disk capacity notation + refactoring)

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Fixing more typos

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Fixing another doc issue

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Securing communication between Cruise Control and Kafka (#37)

* Adding TLS communication between CC and Kafka

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Adding TLS communication between metric reporter and Kafka

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Fixed metrics report to use TLS hostname verification (#38)

Moved metrics topic to Cruise Control configuration
Hiding truststore and keystore passwords in the log

Signed-off-by: Paolo Patierno <ppatierno@live.com>

Co-authored-by: Paolo Patierno <ppatierno@live.com>
Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Removing CC replicas, refactor Dockerfile + CC dependencies

Signed-off-by: Kyle Liberti <kliberti@redhat.com>

* Added security context, network policy and tests (#39)

* Added security context and tests

Signed-off-by: Paolo Patierno <ppatierno@live.com>

* Added network policy for the cruise control REST API port

Signed-off-by: Paolo Patierno <ppatierno@live.com>

* Added CruiseControl ST to regression tests

Signed-off-by: Paolo Patierno <ppatierno@live.com>

* Fixed NPE on Cruise Control network policy

Signed-off-by: Paolo Patierno <ppatierno@live.com>

Co-authored-by: Thomas Cooper <tcooper@redhat.com>
Co-authored-by: Paolo Patierno <ppatierno@live.com>

api/src/main/java/io/strimzi/api/kafka/model/CruiseControlResources.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.api.kafka.model;
+
+/**
+ * Encapsulates the naming scheme used for the resources which the Cluster Operator manages for a
+ * {@code CruiseControl} cluster.
+ */
+public class CruiseControlResources {
+
+    /**
+     * Returns the name of the Cruise Control {@code Deployment} for a {@code Kafka} cluster of the given name.
+     * @param clusterName  The {@code metadata.name} of the {@code Kafka} resource.
+     * @return The name of the corresponding Cruise Control {@code Deployment}.
+     */
+    public static String deploymentName(String clusterName) {
+        return clusterName + "-cruise-control";
+    }
+
+    /**
+     * Returns the name of the Cruise Control {@code ServiceAccount} for a {@code Kafka} cluster of the given name.
+     * @param clusterName  The {@code metadata.name} of the {@code Kafka} resource.
+     * @return The name of the corresponding Cruise Control {@code ServiceAccount}.
+     */
+    public static String serviceAccountName(String clusterName) {
+        return deploymentName(clusterName);
+    }
+
+    /**
+     * Returns the name of the Cruise Control {@code Service} for a {@code Kafka} cluster of the given name.
+     * @param clusterName  The {@code metadata.name} of the {@code Kafka} resource.
+     * @return The name of the corresponding Cruise Control {@code Service}.
+     */
+    public static String serviceName(String clusterName) {
+        return deploymentName(clusterName);
+    }
+
+    /**
+     * Returns the name of the Cruise Control {@code Secret} for a {@code Kafka} cluster of the given name.
+     * This {@code Secret} will only exist if {@code Kafka.spec.cruiseControl} is configured in the
+     * {@code Kafka} resource with the given name.
+     *
+     * @param clusterName  The {@code metadata.name} of the {@code Kafka} resource.
+     * @return The name of the corresponding Cruise Control {@code Secret}.
+     */
+    public static String secretName(String clusterName) {
+        return deploymentName(clusterName) + "-certs";
+    }
+
+    /**
+     * Returns the name of the Cruise Control log {@code ConfigMap} for a {@code Kafka} cluster of the given name.
+     * @param clusterName  The {@code metadata.name} of the {@code Kafka} resource.
+     * @return The name of the corresponding Cruise Control log {@code ConfigMap}.
+     */
+    public static String logAndMetricsConfigMapName(String clusterName) {
+        return clusterName + "-cruise-control-config";
+    }
+}

api/src/main/java/io/strimzi/api/kafka/model/CruiseControlSpec.java

@@ -0,0 +1,164 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.api.kafka.model;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import io.fabric8.kubernetes.api.model.ResourceRequirements;
+import io.strimzi.api.kafka.model.balancing.BrokerCapacity;
+import io.strimzi.api.kafka.model.template.CruiseControlTemplate;
+import io.strimzi.crdgenerator.annotations.Description;
+import io.sundr.builder.annotations.Buildable;
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import lombok.EqualsAndHashCode;
+
+@Buildable(
+        editableEnabled = false,
+        generateBuilderPackage = false,
+        builderPackage = "io.fabric8.kubernetes.api.builder"
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonPropertyOrder({
+        "image", "config",
+        "livenessProbe", "readinessProbe",
+        "jvmOptions", "resources",
+        "logging", "tlsSidecar", "template", "brokerCapacity"})
+@EqualsAndHashCode
+public class CruiseControlSpec implements UnknownPropertyPreserving, Serializable {
+    private static final long serialVersionUID = 1L;
+
+    // For the full configuration list refer to https://github.com/linkedin/cruise-control/wiki/Configurations
+    public static final String FORBIDDEN_PREFIXES = "bootstrap.servers, client.id, zookeeper., network., security., failed.brokers.zk.path,"
+        + "webserver.http., webserver.api.urlprefix, webserver.session.path, webserver.accesslog., two.step., request.reason.required,"
+        + "metric.reporter.sampler.bootstrap.servers, metric.reporter.topic, partition.metric.sample.store.topic, broker.metric.sample.store.topic,"
+        + "capacity.config.file, self.healing., ssl.";
+    public static final String FORBIDDEN_PREFIX_EXCEPTIONS = "ssl.cipher.suites, ssl.protocol, ssl.enabled.protocols";
+
+    private String image;
+    private TlsSidecar tlsSidecar;
+    private ResourceRequirements resources;
+    private Probe livenessProbe;
+    private Probe readinessProbe;
+    private JvmOptions jvmOptions;
+    private Logging logging;
+    private CruiseControlTemplate template;
+    private BrokerCapacity brokerCapacity;
+    private Map<String, Object> config = new HashMap<>(0);
+    private Map<String, Object> additionalProperties = new HashMap<>(0);
+
+    @Description("The docker image for the pods.")
+    @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+    public String getImage() {
+        return image;
+    }
+
+    public void setImage(String image) {
+        this.image = image;
+    }
+
+    @Description("TLS sidecar configuration")
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public TlsSidecar getTlsSidecar() {
+        return tlsSidecar;
+    }
+
+    public void setTlsSidecar(TlsSidecar tlsSidecar) {
+        this.tlsSidecar = tlsSidecar;
+    }
+
+    @Description("The Cruise Control `brokerCapacity` configuration.")
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public BrokerCapacity getBrokerCapacity() {
+        return brokerCapacity;
+    }
+
+    public void setBrokerCapacity(BrokerCapacity brokerCapacity) {
+        this.brokerCapacity = brokerCapacity;
+    }
+
+    @Description("The Cruise Control configuration. For a full list of configuration options refer to" +
+            " https://github.com/linkedin/cruise-control/wiki/Configurations. Note that properties " +
+            "with the following prefixes cannot be set: " + FORBIDDEN_PREFIXES)
+    @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    public Map<String, Object> getConfig() {
+        return config;
+    }
+
+    public void setConfig(Map<String, Object> config) {
+        this.config = config;
+    }
+
+    @Description("Logging configuration (log4j1) for Cruise Control.")
+    @JsonInclude(value = JsonInclude.Include.NON_NULL)
+    public Logging getLogging() {
+        return logging;
+    }
+
+    public void setLogging(Logging logging) {
+        this.logging = logging;
+    }
+
+    @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    @Description("JVM Options for the Cruise Control container")
+    public JvmOptions getJvmOptions() {
+        return jvmOptions;
+    }
+
+    public void setJvmOptions(JvmOptions jvmOptions) {
+        this.jvmOptions = jvmOptions;
+    }
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Description("CPU and memory resources to reserve for the Cruise Control container")
+    public ResourceRequirements getResources() {
+        return resources;
+    }
+
+    public void setResources(ResourceRequirements resources) {
+        this.resources = resources;
+    }
+
+    @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    @Description("Pod liveness checking for the Cruise Control container")
+    public Probe getLivenessProbe() {
+        return livenessProbe;
+    }
+
+    public void setLivenessProbe(Probe livenessProbe) {
+        this.livenessProbe = livenessProbe;
+    }
+
+    @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+    @Description("Pod readiness checking for the Cruise Control container.")
+    public Probe getReadinessProbe() {
+        return readinessProbe;
+    }
+
+    public void setReadinessProbe(Probe readinessProbe) {
+        this.readinessProbe = readinessProbe;
+    }
+
+    @Description("Template to specify how Cruise Control resources, `Deployments` and `Pods`, are generated.")
+    @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    public CruiseControlTemplate getTemplate() {
+        return template;
+    }
+
+    public void setTemplate(CruiseControlTemplate template) {
+        this.template = template;
+    }
+
+    @Override
+    public Map<String, Object> getAdditionalProperties() {
+        return this.additionalProperties;
+    }
+
+    @Override
+    public void setAdditionalProperty(String name, Object value) {
+        this.additionalProperties.put(name, value);
+    }
+}

api/src/main/java/io/strimzi/api/kafka/model/KafkaClusterSpec.java

@@ -48,7 +48,9 @@ public class KafkaClusterSpec implements UnknownPropertyPreserving, Serializable
 
     public static final String FORBIDDEN_PREFIXES = "listeners, advertised., broker., listener., host.name, port, "
             + "inter.broker.listener.name, sasl., ssl., security., password., principal.builder.class, log.dir, "
-            + "zookeeper.connect, zookeeper.set.acl, authorizer., super.user";
+            + "zookeeper.connect, zookeeper.set.acl, authorizer., super.user"
+            + "cruise.control.metrics.topic, cruise.control.metrics.reporter.bootstrap.servers";
+
     public static final String FORBIDDEN_PREFIX_EXCEPTIONS = "zookeeper.connection.timeout.ms, ssl.cipher.suites, ssl.protocol, ssl.enabled.protocols";
 
     protected Storage storage;

api/src/main/java/io/strimzi/api/kafka/model/KafkaSpec.java

@@ -42,6 +42,7 @@ public class KafkaSpec implements UnknownPropertyPreserving, Serializable {
     private CertificateAuthority clusterCa;
     private JmxTransSpec jmxTrans;
     private KafkaExporterSpec kafkaExporter;
+    private CruiseControlSpec cruiseControl;
 
     private CertificateAuthority clientsCa;
     private List<String> maintenanceTimeWindows;
@@ -136,6 +137,15 @@ public void setKafkaExporter(KafkaExporterSpec kafkaExporter) {
         this.kafkaExporter = kafkaExporter;
     }
 
+    @Description("Configuration for Cruise Control deployment. Deploys a Cruise Control instance when specified")
+    public CruiseControlSpec getCruiseControl() {
+        return cruiseControl;
+    }
+
+    public void setCruiseControl(CruiseControlSpec cruiseControl) {
+        this.cruiseControl = cruiseControl;
+    }
+
     @Override
     public Map<String, Object> getAdditionalProperties() {
         return this.additionalProperties;

api/src/main/java/io/strimzi/api/kafka/model/balancing/BrokerCapacity.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.api.kafka.model.balancing;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import io.strimzi.api.kafka.model.UnknownPropertyPreserving;
+import io.strimzi.crdgenerator.annotations.Description;
+import io.strimzi.crdgenerator.annotations.Maximum;
+import io.strimzi.crdgenerator.annotations.Minimum;
+import io.strimzi.crdgenerator.annotations.Pattern;
+import io.sundr.builder.annotations.Buildable;
+import lombok.EqualsAndHashCode;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Representation of the Cruise Control broker capacity settings. Since the Kafka brokers
+ * in Strimzi are homogeneous, the capacity values for each resource will be
+ * used for every broker.
+ */
+@Buildable(
+        editableEnabled = false,
+        generateBuilderPackage = false,
+        builderPackage = "io.fabric8.kubernetes.api.builder"
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonPropertyOrder({"disk", "cpuUtilization", "inboundNetwork", "outboundNetwork"})
+@EqualsAndHashCode
+public class BrokerCapacity implements UnknownPropertyPreserving, Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private String disk;
+    private Integer cpuUtilization;
+    private String inboundNetwork;
+    private String outboundNetwork;
+    private Map<String, Object> additionalProperties = new HashMap<>(0);
+
+    @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+    @Pattern("^[0-9]+([.][0-9]*)?([KMGTPE]i?|e[0-9]+)?$")
+    @Description("Broker capacity for disk in bytes, for example, 100Gi.")
+    public String getDisk() {
+        return disk;
+    }
+
+    public void setDisk(String disk) {
+        this.disk = disk;
+    }
+
+    @Minimum(0)
+    @Maximum(100)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Description("Broker capacity for CPU resource utilization as a percentage (0 - 100).")
+    public Integer getCpuUtilization() {
+        return cpuUtilization;
+    }
+
+    public void setCpuUtilization(Integer cpuUtilization) {
+        this.cpuUtilization = cpuUtilization;
+    }
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Pattern("[0-9]+([KMG]i?)?B/s")
+    @Description("Broker capacity for inbound network throughput in bytes per second, for example, 10000KB/s")
+    public String getInboundNetwork() {
+        return inboundNetwork;
+    }
+
+    public void setInboundNetwork(String inboundNetwork) {
+        this.inboundNetwork = inboundNetwork;
+    }
+
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Pattern("[0-9]+([KMG]i?)?B/s")
+    @Description("Broker capacity for outbound network throughput in bytes per second, for example 10000KB/s")
+    public String getOutboundNetwork() {
+        return outboundNetwork;
+    }
+
+    public void setOutboundNetwork(String outboundNetwork) {
+        this.outboundNetwork = outboundNetwork;
+    }
+
+    @Override
+    public Map<String, Object> getAdditionalProperties() {
+        return this.additionalProperties;
+    }
+
+    @Override
+    public void setAdditionalProperty(String name, Object value) {
+        this.additionalProperties.put(name, value);
+    }
+}