Configures a Cruise Control cluster.

Configuration options relate to:

* Goals configuration
* Capacity limits for resource distribution goals

The `config` properties are one part of the overall configuration for the resource.
Use the `config` properties to configure Cruise Control options as keys.

.Example Cruise Control configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  # ...
  cruiseControl:
    # ...
    config:
      # Note that `default.goals` (superset) must also include all `hard.goals` (subset)
      default.goals: >
        com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,
        com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal
      hard.goals: >
        com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal
      cpu.balance.threshold: 1.1
      metadata.max.age.ms: 300000
      send.buffer.bytes: 131072
      webserver.http.cors.enabled: true
      webserver.http.cors.origin: "*"
      webserver.http.cors.exposeheaders: "User-Task-ID,Content-Type"
    # ...
----

The values can be one of the following JSON types:

* String
* Number
* Boolean

*Exceptions*

You can specify and configure the options listed in the {CruiseControlConfigDocs}.

However, Strimzi takes care of configuring and managing options related to the following, which cannot be changed:

* Security (encryption, authentication, and authorization)
* Connection to the Kafka cluster
* Client ID configuration
* ZooKeeper connectivity
* Web server configuration
* Self healing

Properties with the following prefixes cannot be set:

* `bootstrap.servers`
* `capacity.config.file`
* `client.id`
* `failed.brokers.zk.path`
* `kafka.broker.failure.detection.enable`
* `metric.reporter.sampler.bootstrap.servers`
* `network.`
* `request.reason.required`
* `security.`
* `self.healing.`
* `ssl.`
* `topic.config.provider.class`
* `two.step.`
* `webserver.accesslog.`
* `webserver.api.urlprefix`
* `webserver.http.`
* `webserver.session.path`
* `zookeeper.`

If the `config` property contains an option that cannot be changed, it is disregarded, and a warning message is logged to the Cluster Operator log file.
All other supported options are forwarded to Cruise Control, including the following exceptions to the options configured by Strimzi:

* Any `ssl` configuration for xref:con-common-configuration-ssl-reference[supported TLS versions and cipher suites]
* Configuration for `webserver` properties to enable Cross-Origin Resource Sharing (CORS)

[id='property-cruise-control-config-cors-{context}']
= Cross-Origin Resource Sharing (CORS)

Cross-Origin Resource Sharing (CORS) is a HTTP mechanism for controlling access to REST APIs.
Restrictions can be on access methods or originating URLs of client applications.
You can enable CORS with Cruise Control using the `webserver.http.cors.enabled` property in the `config`.
When enabled, CORS permits read access to the Cruise Control REST API from applications that have different originating URLs than Strimzi.
This allows applications from specified origins to use `GET` requests to fetch information about the Kafka cluster through the Cruise Control API.
For example, applications can fetch information on the current cluster load or the most recent optimization proposal.
`POST` requests are not permitted.

NOTE: For more information on using CORS with Cruise Control, see {CruiseControlApiDocs}.

.Enabling CORS for Cruise Control

You enable and configure CORS in `Kafka.spec.cruiseControl.config`.
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  # ...
  cruiseControl:
    # ...
    config:
      webserver.http.cors.enabled: true # <1>
      webserver.http.cors.origin: "*" # <2>
      webserver.http.cors.exposeheaders: "User-Task-ID,Content-Type" # <3>

    # ...
----
<1> Enables CORS.
<2> Specifies permitted origins for the `Access-Control-Allow-Origin` HTTP response header. You can use a wildcard or specify a single origin as a URL. If you use a wildcard, a response is returned following requests from any origin.
<3> Exposes specified header names for the `Access-Control-Expose-Headers` HTTP response header. Applications in permitted origins can read responses with the specified headers.

= Cruise Control REST API security

The Cruise Control REST API is secured with HTTP Basic authentication and SSL to protect the cluster against potentially destructive Cruise Control operations, such as decommissioning Kafka brokers.
We recommend that Cruise Control in Strimzi is **only used with these settings enabled**.

However, it is possible to disable these settings by specifying the following Cruise Control configuration:

* To disable the built-in HTTP Basic authentication, set `webserver.security.enable` to `false`.
* To disable the built-in SSL, set `webserver.ssl.enable` to `false`.

.Cruise Control configuration to disable API authorization, authentication, and SSL
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  # ...
  cruiseControl:
    config:
      webserver.security.enable: false
      webserver.ssl.enable: false
# ...
----

[id='property-cruise-control-capacity-api-users-{context}']
= API users

With the necessary permissions, create REST API users to safely access a secured Cruise Control REST API directly.

This allows roles and permissions to be defined to allow advanced users and third-party applications to access the Cruise Control REST API without having to disable basic HTTP authentication.

The following use cases would benefit from accessing the Cruise Control API without disabling API security:

* Monitoring a Strimzi-managed Kafka cluster with the Cruise Control user interface.
* Gathering Cruise Control-specific statistical information not available through Strimzi or Cruise Control sensor metrics, such as detailed information surrounding cluster and partition load and user tasks.
* Debugging Cruise Control in a secured environment.

Cruise Control reads authentication credentials for API users in Jetty's `HashLoginService` file format.

Standard Cruise Control `USER` and `VIEWER` roles are supported.

* `USER` has access to all the `GET` endpoints except `bootstrap` and `train`.
* `VIEWER` has access to `kafka_cluster_state`, `user_tasks`, and `review_board` endpoints.

In this example, we define two custom API users in the supported format in a text file called `cruise-control-auth.txt`:

[source]
----
userOne: passwordOne, USER
userTwo: passwordTwo, VIEWER
----

Then, use this file to create a secret with the following command:

[source]
----
kubectl create secret generic cruise-control-api-users-secret  --from-file=cruise-control-auth.txt=cruise-control-auth.txt
----

Next, we reference the secret in the `spec.cruiseControl.apiUsers` section of the Kafka resource:

.Example Cruise Control apiUsers configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  # ...
  cruiseControl:
    # ...
    apiUsers:
      type: hashLoginService
      valueFrom:
        secretKeyRef:
          name: cruise-control-api-users-secret
          key: cruise-control-auth.txt
     ...
----

Strimzi then decodes and uses the contents of this secret to populate Cruise Control's API authentication credentials file.

[id='property-cruise-control-broker-capacity-{context}']
= Configuring capacity limits

Cruise Control uses capacity limits to determine if optimization goals for resource capacity limits are being broken.
There are four goals of this type:

* `DiskCapacityGoal`            - Disk utilization capacity
* `CpuCapacityGoal`             - CPU utilization capacity
* `NetworkInboundCapacityGoal`  - Network inbound utilization capacity
* `NetworkOutboundCapacityGoal` - Network outbound utilization capacity

You specify capacity limits for Kafka broker resources in the `brokerCapacity` property in `Kafka.spec.cruiseControl` .
They are enabled by default and you can change their default values.
Capacity limits can be set for the following broker resources:

* `cpu`             - CPU resource in millicores or CPU cores (Default: 1)
* `inboundNetwork`  - Inbound network throughput in byte units per second (Default: 10000KiB/s)
* `outboundNetwork` - Outbound network throughput in byte units per second (Default: 10000KiB/s)

For network throughput, use an integer value with standard Kubernetes byte units (K, M, G) or their bibyte (power of two) equivalents (Ki, Mi, Gi) per second.

NOTE: Disk and CPU capacity limits are automatically generated by Strimzi, so you do not need to set them.
In order to guarantee accurate rebalance proposals when using CPU goals, you can set CPU requests equal to CPU limits in `Kafka.spec.kafka.resources`.
That way, all CPU resources are reserved upfront and are always available.
This configuration allows Cruise Control to properly evaluate the CPU utilization when preparing the rebalance proposals based on CPU goals.
In cases where you cannot set CPU requests equal to CPU limits in `Kafka.spec.kafka.resources`, you can set the CPU capacity manually for the same accuracy.

.Example Cruise Control brokerCapacity configuration using bibyte units
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  # ...
  cruiseControl:
    # ...
    brokerCapacity:
      cpu: "2"
      inboundNetwork: 10000KiB/s
      outboundNetwork: 10000KiB/s
    # ...
----

[id='property-cruise-control-capacity-overrides-{context}']
= Configuring capacity overrides

Brokers might be running on nodes with heterogeneous network or CPU resources.
If that's the case, specify `overrides` that set the network capacity and CPU limits for each broker.
The overrides ensure an accurate rebalance between the brokers.
Override capacity limits can be set for the following broker resources:

* `cpu`             - CPU resource in millicores or CPU cores (Default: 1)
* `inboundNetwork`  - Inbound network throughput in byte units per second (Default: 10000KiB/s)
* `outboundNetwork` - Outbound network throughput in byte units per second (Default: 10000KiB/s)

.An example of Cruise Control capacity overrides configuration using bibyte units
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  # ...
  cruiseControl:
    # ...
    brokerCapacity:
      cpu: "1"
      inboundNetwork: 10000KiB/s
      outboundNetwork: 10000KiB/s
      overrides:
      - brokers: [0]
        cpu: "2.755"
        inboundNetwork: 20000KiB/s
        outboundNetwork: 20000KiB/s
      - brokers: [1, 2]
        cpu: 3000m
        inboundNetwork: 30000KiB/s
        outboundNetwork: 30000KiB/s
----

CPU capacity is determined using configuration values in the following order of precedence, with the highest priority first:

. `Kafka.spec.cruiseControl.brokerCapacity.overrides.cpu` that define custom CPU capacity limits for individual brokers
. `Kafka.cruiseControl.brokerCapacity.cpu` that defines custom CPU capacity limits for all brokers in the kafka cluster
. `Kafka.spec.kafka.resources.requests.cpu` that defines the CPU resources that are reserved for each broker in the Kafka cluster.
. `Kafka.spec.kafka.resources.limits.cpu` that defines the maximum CPU resources that can be consumed by each broker in the Kafka cluster.

This order of precedence is the sequence in which different configuration values are considered when determining the actual capacity limit for a Kafka broker. 
For example, broker-specific overrides take precedence over capacity limits for all brokers.
If none of the CPU capacity configurations are specified, the default CPU capacity for a Kafka broker is set to 1 CPU core. 

For more information, refer to the xref:type-BrokerCapacity-reference[BrokerCapacity schema reference].

[id='property-cruise-control-logging-{context}']
= Logging

Cruise Control has its own configurable logger:

* `rootLogger.level`

Cruise Control uses the Apache `log4j2` logger implementation.

Use the `logging` property to configure loggers and logger levels.

You can set the log levels by specifying the logger and level directly (inline) or use a custom (external) ConfigMap.
If a ConfigMap is used, you set `logging.valueFrom.configMapKeyRef.name` property to the name of the ConfigMap containing the external logging configuration. Inside the ConfigMap, the logging configuration is described using `log4j.properties`. Both `logging.valueFrom.configMapKeyRef.name` and `logging.valueFrom.configMapKeyRef.key` properties are mandatory. A ConfigMap using the exact logging configuration specified is created with the custom resource when the Cluster Operator is running, then recreated after each reconciliation. If you do not specify a custom ConfigMap, default logging settings are used. If a specific logger value is not set, upper-level logger settings are inherited for that logger.

Here we see examples of `inline` and `external` logging.
The `inline` logging specifies the root logger level.
You can also set log levels for specific classes or loggers by adding them to the loggers property.

.Inline logging
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
# ...
spec:
  cruiseControl:
    # ...
    logging:
      type: inline
      loggers:
        rootLogger.level: INFO
        logger.exec.name: com.linkedin.kafka.cruisecontrol.executor.Executor # <1>
        logger.exec.level: TRACE # <2>
        logger.go.name: com.linkedin.kafka.cruisecontrol.analyzer.GoalOptimizer # <3>
        logger.go.level: DEBUG # <4>
    # ...
----
<1> Creates a logger for the Cruise Control `Executor` class.
<2> Sets the logging level for the `Executor` class.
<3> Creates a logger for the Cruise Control `GoalOptimizer` class.
<4> Sets the logging level for the `GoalOptimizer` class.

NOTE: When investigating an issue with Cruise Control, it's usually sufficient to change the `rootLogger` to `DEBUG` to get more detailed logs. However, keep in mind that setting the log level to `DEBUG` may result in a large amount of log output and may have performance implications.

.External logging
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
# ...
spec:
  cruiseControl:
    # ...
    logging:
      type: external
      valueFrom:
        configMapKeyRef:
          name: customConfigMap
          key: cruise-control-log4j.properties
    # ...
----

.Garbage collector (GC)

Garbage collector logging can also be enabled (or disabled) using the xref:con-common-configuration-garbage-collection-reference[`jvmOptions` property].
