handle error: storage belongs to different cluster

Author: BalabaDmitri

src/charm.py

@@ -478,6 +478,9 @@ 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 start postgresql. Storage belongs to a foreign cluster")
+                return
         except RetryError:
             self.unit.status = BlockedStatus("failed to update cluster members on member")
             return

tests/integration/ha_tests/test_self_healing.py

@@ -9,6 +9,7 @@
 from pytest_operator.plugin import OpsTest
 from tenacity import Retrying, stop_after_delay, wait_fixed
 
+from .test_restore_cluster import SECOND_APPLICATION
 from ..helpers import (
     CHARM_SERIES,
     db_connect,
@@ -554,8 +555,19 @@ async def test_network_cut_without_ip_change(
 @pytest.mark.group(1)
 async def test_deploy_zero_units(ops_test: OpsTest):
     """Scale the database to zero units and scale up again."""
-    app = await app_name(ops_test)
+    charm = await ops_test.build_charm(".")
+    async with ops_test.fast_forward():
+        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"},
+        )
+        await ops_test.model.wait_for_idle(status="active", timeout=1500)
 
+    app = await app_name(ops_test)
     dbname = f"{APPLICATION_NAME.replace('-', '_')}_first_database"
     connection_string, _ = await get_db_connection(ops_test, dbname=dbname)
 
@@ -580,6 +592,13 @@ async def test_deploy_zero_units(ops_test: OpsTest):
         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)
@@ -601,13 +620,23 @@ async def test_deploy_zero_units(ops_test: OpsTest):
     await add_unit_with_storage(ops_test, app=app, storage=primary_storage)
     await ops_test.model.wait_for_idle(status="active", timeout=1500)
 
+    logger.info("scaling database to tow unit using foreign cluster's storage")
+    await add_unit_with_storage(ops_test, app=app, storage=second_storage)
+    await ops_test.model.block_until(
+        lambda:
+        timeout=1500
+    )
+
     connection_string, _ = await get_db_connection(ops_test, dbname=dbname)
     logger.info("checking whether writes are increasing")
     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]