[DPE-5611] Remove cruise-control metrics reporter if no balancer (#140)

Author: Batalex

pyproject.toml

@@ -27,6 +27,7 @@ name = "kafka-k8s-operator"
 version = "1.0"
 description = "kafka-k8s-operator"
 authors = []
+package-mode = false
 
 [tool.poetry.dependencies]
 python = "^3.10"

src/managers/config.py

@@ -21,6 +21,7 @@
 from core.workload import CharmedKafkaPaths, WorkloadBase
 from literals import (
     ADMIN_USER,
+    BALANCER,
     BALANCER_GOALS_TESTING,
     BROKER,
     DEFAULT_BALANCER_GOALS,
@@ -43,6 +44,8 @@
 authorizer.class.name=kafka.security.authorizer.AclAuthorizer
 allow.everyone.if.no.acl.found=false
 auto.create.topics.enable=false
+"""
+KAFKA_CRUISE_CONTROL_OPTIONS = """
 metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter
 """
 TESTING_OPTIONS = """
@@ -596,13 +599,16 @@ def server_properties(self) -> list[str]:
             + self.default_replication_properties
             + self.auth_properties
             + self.rack_properties
-            + self.metrics_reporter_properties
             + DEFAULT_CONFIG_OPTIONS.split("\n")
         )
 
         if self.state.cluster.tls_enabled and self.state.unit_broker.certificate:
             properties += self.tls_properties + self.zookeeper_tls_properties
 
+        if self.state.runs_balancer or BALANCER.value in self.state.peer_cluster.roles:
+            properties += KAFKA_CRUISE_CONTROL_OPTIONS.splitlines()
+            properties += self.metrics_reporter_properties
+
         if self.config.profile == PROFILE_TESTING:
             properties += TESTING_OPTIONS.split("\n")
 

tests/integration/test_balancer.py

@@ -326,9 +326,9 @@ async def test_tls(self, ops_test: OpsTest):
 
         await ops_test.model.wait_for_idle(
             apps=list({APP_NAME, ZK_NAME, self.balancer_app}),
-            status="active",
             idle_period=30,
             timeout=1800,
+            raise_on_error=False,
         )
         async with ops_test.fast_forward(fast_interval="30s"):
             await asyncio.sleep(120)  # ensure update-status adds broker-capacities if missed

tests/unit/test_config.py

@@ -25,6 +25,7 @@
     JVM_MEM_MIN_GB,
     OAUTH_REL_NAME,
     PEER,
+    PEER_CLUSTER_ORCHESTRATOR_RELATION,
     REL_NAME,
     SUBSTRATE,
     ZK,
@@ -575,3 +576,19 @@ def test_super_users(harness: Harness[KafkaCharm]):
     harness.update_relation_data(appii_relation_id, "appii", {"extra-user-roles": "consumer"})
 
     assert len(harness.charm.state.super_users.split(";")) == (len(INTERNAL_USERS) + 1)
+
+
+def test_cruise_control_reporter_only_with_balancer(harness: Harness[KafkaCharm]):
+    reporters_config_value = "metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter"
+    # Default roles value does not include balancer
+    assert reporters_config_value not in harness.charm.broker.config_manager.server_properties
+
+    with harness.hooks_disabled():
+        peer_cluster_relation_id = harness.add_relation(
+            PEER_CLUSTER_ORCHESTRATOR_RELATION, CHARM_KEY
+        )
+        harness.update_relation_data(
+            peer_cluster_relation_id, harness.charm.app.name, {"roles": "broker,balancer"}
+        )
+
+    assert reporters_config_value in harness.charm.broker.config_manager.server_properties