handle error: storage belongs to different cluster

Author: BalabaDmitri

src/charm.py

@@ -478,6 +478,11 @@ def _on_peer_relation_changed(self, event: HookEvent):
         try:
             # Update the members of the cluster in the Patroni configuration on this unit.
             self.update_config()
+            if self._patroni.cluster_system_id_mismatch(unit_name=self.unit.name):
+                self.unit.status = BlockedStatus(
+                    "Failed to start postgresql. The storage belongs to a third-party cluster"
+                )
+                return
         except RetryError:
             self.unit.status = BlockedStatus("failed to update cluster members on member")
             return

src/cluster.py

@@ -4,6 +4,7 @@
 
 """Helper class used to manage cluster lifecycle."""
 
+import glob
 import logging
 import os
 import pwd
@@ -647,3 +648,39 @@ def update_synchronous_node_count(self, units: int = None) -> None:
                 # Check whether the update was unsuccessful.
                 if r.status_code != 200:
                     raise UpdateSyncNodeCountError(f"received {r.status_code}")
+
+    def cluster_system_id_mismatch(self, unit_name: str) -> bool:
+        """Check if the Patroni service is down.
+
+        If there is the error storage belongs to third-party cluster in its logs.
+
+        Returns:
+            "True" if an error occurred due to the fact that the storage belongs to someone else's cluster.
+        """
+        last_log_file = self._last_patroni_log_file()
+        unit_name = unit_name.replace("/", "-")
+        if (
+            f" CRITICAL: system ID mismatch, node {unit_name} belongs to a different cluster:"
+            in last_log_file
+        ):
+            return True
+        return False
+
+    def _last_patroni_log_file(self) -> str:
+        """Get last log file content of Patroni service.
+
+        If there is no available log files, empty line will be returned.
+
+        Returns:
+            Content of last log file of Patroni service.
+        """
+        log_files = glob.glob(f"{PATRONI_LOGS_PATH}/*.log")
+        if len(log_files) == 0:
+            return ""
+        latest_file = max(log_files, key=os.path.getmtime)
+        try:
+            with open(latest_file) as last_log_file:
+                return last_log_file.read()
+        except OSError as e:
+            logger.exception("Failed to read last patroni log file", exc_info=e)
+            return ""

tests/integration/ha_tests/helpers.py

@@ -807,7 +807,7 @@ def storage_id(ops_test, unit_name):
             return line.split()[1]
 
 
-async def add_unit_with_storage(ops_test, app, storage):
+async def add_unit_with_storage(ops_test, app, storage, is_blocked: bool = False):
     """Adds unit with storage.
 
     Note: this function exists as a temporary solution until this issue is resolved:
@@ -820,7 +820,14 @@ async def add_unit_with_storage(ops_test, app, storage):
     return_code, _, _ = await ops_test.juju(*add_unit_cmd)
     assert return_code == 0, "Failed to add unit with storage"
     async with ops_test.fast_forward():
-        await ops_test.model.wait_for_idle(apps=[app], status="active", timeout=2000)
+        if is_blocked:
+            application = ops_test.model.applications[app]
+            await ops_test.model.block_until(
+                lambda: "blocked" in {unit.workload_status for unit in application.units},
+                timeout=1500,
+            )
+        else:
+            await ops_test.model.wait_for_idle(apps=[app], status="active", timeout=1500)
     assert (
         len(ops_test.model.applications[app].units) == expected_units
     ), "New unit not added to model"

tests/integration/ha_tests/test_restore_cluster.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 # Copyright 2023 Canonical Ltd.
 # See LICENSE file for licensing details.
+import asyncio
 import logging
 
 import pytest
@@ -37,24 +38,23 @@ async def test_build_and_deploy(ops_test: OpsTest) -> None:
     charm = await ops_test.build_charm(".")
     async with ops_test.fast_forward():
         # Deploy the first cluster with reusable storage
-        await ops_test.model.deploy(
-            charm,
-            application_name=FIRST_APPLICATION,
-            num_units=3,
-            series=CHARM_SERIES,
-            storage={"pgdata": {"pool": "lxd-btrfs", "size": 2048}},
-            config={"profile": "testing"},
+        await asyncio.gather(
+            ops_test.model.deploy(
+                charm,
+                application_name=FIRST_APPLICATION,
+                num_units=3,
+                series=CHARM_SERIES,
+                storage={"pgdata": {"pool": "lxd-btrfs", "size": 2048}},
+                config={"profile": "testing"},
+            ),
+            ops_test.model.deploy(
+                charm,
+                application_name=SECOND_APPLICATION,
+                num_units=1,
+                series=CHARM_SERIES,
+                config={"profile": "testing"},
+            ),
         )
-
-        # Deploy the second cluster
-        await ops_test.model.deploy(
-            charm,
-            application_name=SECOND_APPLICATION,
-            num_units=1,
-            series=CHARM_SERIES,
-            config={"profile": "testing"},
-        )
-
         await ops_test.model.wait_for_idle(status="active", timeout=1500)
 
         # TODO have a better way to bootstrap clusters with existing storage

tests/integration/ha_tests/test_self_healing.py

@@ -57,6 +57,7 @@
     validate_test_data,
     wait_network_restore,
 )
+from .test_restore_cluster import SECOND_APPLICATION
 
 logger = logging.getLogger(__name__)
 
@@ -79,7 +80,15 @@ async def test_build_and_deploy(ops_test: OpsTest) -> None:
         async with ops_test.fast_forward():
             await ops_test.model.deploy(
                 charm,
-                num_units=3,
+                num_units=1,
+                series=CHARM_SERIES,
+                storage={"pgdata": {"pool": "lxd-btrfs", "size": 2048}},
+                config={"profile": "testing"},
+            )
+            await ops_test.model.deploy(
+                charm,
+                num_units=1,
+                application_name=SECOND_APPLICATION,
                 series=CHARM_SERIES,
                 storage={"pgdata": {"pool": "lxd-btrfs", "size": 2048}},
                 config={"profile": "testing"},
@@ -552,10 +561,9 @@ async def test_network_cut_without_ip_change(
 
 
 @pytest.mark.group(1)
-async def test_deploy_zero_units(ops_test: OpsTest):
+async def test_deploy_zero_units(ops_test: OpsTest, charm):
     """Scale the database to zero units and scale up again."""
     app = await app_name(ops_test)
-
     dbname = f"{APPLICATION_NAME.replace('-', '_')}_first_database"
     connection_string, _ = await get_db_connection(ops_test, dbname=dbname)
 
@@ -577,12 +585,20 @@ async def test_deploy_zero_units(ops_test: OpsTest):
         unit_ip_addresses.append(await get_unit_ip(ops_test, unit.name))
 
         # Save detached storage ID
-        if await unit.is_leader_from_status:
+        if await unit.is_leader_from_status():
             primary_storage = storage_id(ops_test, unit.name)
 
+    logger.info(f"get storage id app: {SECOND_APPLICATION}")
+    second_storage = ""
+    for unit in ops_test.model.applications[SECOND_APPLICATION].units:
+        if await unit.is_leader_from_status():
+            second_storage = storage_id(ops_test, unit.name)
+            break
+
     # Scale the database to zero units.
     logger.info("scaling database to zero units")
     await scale_application(ops_test, app, 0)
+    await scale_application(ops_test, SECOND_APPLICATION, 0)
 
     # Checking shutdown units.
     for unit_ip in unit_ip_addresses:
@@ -599,7 +615,9 @@ async def test_deploy_zero_units(ops_test: OpsTest):
     # Scale up to one unit.
     logger.info("scaling database to one unit")
     await add_unit_with_storage(ops_test, app=app, storage=primary_storage)
-    await ops_test.model.wait_for_idle(status="active", timeout=1500)
+    await ops_test.model.wait_for_idle(
+        apps=[APP_NAME, APPLICATION_NAME], status="active", timeout=1500
+    )
 
     connection_string, _ = await get_db_connection(ops_test, dbname=dbname)
     logger.info("checking whether writes are increasing")
@@ -608,6 +626,19 @@ async def test_deploy_zero_units(ops_test: OpsTest):
     logger.info("check test database data")
     await validate_test_data(connection_string)
 
+    logger.info("database scaling up to two units using third-party cluster storage")
+    new_unit = await add_unit_with_storage(
+        ops_test, app=app, storage=second_storage, is_blocked=True
+    )
+
+    logger.info(f"remove unit {new_unit.name} with storage from application {SECOND_APPLICATION}")
+    await ops_test.model.destroy_units(new_unit.name)
+
+    await are_writes_increasing(ops_test)
+
+    logger.info("check test database data")
+    await validate_test_data(connection_string)
+
     # Scale up to two units.
     logger.info("scaling database to two unit")
     prev_units = [unit.name for unit in ops_test.model.applications[app].units]