[DPE-4118] Add support for rescanning cluster for unit rejoin after node drain (#462)

* Add support for rescanning cluster for unit rejoin after node drain

* Add get_cluster_name method to MySQL helpers class

* Delete unnecessary/unused method in MySQL helpers class

* Update data_interface charm lib + fix lint warnings

* Add unittest

Author: shayancanonical

lib/charms/data_platform_libs/v0/data_interfaces.py

@@ -331,7 +331,7 @@ def _on_topic_requested(self, event: TopicRequestedEvent):
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 37
+LIBPATCH = 38
 
 PYDEPS = ["ops>=2.0.0"]
 
@@ -2606,6 +2606,14 @@ def set_version(self, relation_id: int, version: str) -> None:
         """
         self.update_relation_data(relation_id, {"version": version})
 
+    def set_subordinated(self, relation_id: int) -> None:
+        """Raises the subordinated flag in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+        """
+        self.update_relation_data(relation_id, {"subordinated": "true"})
+
 
 class DatabaseProviderEventHandlers(EventHandlers):
     """Provider-side of the database relation handlers."""
@@ -2842,6 +2850,21 @@ def _on_relation_created_event(self, event: RelationCreatedEvent) -> None:
 
     def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
         """Event emitted when the database relation has changed."""
+        is_subordinate = False
+        remote_unit_data = None
+        for key in event.relation.data.keys():
+            if isinstance(key, Unit) and not key.name.startswith(self.charm.app.name):
+                remote_unit_data = event.relation.data[key]
+            elif isinstance(key, Application) and key.name != self.charm.app.name:
+                is_subordinate = event.relation.data[key].get("subordinated") == "true"
+
+        if is_subordinate:
+            if not remote_unit_data:
+                return
+
+            if remote_unit_data.get("state") != "ready":
+                return
+
         # Check which data has changed to emit customs events.
         diff = self._diff(event)
 

lib/charms/mysql/v0/mysql.py

@@ -130,7 +130,7 @@ def wait_until_mysql_connection(self) -> None:
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 61
+LIBPATCH = 62
 
 UNIT_TEARDOWN_LOCKNAME = "unit-teardown"
 UNIT_ADD_LOCKNAME = "unit-add"
@@ -402,6 +402,10 @@ class MySQLRejoinClusterError(Error):
     """Exception raised when there is an issue trying to rejoin a cluster to the cluster set."""
 
 
+class MySQLGetClusterNameError(Error):
+    """Exception raised when there is an issue getting cluster name."""
+
+
 @dataclasses.dataclass
 class RouterUser:
     """MySQL Router user."""
@@ -1792,7 +1796,10 @@ def is_instance_in_cluster(self, unit_label: str) -> bool:
             logger.debug(f"Checking existence of unit {unit_label} in cluster {self.cluster_name}")
 
             output = self._run_mysqlsh_script("\n".join(commands))
-            return MySQLMemberState.ONLINE in output.lower()
+            return (
+                MySQLMemberState.ONLINE in output.lower()
+                or MySQLMemberState.RECOVERING in output.lower()
+            )
         except MySQLClientError:
             # confirmation can fail if the clusteradmin user does not yet exist on the instance
             logger.debug(
@@ -1805,7 +1812,9 @@ def is_instance_in_cluster(self, unit_label: str) -> bool:
         stop=stop_after_attempt(3),
         retry=retry_if_exception_type(TimeoutError),
     )
-    def get_cluster_status(self, extended: Optional[bool] = False) -> Optional[dict]:
+    def get_cluster_status(
+        self, from_instance: Optional[str] = None, extended: Optional[bool] = False
+    ) -> Optional[dict]:
         """Get the cluster status.
 
         Executes script to retrieve cluster status.
@@ -1817,7 +1826,7 @@ def get_cluster_status(self, extended: Optional[bool] = False) -> Optional[dict]
         """
         options = {"extended": extended}
         status_commands = (
-            f"shell.connect('{self.cluster_admin_user}:{self.cluster_admin_password}@{self.instance_address}')",
+            f"shell.connect('{self.cluster_admin_user}:{self.cluster_admin_password}@{from_instance or self.instance_address}')",
             f"cluster = dba.get_cluster('{self.cluster_name}')",
             f"print(cluster.status({options}))",
         )
@@ -2265,6 +2274,34 @@ def get_cluster_set_global_primary_address(
 
         return matches.group(1)
 
+    def get_cluster_name(self, connect_instance_address: Optional[str]) -> Optional[str]:
+        """Get the cluster name from instance.
+
+        Uses the mysql_innodb_cluster_metadata.clusters table in case instance connecting
+        to is offline and thus cannot use mysqlsh.dba.get_cluster().
+        """
+        if not connect_instance_address:
+            connect_instance_address = self.instance_address
+
+        logger.debug(f"Getting cluster name from {connect_instance_address}")
+        get_cluster_name_commands = (
+            f"shell.connect('{self.cluster_admin_user}:{self.cluster_admin_password}@{connect_instance_address}')",
+            'cluster_name = session.run_sql("SELECT cluster_name FROM mysql_innodb_cluster_metadata.clusters;")',
+            "print(f'<CLUSTER_NAME>{cluster_name.fetch_one()[0]}</CLUSTER_NAME>')",
+        )
+
+        try:
+            output = self._run_mysqlsh_script("\n".join(get_cluster_name_commands))
+        except MySQLClientError as e:
+            logger.warning("Failed to get cluster name")
+            raise MySQLGetClusterNameError(e.message)
+
+        matches = re.search(r"<CLUSTER_NAME>(.+)</CLUSTER_NAME>", output)
+        if not matches:
+            return None
+
+        return matches.group(1)
+
     def get_primary_label(self) -> Optional[str]:
         """Get the label of the cluster's primary."""
         status = self.get_cluster_status()

poetry.lock

@@ -979,6 +979,23 @@ def test_get_cluster_endpoints(self, _, _is_cluster_replica):
             ),
         )
 
+    @patch("charms.mysql.v0.mysql.MySQLBase._run_mysqlsh_script")
+    def test_get_cluster_name(self, _run_mysqlsh_script):
+        """Test get_cluster_name() method."""
+        commands = "\n".join(
+            (
+                "shell.connect('clusteradmin:[email protected]')",
+                'cluster_name = session.run_sql("SELECT cluster_name FROM mysql_innodb_cluster_metadata.clusters;")',
+                "print(f'<CLUSTER_NAME>{cluster_name.fetch_one()[0]}</CLUSTER_NAME>')",
+            )
+        )
+
+        _run_mysqlsh_script.return_value = "<CLUSTER_NAME>test_cluster</CLUSTER_NAME>"
+
+        cluster_name = self.mysql.get_cluster_name("1.2.3.4")
+        self.assertEqual(cluster_name, "test_cluster")
+        _run_mysqlsh_script.assert_called_once_with(commands)
+
     @patch("charms.mysql.v0.mysql.MySQLBase._run_mysqlsh_script")
     def test_offline_mode_and_hidden_instance_exists(self, _run_mysqlsh_script):
         """Test the offline_mode_and_hidden_instance_exists() method."""