[DPE-5226] - refactor: make 'broker' the central relation (#148)

Author: Batalex

src/core/cluster.py

@@ -79,14 +79,14 @@ class PeerClusterOrchestratorData(ProviderData, RequirerData):
     """Broker provider data model."""
 
     SECRET_LABEL_MAP = SECRET_LABEL_MAP
-    SECRET_FIELDS = BALANCER.requested_secrets
+    SECRET_FIELDS = BROKER.requested_secrets
 
 
 class PeerClusterData(ProviderData, RequirerData):
     """Broker provider data model."""
 
     SECRET_LABEL_MAP = SECRET_LABEL_MAP
-    SECRET_FIELDS = BROKER.requested_secrets
+    SECRET_FIELDS = BALANCER.requested_secrets
 
 
 class ClusterState(Object):
@@ -126,42 +126,43 @@ def client_relations(self) -> set[Relation]:
         return set(self.model.relations[REL_NAME])
 
     @property
-    def peer_cluster_orchestrator_relations(self) -> set[Relation]:
-        """The `peer-cluster-orchestrator` relations that this charm is providing."""
-        return set(self.model.relations[PEER_CLUSTER_ORCHESTRATOR_RELATION])
+    def peer_cluster_orchestrator_relation(self) -> Relation | None:
+        """The `peer-cluster-orchestrator` relation that this charm is providing."""
+        return self.model.get_relation(PEER_CLUSTER_ORCHESTRATOR_RELATION)
 
     @property
     def peer_cluster_relation(self) -> Relation | None:
         """The `peer-cluster` relation that this charm is requiring."""
         return self.model.get_relation(PEER_CLUSTER_RELATION)
 
     @property
-    def peer_clusters(self) -> set[PeerCluster]:
-        """The state for all related `peer-cluster` applications that this charm is providing for."""
-        peer_clusters = set()
-        balancer_kwargs: dict[str, Any] = {
-            "balancer_username": self.cluster.balancer_username,
-            "balancer_password": self.cluster.balancer_password,
-            "balancer_uris": self.cluster.balancer_uris,
-        }
-        for relation in self.peer_cluster_orchestrator_relations:
-            if not relation.app or not self.runs_balancer:
-                continue
-
-            peer_clusters.add(
-                PeerCluster(
-                    relation=relation,
-                    data_interface=PeerClusterOrchestratorData(self.model, relation.name),
-                    **balancer_kwargs,
-                )
-            )
+    def peer_cluster_orchestrator(self) -> PeerCluster:
+        """The state for the related `peer-cluster-orchestrator` application that this charm is requiring from."""
+        balancer_kwargs: dict[str, Any] = (
+            {
+                "balancer_username": self.cluster.balancer_username,
+                "balancer_password": self.cluster.balancer_password,
+                "balancer_uris": self.cluster.balancer_uris,
+            }
+            if self.runs_balancer
+            else {}
+        )
+        return PeerCluster(
+            relation=self.peer_cluster_relation,
+            data_interface=PeerClusterData(self.model, PEER_CLUSTER_RELATION),
+            **balancer_kwargs,
+        )
 
-        return peer_clusters
+    @property
+    def peer_cluster(self) -> PeerCluster:
+        """The state for the related `peer-cluster` application that this charm is providing to."""
+        return PeerCluster(
+            relation=self.peer_cluster_orchestrator_relation,
+            data_interface=PeerClusterOrchestratorData(
+                self.model, PEER_CLUSTER_ORCHESTRATOR_RELATION
+            ),
+        )
 
-    # FIXME: will need renaming once we use Kraft as the orchestrator
-    # uses the 'already there' BALANCER username now
-    # will need to create one independently with Basic HTTP auth + multiple broker apps
-    # right now, multiple<->multiple is very brittle
     @property
     def balancer(self) -> PeerCluster:
         """The state for the `peer-cluster-orchestrator` related balancer application."""
@@ -175,10 +176,12 @@ def balancer(self) -> PeerCluster:
             else {}
         )
 
-        if self.runs_broker:  # must be requiring, initialise with necessary broker data
+        if self.runs_broker:  # must be providing, initialise with necessary broker data
             return PeerCluster(
-                relation=self.peer_cluster_relation,  # if same app, this will be None and OK
-                data_interface=PeerClusterData(self.model, PEER_CLUSTER_RELATION),
+                relation=self.peer_cluster_orchestrator_relation,  # if same app, this will be None and OK
+                data_interface=PeerClusterOrchestratorData(
+                    self.model, PEER_CLUSTER_ORCHESTRATOR_RELATION
+                ),
                 broker_username=ADMIN_USER,
                 broker_password=self.cluster.internal_user_credentials.get(ADMIN_USER, ""),
                 broker_uris=self.bootstrap_server,
@@ -191,9 +194,7 @@ def balancer(self) -> PeerCluster:
             )
 
         else:  # must be roles=balancer only then, only load with necessary balancer data
-            return list(self.peer_clusters)[
-                0
-            ]  # for broker - balancer relation, currently limited to 1
+            return self.peer_cluster_orchestrator
 
     @property
     def oauth_relation(self) -> Relation | None:
@@ -345,7 +346,7 @@ def default_auth(self) -> AuthMap:
     def enabled_auth(self) -> list[AuthMap]:
         """The currently enabled auth.protocols and their auth.mechanisms, based on related applications."""
         enabled_auth = []
-        if self.client_relations or self.runs_balancer or self.peer_cluster_relation:
+        if self.client_relations or self.runs_balancer or self.peer_cluster_orchestrator_relation:
             enabled_auth.append(self.default_auth)
         if self.oauth_relation:
             enabled_auth.append(AuthMap(self.default_auth.protocol, "OAUTHBEARER"))
@@ -457,7 +458,7 @@ def _balancer_status(self) -> Status:
         if not self.runs_balancer or not self.unit_broker.unit.is_leader():
             return Status.ACTIVE
 
-        if not self.peer_cluster_orchestrator_relations and not self.runs_broker:
+        if not self.peer_cluster_relation and not self.runs_broker:
             return Status.NO_PEER_CLUSTER_RELATION
 
         if not self.balancer.broker_connected:

src/core/models.py

@@ -393,17 +393,17 @@ def mtls_enabled(self) -> bool:
         return self.relation_data.get("mtls", "disabled") == "enabled"
 
     @property
-    def balancer_username(self) -> bool:
+    def balancer_username(self) -> str:
         """Persisted balancer username."""
         return self.relation_data.get("balancer-username", "")
 
     @property
-    def balancer_password(self) -> bool:
+    def balancer_password(self) -> str:
         """Persisted balancer password."""
         return self.relation_data.get("balancer-password", "")
 
     @property
-    def balancer_uris(self) -> bool:
+    def balancer_uris(self) -> str:
         """Persisted balancer uris."""
         return self.relation_data.get("balancer-uris", "")
 

src/events/balancer.py

@@ -107,16 +107,15 @@ def _on_start(self, event: StartEvent | PebbleReadyEvent) -> None:
             return
 
         if not self.charm.state.cluster.balancer_password:
-            external_cluster = next(iter(self.charm.state.peer_clusters), None)
             payload = {
                 "balancer-username": BALANCER_WEBSERVER_USER,
                 "balancer-password": self.charm.workload.generate_password(),
                 "balancer-uris": f"{self.charm.state.unit_broker.host}:{BALANCER_WEBSERVER_PORT}",
             }
             # Update relation data intra & extra cluster (if it exists)
             self.charm.state.cluster.update(payload)
-            if external_cluster:
-                external_cluster.update(payload)
+            if self.charm.state.peer_cluster_orchestrator:
+                self.charm.state.peer_cluster_orchestrator.update(payload)
 
         self.config_manager.set_cruise_control_properties()
         self.config_manager.set_broker_capacities()

src/events/peer_cluster.py

@@ -61,7 +61,7 @@ def __init__(self, charm: "KafkaCharm") -> None:
 
         # ensures data updates, eventually
         self.framework.observe(
-            getattr(self.charm.on, "update_status"), self._on_peer_cluster_changed
+            getattr(self.charm.on, "update_status"), self._on_peer_cluster_orchestrator_changed
         )
 
     def _on_secret_changed_event(self, _: SecretChangedEvent) -> None:
@@ -94,8 +94,8 @@ def _on_peer_cluster_changed(self, event: RelationChangedEvent) -> None:
         """Generic handler for peer-cluster `relation-changed` events."""
         if (
             not self.charm.unit.is_leader()
-            or not self.charm.state.runs_broker
-            or "balancer" not in self.charm.state.balancer.roles
+            or not self.charm.state.runs_balancer  # only balancer need to handle this event
+            or not self.charm.state.balancer.roles  # ensures secrets have set-up before writing
         ):
             return
 
@@ -104,40 +104,40 @@ def _on_peer_cluster_changed(self, event: RelationChangedEvent) -> None:
         # will no-op if relation does not exist
         self.charm.state.balancer.update(
             {
-                "roles": self.charm.state.roles,
-                "broker-username": self.charm.state.balancer.broker_username,
-                "broker-password": self.charm.state.balancer.broker_password,
-                "broker-uris": self.charm.state.balancer.broker_uris,
-                "racks": str(self.charm.state.balancer.racks),
-                "broker-capacities": json.dumps(self.charm.state.balancer.broker_capacities),
-                "zk-uris": self.charm.state.balancer.zk_uris,
-                "zk-username": self.charm.state.balancer.zk_username,
-                "zk-password": self.charm.state.balancer.zk_password,
+                "balancer-username": self.charm.state.balancer.balancer_username,
+                "balancer-password": self.charm.state.balancer.balancer_password,
+                "balancer-uris": self.charm.state.balancer.balancer_uris,
             }
         )
 
         self.charm.on.config_changed.emit()  # ensure both broker+balancer get a changed event
 
     def _on_peer_cluster_orchestrator_changed(self, event: RelationChangedEvent) -> None:
         """Generic handler for peer-cluster-orchestrator `relation-changed` events."""
-        if not self.charm.unit.is_leader() or not self.charm.state.runs_balancer:
+        if (
+            not self.charm.unit.is_leader()
+            or not self.charm.state.runs_broker  # only broker needs handle this event
+            or "balancer"
+            not in self.charm.state.balancer.roles  # ensures secret have set-up before writing, and only writing to balancers
+        ):
             return
 
         self._default_relation_changed(event)
 
-        for peer_cluster in self.charm.state.peer_clusters:
-            if "broker" not in peer_cluster.roles:
-                # TODO: maybe a log here?
-                continue
-
-            # will no-op if relation does not exist
-            peer_cluster.update(
-                {
-                    "balancer-username": self.charm.state.balancer.balancer_username,
-                    "balancer-password": self.charm.state.balancer.balancer_password,
-                    "balancer-uris": self.charm.state.balancer.balancer_uris,
-                }
-            )
+        # will no-op if relation does not exist
+        self.charm.state.balancer.update(
+            {
+                "roles": self.charm.state.roles,
+                "broker-username": self.charm.state.balancer.broker_username,
+                "broker-password": self.charm.state.balancer.broker_password,
+                "broker-uris": self.charm.state.balancer.broker_uris,
+                "racks": str(self.charm.state.balancer.racks),
+                "broker-capacities": json.dumps(self.charm.state.balancer.broker_capacities),
+                "zk-uris": self.charm.state.balancer.zk_uris,
+                "zk-username": self.charm.state.balancer.zk_username,
+                "zk-password": self.charm.state.balancer.zk_password,
+            }
+        )
 
         self.charm.on.config_changed.emit()  # ensure both broker+balancer get a changed event
 

src/literals.py

@@ -128,7 +128,7 @@ def __eq__(self, value: object, /) -> bool:
     value="broker",
     service="kafka",
     paths=PATHS["kafka"],
-    relation=PEER_CLUSTER_RELATION,
+    relation=PEER_CLUSTER_ORCHESTRATOR_RELATION,
     requested_secrets=[
         "balancer-username",
         "balancer-password",
@@ -139,7 +139,7 @@ def __eq__(self, value: object, /) -> bool:
     value="balancer",
     service="balancer",
     paths=PATHS["cruise-control"],
-    relation=PEER_CLUSTER_ORCHESTRATOR_RELATION,
+    relation=PEER_CLUSTER_RELATION,
     requested_secrets=[
         "broker-username",
         "broker-password",

tests/integration/helpers.py

@@ -600,18 +600,18 @@ def balancer_is_ready(ops_test: OpsTest, app_name: str) -> bool:
     pwd = get_secret_by_label(ops_test=ops_test, label=f"{PEER}.{app_name}.app", owner=app_name)[
         "balancer-password"
     ]
-    monitor_state = check_output(
-        f"JUJU_MODEL={ops_test.model_full_name} juju ssh {app_name}/leader sudo -i 'curl http://localhost:9090/kafkacruisecontrol/state?json=True'"
-        f" -u {BALANCER_WEBSERVER_USER}:{pwd}",
-        stderr=PIPE,
-        shell=True,
-        universal_newlines=True,
-    )
 
     try:
+        monitor_state = check_output(
+            f"JUJU_MODEL={ops_test.model_full_name} juju ssh {app_name}/leader sudo -i 'curl http://localhost:9090/kafkacruisecontrol/state?json=True'"
+            f" -u {BALANCER_WEBSERVER_USER}:{pwd}",
+            stderr=PIPE,
+            shell=True,
+            universal_newlines=True,
+        )
         monitor_state_json = json.loads(monitor_state).get("MonitorState", {})
         executor_state_json = json.loads(monitor_state).get("ExecutorState", {})
-    except json.JSONDecodeError as e:
+    except (json.JSONDecodeError, CalledProcessError) as e:
         logger.error(e)
         return False
 

tests/integration/test_balancer.py

@@ -114,8 +114,8 @@ async def test_relate_not_enough_brokers(self, ops_test: OpsTest):
         await ops_test.model.add_relation(PRODUCER_APP, APP_NAME)
         if self.balancer_app != APP_NAME:
             await ops_test.model.add_relation(
-                f"{APP_NAME}:{PEER_CLUSTER_RELATION}",
-                f"{BALANCER_APP}:{PEER_CLUSTER_ORCHESTRATOR_RELATION}",
+                f"{APP_NAME}:{PEER_CLUSTER_ORCHESTRATOR_RELATION}",
+                f"{BALANCER_APP}:{PEER_CLUSTER_RELATION}",
             )
 
         await ops_test.model.wait_for_idle(

tests/integration/test_charm.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import logging
+from subprocess import PIPE, check_output
 
 import pytest
 import requests
@@ -72,17 +73,20 @@ async def test_consistency_between_workload_and_metadata(ops_test: OpsTest):
 
 @pytest.mark.abort_on_fail
 async def test_remove_zk_relation_relate(ops_test: OpsTest):
-    remove_relation_cmd = f"remove-relation {APP_NAME} {ZK_NAME}"
-    await ops_test.juju(*remove_relation_cmd.split(), check=True)
-    await ops_test.model.wait_for_idle(
-        apps=[APP_NAME, ZK_NAME], idle_period=60, timeout=3600, raise_on_error=False
+    check_output(
+        f"JUJU_MODEL={ops_test.model_full_name} juju remove-relation {APP_NAME} {ZK_NAME}",
+        stderr=PIPE,
+        shell=True,
+        universal_newlines=True,
     )
 
-    assert ops_test.model.applications[APP_NAME].status == "blocked"
-    assert ops_test.model.applications[ZK_NAME].status == "active"
+    await ops_test.model.wait_for_idle(
+        apps=[APP_NAME, ZK_NAME], idle_period=40, timeout=3600, raise_on_error=False
+    )
 
     await ops_test.model.add_relation(APP_NAME, ZK_NAME)
-    async with ops_test.fast_forward(fast_interval="60s"):
+
+    async with ops_test.fast_forward(fast_interval="90s"):
         await ops_test.model.wait_for_idle(
             apps=[APP_NAME, ZK_NAME],
             status="active",