[DPE-4183] - fix: only handle quorum removal on relation-departed (#146)

Author: marcoppenheimer

.github/workflows/ci.yaml

@@ -93,6 +93,50 @@ jobs:
         env:
           CI_PACKED_CHARMS: ${{ needs.build.outputs.charms }}
 
+  integration-test-scaling:
+    strategy:
+      fail-fast: false
+      matrix:
+        tox-environments:
+          - integration-scaling
+    name: ${{ matrix.tox-environments }}
+    needs:
+      - lint
+      - unit-test
+      - build
+      - integration-test
+    runs-on: [self-hosted, linux, X64, jammy, xlarge]
+    timeout-minutes: 120
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+      - name: Setup operator environment
+        # TODO: Replace with custom image on self-hosted runner
+        uses: charmed-kubernetes/actions-operator@main
+        with:
+          provider: lxd
+          juju-channel: 3.4/stable
+          bootstrap-options: "--agent-version 3.4.2"
+      - name: Download packed charm(s)
+        uses: actions/download-artifact@v3
+        with:
+          name: ${{ needs.build.outputs.artifact-name }}
+      - name: Select tests
+        id: select-tests
+        run: |
+          if [ "${{ github.event_name }}" == "schedule" ]
+          then
+            echo Running unstable and stable tests
+            echo "mark_expression=" >> $GITHUB_OUTPUT
+          else
+            echo Skipping unstable tests
+            echo "mark_expression=not unstable" >> $GITHUB_OUTPUT
+          fi
+      - name: Run integration tests
+        run: tox run -e ${{ matrix.tox-environments }} -- -m '${{ steps.select-tests.outputs.mark_expression }}'
+        env:
+          CI_PACKED_CHARMS: ${{ needs.build.outputs.charms }}
+
   integration-test-ha:
     strategy:
       fail-fast: false
@@ -105,6 +149,7 @@ jobs:
       - unit-test
       - build
       - integration-test
+      - integration-test-scaling
     runs-on: ubuntu-latest
     timeout-minutes: 120
     steps:

metadata.yaml

@@ -42,5 +42,5 @@ storage:
   data:
     type: filesystem
     description: Directories where snapshot and transaction data is stored
-    minimum-size: 10G
+    minimum-size: 1G
     location: /var/snap/charmed-zookeeper/common/var/lib/zookeeper

src/charm.py

@@ -9,10 +9,11 @@
 
 from charms.grafana_agent.v0.cos_agent import COSAgentProvider
 from charms.rolling_ops.v0.rollingops import RollingOpsManager
+from charms.zookeeper.v0.client import QuorumLeaderNotFoundError
+from kazoo.exceptions import BadVersionError, ReconfigInProcessError
 from ops.charm import (
     CharmBase,
     InstallEvent,
-    LeaderElectedEvent,
     RelationDepartedEvent,
     SecretChangedEvent,
     StorageAttachedEvent,
@@ -115,7 +116,7 @@ def __init__(self, *args):
             getattr(self.on, "cluster_relation_joined"), self._on_cluster_relation_changed
         )
         self.framework.observe(
-            getattr(self.on, "cluster_relation_departed"), self._on_cluster_relation_changed
+            getattr(self.on, "cluster_relation_departed"), self._on_cluster_relation_departed
         )
 
         self.framework.observe(
@@ -183,10 +184,6 @@ def _on_cluster_relation_changed(self, event: EventBase) -> None:
         # even if leader has not started, attempt update quorum
         self.update_quorum(event=event)
 
-        # don't delay scale-down leader ops by restarting dying unit
-        if getattr(event, "departing_unit", None) == self.unit:
-            return
-
         # check whether restart is needed for all `*_changed` events
         # only restart where necessary to avoid slowdowns
         # config_changed call here implicitly updates jaas + zoo.cfg
@@ -212,9 +209,39 @@ def _on_cluster_relation_changed(self, event: EventBase) -> None:
             self._set_status(Status.SERVICE_UNHEALTHY)
             return
 
+        # in case server was erroneously removed from the quorum
+        if not self.state.stale_quorum and not self.quorum_manager.server_in_quorum:
+            self._set_status(Status.SERVICE_NOT_QUORUM)
+            return
+
         self._set_status(Status.ACTIVE)
 
-    def _on_storage_attached(self, event: StorageAttachedEvent) -> None:
+    def _on_cluster_relation_departed(self, event: RelationDepartedEvent) -> None:
+        """Handler for `relation_departed` events."""
+        # is related to issue found in https://bugs.launchpad.net/juju/+bug/2053055
+        # likely due to a controller upgrade or a cloud maintenance with machines being reshuffled
+        # periodically, juju would emit a LeaderElected event, and would return no peer units
+        # the leader would then remove all other units from the quorum, which when restarted, would fail
+        if not event.departing_unit:
+            return
+
+        departing_server_id = (
+            int(event.departing_unit.name.split("/")[1]) + 1
+        )  # server-ids must be positive integers
+
+        try:
+            self.quorum_manager.client.remove_members(members=[f"server.{departing_server_id}"])
+        except (
+            ReconfigInProcessError,  # another unit already handling
+            BadVersionError,  # another unit handled
+            QuorumLeaderNotFoundError,  # this unit is departing, can't find leader in peer data
+        ):
+            pass
+
+        # NOTE: if the leader is also going down, it may miss the event to set {unit.id: removed}
+        # to avoid this, eventual clean up occurs during update-status calling update_quorum
+
+    def _on_storage_attached(self, _: StorageAttachedEvent) -> None:
         """Handler for `storage_attached` events."""
         self.workload.exec(["chmod", "750", f"{self.workload.paths.data_path}"])
         self.workload.exec(["chown", f"{USER}:{GROUP}", f"{self.workload.paths.data_path}"])
@@ -356,10 +383,6 @@ def update_quorum(self, event: EventBase) -> None:
 
         if (
             self.state.stale_quorum  # in the case of scale-up
-            or isinstance(  # to run without delay to maintain quorum on scale down
-                event,
-                (RelationDepartedEvent, LeaderElectedEvent),
-            )
             or self.state.healthy  # to ensure run on update-status
         ):
             updated_servers = self.quorum_manager.update_cluster()
@@ -386,7 +409,7 @@ def update_quorum(self, event: EventBase) -> None:
                 logger.debug("tls disabled - switching to non-ssl")
                 self.state.cluster.update({"quorum": "non-ssl"})
 
-            if self.state.all_units_quorum:
+            if self.state.all_units_same_encryption:
                 logger.debug(
                     "all units running desired encryption - removing switching-encryption"
                 )

src/core/cluster.py

@@ -315,7 +315,7 @@ def all_units_unified(self) -> bool:
         return True
 
     @property
-    def all_units_quorum(self) -> bool:
+    def all_units_same_encryption(self) -> bool:
         """Flag to check if all units are using the same quorum encryption."""
         if not self.cluster:
             return False
@@ -369,7 +369,7 @@ def stable(self) -> Status:
     @property
     def ready(self) -> Status:
         """Gets appropriate Status if the charm is ready to handle related applications."""
-        if not self.all_units_quorum:
+        if not self.all_units_same_encryption:
             return Status.NOT_ALL_QUORUM
 
         if self.cluster.switching_encryption:

src/core/models.py

@@ -191,7 +191,6 @@ def __init__(
         substrate: SUBSTRATES,
     ):
         super().__init__(relation, data_interface, component, substrate)
-        # Lint :-/ It can't resolve the subtype otherwise, even though the same assignment happens in super()
         self.data_interface = data_interface
         self.app = component
 

src/literals.py

@@ -68,6 +68,7 @@ class Status(Enum):
         BlockedStatus("unable to install zookeeper service"), "ERROR"
     )
     SERVICE_NOT_RUNNING = StatusLevel(BlockedStatus("zookeeper service not running"), "ERROR")
+    SERVICE_NOT_QUORUM = StatusLevel(BlockedStatus("unit not in the zookeeper quorum"), "ERROR")
     CONTAINER_NOT_CONNECTED = StatusLevel(
         MaintenanceStatus("zookeeper container not ready"), "DEBUG"
     )

src/managers/quorum.py

@@ -22,6 +22,7 @@
 from ops.charm import RelationEvent
 
 from core.cluster import ClusterState
+from core.models import ZKServer
 from literals import CLIENT_PORT
 
 logger = logging.getLogger(__name__)
@@ -102,17 +103,19 @@ def get_hostname_mapping(self) -> dict[str, str]:
 
         return {"hostname": hostname, "fqdn": fqdn, "ip": ip}
 
-    def _get_updated_servers(self, add: list[str], remove: list[str]) -> dict[str, str]:
+    def _get_updated_servers(self, add: list[str]) -> dict[str, str]:
         """Simple wrapper for building `updated_servers` for passing to app data updates."""
-        servers_to_update = add + remove
-
         updated_servers = {}
-        for server_string in servers_to_update:
+        for server_string in add:
             unit_id = str(int(re.findall(r"server.([0-9]+)", server_string)[0]) - 1)
             if server_string in add:
                 updated_servers[unit_id] = "added"
-            elif server_string in remove:
-                updated_servers[unit_id] = "removed"
+
+        # settings units to removed that were handled in relation-departed
+        # also set here in case leader missed the event
+        for added_id in self.state.cluster.added_unit_ids:
+            if added_id not in [server.unit_id for server in self.state.servers]:
+                updated_servers[str(added_id)] = "removed"
 
         return updated_servers
 
@@ -123,7 +126,6 @@ def update_cluster(self) -> dict[str, str]:
 
         After grabbing all the "started" units that the leader can see in the peer relation
             unit data.
-        Removes members not in the quorum anymore (i.e `relation_departed`/`leader_elected` event)
         Adds new members to the quorum (i.e `relation_joined` event).
 
         Returns:
@@ -138,21 +140,12 @@ def update_cluster(self) -> dict[str, str]:
         active_server_strings = {server.server_string for server in self.state.started_servers}
 
         try:
-            # remove units first, faster due to no startup/sync delay
-            zk_members = self.client.server_members
-            servers_to_remove = list(zk_members - active_server_strings)
-            logger.debug(f"{servers_to_remove=}")
-
-            self.client.remove_members(members=servers_to_remove)
-
             # sorting units to ensure units are added in id order
             zk_members = self.client.server_members
             servers_to_add = sorted(active_server_strings - zk_members)
-            logger.debug(f"{servers_to_add=}")
-
             self.client.add_members(members=servers_to_add)
 
-            return self._get_updated_servers(add=servers_to_add, remove=servers_to_remove)
+            return self._get_updated_servers(add=servers_to_add)
 
         # caught errors relate to a unit/zk_server not yet being ready to change
         except (
@@ -165,6 +158,17 @@ def update_cluster(self) -> dict[str, str]:
             logger.warning(str(e))
             return {}
 
+    def server_in_quorum(self, server: ZKServer) -> bool:
+        """Checks if server is in current quorum.
+
+        Args:
+            server: the server to check
+
+        Returns:
+            True if server is found in the quorum. Otherwise False.
+        """
+        return server.server_string in self.client.server_members
+
     @staticmethod
     def _is_child_of(path: str, chroots: Set[str]) -> bool:
         """Checks if given path is a child znode from a set of chroot paths.

tests/integration/ha/test_scaling.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+import logging
+
+import helpers
+import pytest
+from pytest_operator.plugin import OpsTest
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.skip_if_deployed
+@pytest.mark.abort_on_fail
+async def test_deploy_active(ops_test: OpsTest):
+    charm = await ops_test.build_charm(".")
+    await ops_test.model.deploy(
+        charm,
+        application_name=helpers.APP_NAME,
+        num_units=3,
+        storage={"data": {"pool": "lxd-btrfs", "size": 10240}},
+    )
+    await helpers.wait_idle(ops_test, units=3)
+
+
+@pytest.mark.abort_on_fail
+async def test_simple_scale_up(ops_test: OpsTest):
+    await ops_test.model.applications[helpers.APP_NAME].add_units(count=3)
+    await helpers.wait_idle(ops_test, units=6)
+
+
+@pytest.mark.abort_on_fail
+async def test_simple_scale_down(ops_test: OpsTest):
+    await ops_test.model.applications[helpers.APP_NAME].destroy_units(
+        f"{helpers.APP_NAME}/5", f"{helpers.APP_NAME}/4", f"{helpers.APP_NAME}/3"
+    )
+    await helpers.wait_idle(ops_test, units=3)
+
+    # scaling back up
+    await ops_test.model.applications[helpers.APP_NAME].add_units(count=3)
+    await helpers.wait_idle(ops_test, units=6)
+
+
+@pytest.mark.abort_on_fail
+async def test_complex_scale_down(ops_test: OpsTest):
+    hosts = helpers.get_hosts(ops_test)
+
+    quorum_leader_name = helpers.get_leader_name(ops_test, hosts)
+    charm_leader_name = None
+    other_unit_name = None
+
+    for unit in ops_test.model.applications[helpers.APP_NAME].units:
+        if await unit.is_leader_from_status():
+            charm_leader_name = unit.name
+
+        if unit.name not in [charm_leader_name, quorum_leader_name]:
+            other_unit_name = unit.name
+
+    await ops_test.model.applications[helpers.APP_NAME].destroy_units(
+        quorum_leader_name, charm_leader_name, other_unit_name
+    )
+    await helpers.wait_idle(ops_test, units=3)