[DPE-5591] refactor: Rework status handling (#141)

Author: Batalex

poetry.lock

@@ -31,7 +31,7 @@ package-mode = false
 
 [tool.poetry.dependencies]
 python = "^3.10"
-ops = ">=2.4.1"
+ops = ">=2.17.0"
 kazoo = ">=2.8.0"
 tenacity = ">=8.0.1"
 pure-sasl = ">=0.6.2"
@@ -85,7 +85,7 @@ pytest = ">=7.2"
 coverage = { extras = ["toml"], version = ">7.0" }
 pytest-mock = "^3.11.1"
 packaging = "^23.1"
-ops-scenario = "^7.0.0"
+ops = { version = ">=2.17.0", extras = ["testing"] }
 
 [tool.poetry.group.security]
 optional = true

pyproject.toml

@@ -3,7 +3,7 @@ attrs==24.2.0 ; python_version >= "3.10" and python_version < "4.0"
 certifi==2024.8.30 ; python_version >= "3.10" and python_version < "4.0"
 cffi==1.17.1 ; python_version >= "3.10" and python_version < "4.0" and platform_python_implementation != "PyPy"
 charset-normalizer==3.4.0 ; python_version >= "3.10" and python_version < "4.0"
-cryptography==43.0.1 ; python_version >= "3.10" and python_version < "4.0"
+cryptography==43.0.3 ; python_version >= "3.10" and python_version < "4.0"
 exceptiongroup==1.2.2 ; python_version >= "3.10" and python_version < "3.11"
 h11==0.14.0 ; python_version >= "3.10" and python_version < "4.0"
 httpcore==1.0.6 ; python_version >= "3.10" and python_version < "4.0"

requirements.txt

@@ -12,7 +12,6 @@
 from charms.prometheus_k8s.v0.prometheus_scrape import MetricsEndpointProvider
 from charms.rolling_ops.v0.rollingops import RollingOpsManager
 from ops import (
-    ActiveStatus,
     CollectStatusEvent,
     EventBase,
     StatusBase,
@@ -51,6 +50,7 @@ def __init__(self, *args):
         super().__init__(*args)
         self.name = CHARM_KEY
         self.substrate: Substrates = SUBSTRATE
+        self.pending_inactive_statuses: list[Status] = []
 
         # Common attrs init
         self.state = ClusterState(self, substrate=self.substrate)
@@ -62,6 +62,7 @@ def __init__(self, *args):
         self.restart = RollingOpsManager(self, relation="restart", callback=self._restart_broker)
 
         self.framework.observe(getattr(self.on, "config_changed"), self._on_roles_changed)
+        self.framework.observe(self.on.collect_unit_status, self._on_collect_status)
         self.framework.observe(self.on.collect_app_status, self._on_collect_status)
 
         # peer-cluster events are shared between all roles, so necessary to init here to avoid instantiating multiple times
@@ -128,24 +129,12 @@ def _set_status(self, key: Status) -> None:
         log_level: DebugLevel = key.value.log_level
 
         getattr(logger, log_level.lower())(status.message)
-        self.unit.status = status
+        # self.unit.status = status
+        self.pending_inactive_statuses.append(key)
 
     def _on_collect_status(self, event: CollectStatusEvent):
-        ready_to_start = self.state.ready_to_start.value.status
-        event.add_status(ready_to_start)
-
-        if not isinstance(ready_to_start, ActiveStatus):
-            return
-
-        if not self.state.runs_broker:
-            # early return, the next checks only concern the broker
-            return
-
-        if not self.broker.workload.active():
-            event.add_status(Status.BROKER_NOT_RUNNING.value.status)
-
-        if not self.state.zookeeper.broker_active():
-            event.add_status(Status.ZK_NOT_CONNECTED.value.status)
+        for status in self.pending_inactive_statuses + [Status.ACTIVE]:
+            event.add_status(status.value.status)
 
 
 if __name__ == "__main__":

src/charm.py

@@ -11,7 +11,6 @@
 
 from ops import (
     ActionEvent,
-    ActiveStatus,
     EventBase,
     InstallEvent,
     Object,
@@ -61,6 +60,9 @@ def __init__(self, charm) -> None:
             config=self.charm.config,
         )
 
+        # Before fast exit to avoid silently ignoring the action
+        self.framework.observe(getattr(self.charm.on, "rebalance_action"), self.rebalance)
+
         # Fast exit after workload instantiation, but before any event observer
         if BALANCER.value not in self.charm.config.roles or not self.charm.unit.is_leader():
             return
@@ -82,8 +84,6 @@ def __init__(self, charm) -> None:
         self.framework.observe(self.charm.on.update_status, self._on_config_changed)
         self.framework.observe(self.charm.on.config_changed, self._on_config_changed)
 
-        self.framework.observe(getattr(self.charm.on, "rebalance_action"), self.rebalance)
-
     def _on_install(self, event: InstallEvent) -> None:
         """Handler for `install` event."""
         if not self.workload.container_can_connect:
@@ -101,8 +101,9 @@ def _on_install(self, event: InstallEvent) -> None:
 
     def _on_start(self, event: StartEvent | PebbleReadyEvent) -> None:
         """Handler for `start` or `pebble-ready` events."""
-        self.charm._set_status(self.charm.state.ready_to_start)
-        if not isinstance(self.charm.unit.status, ActiveStatus):
+        current_status = self.charm.state.ready_to_start
+        if current_status is not Status.ACTIVE:
+            self.charm._set_status(current_status)
             event.defer()
             return
 
@@ -206,33 +207,36 @@ def rebalance(self, event: ActionEvent) -> None:
             available_brokers = [int(broker.split("/")[1]) for broker in brokers]
 
         failure_conditions = [
-            (not self.charm.unit.is_leader(), "Action must be ran on the application leader"),
             (
-                not self.balancer_manager.cruise_control.monitoring,
+                lambda: not self.charm.unit.is_leader(),
+                "Action must be ran on the application leader",
+            ),
+            (
+                lambda: not self.balancer_manager.cruise_control.monitoring,
                 "CruiseControl balancer service is not yet ready",
             ),
             (
-                self.balancer_manager.cruise_control.executing,
+                lambda: self.balancer_manager.cruise_control.executing,
                 "CruiseControl balancer service is currently executing a task, please try again later",
             ),
             (
-                not self.balancer_manager.cruise_control.ready,
+                lambda: not self.balancer_manager.cruise_control.ready,
                 "CruiseControl balancer service has not yet collected enough data to provide a partition reallocation proposal",
             ),
             (
-                event.params["mode"] in (MODE_ADD, MODE_REMOVE)
+                lambda: event.params["mode"] in (MODE_ADD, MODE_REMOVE)
                 and event.params.get("brokerid", None) is None,
                 "'add' and 'remove' rebalance action require passing the 'brokerid' parameter",
             ),
             (
-                event.params["mode"] in (MODE_ADD, MODE_REMOVE)
+                lambda: event.params["mode"] in (MODE_ADD, MODE_REMOVE)
                 and event.params.get("brokerid") not in available_brokers,
                 "invalid brokerid",
             ),
         ]
 
         for check, msg in failure_conditions:
-            if check:
+            if check():
                 logging.error(msg)
                 event.set_results({"error": msg})
                 event.fail(msg)
@@ -260,8 +264,6 @@ def rebalance(self, event: ActionEvent) -> None:
 
         event.set_results(sanitised_response)
 
-        self.charm._set_status(Status.ACTIVE)
-
     @property
     def healthy(self) -> bool:
         """Checks and updates various charm lifecycle states.
@@ -273,8 +275,9 @@ def healthy(self) -> bool:
         if not self.charm.state.runs_balancer:
             return True
 
-        self.charm._set_status(self.charm.state.ready_to_start)
-        if not isinstance(self.charm.unit.status, ActiveStatus):
+        current_status = self.charm.state.ready_to_start
+        if current_status is not Status.ACTIVE:
+            self.charm._set_status(current_status)
             return False
 
         if not self.workload.active() and self.charm.unit.is_leader():

src/events/balancer.py

@@ -11,7 +11,6 @@
 
 from charms.operator_libs_linux.v1.snap import SnapError
 from ops import (
-    ActiveStatus,
     EventBase,
     InstallEvent,
     Object,
@@ -166,8 +165,9 @@ def _on_start(self, event: StartEvent | PebbleReadyEvent) -> None:
 
         self.update_external_services()
 
-        self.charm._set_status(self.charm.state.ready_to_start)
-        if not isinstance(self.charm.unit.status, ActiveStatus):
+        current_status = self.charm.state.ready_to_start
+        if current_status is not Status.ACTIVE:
+            self.charm._set_status(current_status)
             event.defer()
             return
 
@@ -197,7 +197,7 @@ def _on_start(self, event: StartEvent | PebbleReadyEvent) -> None:
         self.charm.on.update_status.emit()
 
         # only log once on successful 'on-start' run
-        if isinstance(self.charm.unit.status, ActiveStatus):
+        if not self.charm.pending_inactive_statuses:
             logger.info(f'Broker {self.charm.unit.name.split("/")[1]} connected')
 
     def _on_config_changed(self, event: EventBase) -> None:
@@ -312,8 +312,6 @@ def _on_update_status(self, _: UpdateStatusEvent) -> None:
             self.charm._set_status(Status.BROKER_NOT_RUNNING)
             return
 
-        self.charm._set_status(Status.ACTIVE)
-
     def _on_secret_changed(self, event: SecretChangedEvent) -> None:
         """Handler for `secret_changed` events."""
         if not event.secret.label or not self.charm.state.cluster.relation:
@@ -387,8 +385,9 @@ def healthy(self) -> bool:
         Returns:
             True if service is alive and active. Otherwise False
         """
-        self.charm._set_status(self.charm.state.ready_to_start)
-        if not isinstance(self.charm.unit.status, ActiveStatus):
+        current_status = self.charm.state.ready_to_start
+        if current_status is not Status.ACTIVE:
+            self.charm._set_status(current_status)
             return False
 
         if not self.workload.active():

src/events/broker.py

@@ -26,7 +26,6 @@
     RelationJoinedEvent,
 )
 from ops.framework import Object
-from ops.model import ActiveStatus
 
 from literals import TLS_RELATION, TRUSTED_CA_RELATION, TRUSTED_CERTIFICATE_RELATION, Status
 
@@ -138,7 +137,6 @@ def _trusted_relation_created(self, event: EventBase) -> None:
 
         # Create a "mtls" flag so a new listener (CLIENT_SSL) is created
         self.charm.state.cluster.update({"mtls": "enabled"})
-        self.charm.app.status = ActiveStatus()
 
     def _trusted_relation_joined(self, event: RelationJoinedEvent) -> None:
         """Generate a CSR so the tls-certificates operator works as expected."""

src/events/tls.py

@@ -121,8 +121,13 @@ def _on_zookeeper_broken(self, _: RelationEvent) -> None:
         # Kafka keeps a meta.properties in every log.dir with a unique ClusterID
         # this ID is provided by ZK, and removing it on relation-broken allows
         # re-joining to another ZK cluster.
-        for storage in self.charm.model.storages["data"]:
-            self.charm.workload.exec(["rm", f"{storage.location}/meta.properties"])
+        self.charm.workload.exec(
+            [
+                "bash",
+                "-c",
+                f"""find {self.charm.workload.paths.data_path} -type f -name meta.properties -delete || true""",
+            ]
+        )
 
         if not self.charm.unit.is_leader():
             return