// Module included in the following assemblies:
//
// assembly-cruise-control-concepts.adoc

[id='ref-cruise-control-configuration-{context}']
= Cruise Control configuration

The `config` property in `Kafka.spec.cruiseControl` contains configuration options as keys with values as one of the following JSON types:

* String
* Number
* Boolean

NOTE: Strings that look like JSON or YAML will need to be explicitly quoted.

You can specify and configure all the options listed in the "Configurations" section of the {CruiseControlConfigDocs}, apart from those managed directly by {ProductName}.
Specifically, you *cannot* modify configuration options with keys equal to or starting with one of the following strings:

* `bootstrap.servers`
* `zookeeper.`
* `ssl.`
* `security.`
* `failed.brokers.zk.path`
* `webserver.http.port`
* `webserver.http.address`
* `webserver.api.urlprefix`
* `metric.reporter.sampler.bootstrap.servers`
* `metric.reporter.topic`
* `metric.reporter.topic.pattern`
* `partition.metric.sample.store.topic`
* `broker.metric.sample.store.topic`
* `capacity.config.file`
* `skip.sample.store.topic.rack.awareness.check`
* `cruise.control.metrics.topic`
* `sasl.`

If restricted options are specified, they are ignored and a warning message is printed to the Cluster Operator log file.
All supported options are passed to Cruise Control.

.An example Cruise Control configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  # ...
  cruiseControl:
    # ...
    config:
      default.goals: >
         com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,
         com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal
      cpu.balance.threshold: 1.1
      metadata.max.age.ms: 300000
      send.buffer.bytes: 131072
    # ...
----
== Capacity configuration

Cruise Control uses _capacity limits_ to determine if optimization goals for broker resources are being broken.
An optimization will fail if a goal is broken, preventing the optimization from being used to generate an optimization proposal for balanced partition assignments.
For example, an optimization that would cause a broker to exceed its CPU capacity would not be used if the `CpuCapacityGoal` is set as a hard goal.

You specify capacity limits for broker resources in the `brokerCapacity` property in `Kafka.spec.cruiseControl` .
Capacity limits can be set for the following broker resources in the described units:

* `disk`            - Disk storage in bytes
* `cpuUtilization`  - CPU utilization as a percent (0-100)
* `inboundNetwork`  - Inbound network throughput in bytes per second
* `outboundNetwork` - Outbound network throughput in bytes per second

Because {ProductName} Kafka brokers are homogeneous, Cruise Control applies the same capacity limits to every broker it is monitoring.

.An example Cruise Control brokerCapacity configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  # ...
  cruiseControl:
    # ...
    brokerCapacity:
      disk: 100G
      cpuUtilization: 100
      inboundNetwork: 10000KB/s
      outboundNetwork: 10000KB/s
    # ...
----


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

== Topic creation and configuration

Cruise Control requires the following three topics to be created in order to function properly:

* `strimzi.cruisecontrol.partitionmetricsamples`
* `strimzi.cruisecontrol.modeltrainingsamples`
* `strimzi.cruisecontrol.metrics`

Cruise Control will create the `strimzi.cruisecontrol.partitionmetricsamples` and `strimzi.cruisecontrol.modeltrainingsamples` topics after the Cruise Control metric reporters create the `strimzi.cruisecontrol.metrics` topic.
However, if automatic topic creation is disabled in Kafka with a configuration of `auto.create.topics.enable: false` in `spec.kafka.config` when starting a new Kafka cluster, the Cruise Control metric reporters will be unable to create the `strimzi.cruisecontrol.metrics` topic.
In this situation, the `strimzi.cruisecontrol.metrics` topic will need to be created manually.

To manually create the `strimzi.cruisecontrol.metrics` Cruise Control topic in your Kubernetes cluster:
[source,shell,subs="attributes+"]
----
kubectl apply -f examples/cruise-control/cruise-control-topic.yaml
----

Since log compaction may remove records needed by Cruise Control, all topics created by Cruise Control must be configured with `cleanup.policy=delete` to disable log compaction.
Cruise Control will automatically disable log compaction for the `strimzi.cruisecontrol.partitionmetricsamples` and `strimzi.cruisecontrol.modeltrainingsamples` topics.
The Cruise Control metric reporters will attempt to disable log compaction for the `strimzi.cruisecontrol.metrics` topic but will fail when being started with a new Kafka cluster.
This will only become a problem when log compaction is enabled in Kafka with the setting `log.cleanup.policy=compact` in the `spec.kafka.config`.
In this situation, log compaction will be enabled for `strimzi.cruisecontrol.metrics` topic and will need to be overridden with a `cleanup.policy=delete` in the `strimzi.cruisecontrol.metrics` KafkaTopic.

Here we see an example of log compaction being disabled in a Cruise Control KafkaTopic.

[source,yaml,subs="+quotes,attributes"]
----
apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaTopic
metadata:
  name: strimzi.cruisecontrol.metrics
spec:
  partitions: 1
  replicas: 1
  config:
    cleanup.policy: delete
----

== Logging configuration

Cruise Control has its own configurable logger:

* `cruisecontrol.root.logger`

Cruise Control uses the Apache `log4j` 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.name` property to the name of the ConfigMap containing the external logging configuration. Inside the ConfigMap, the logging configuration is described using `log4j.properties`.

Here we see examples of `inline` and `external` logging.

.Inline logging
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
# ...
spec:
  cruiseControl:
    # ...
    logging:
      type: inline
      loggers:
        cruisecontrol.root.logger: "INFO"
    # ...
----

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