[MISC] Update MySQL charm libs (#621)

Author: sinclert-canonical

lib/charms/mysql/v0/async_replication.py

@@ -54,7 +54,7 @@
 # The unique Charmhub library identifier, never change it
 LIBID = "4de21f1a022c4e2c87ac8e672ec16f6a"
 LIBAPI = 0
-LIBPATCH = 6
+LIBPATCH = 7
 
 RELATION_OFFER = "replication-offer"
 RELATION_CONSUMER = "replication"
@@ -553,7 +553,7 @@ def _on_offer_relation_changed(self, event):
             unit_label = remote_data["node-label"]
 
             logger.debug("Looking for a donor node")
-            _, ro, _ = self._charm._mysql.get_cluster_endpoints(get_ips=False)
+            _, ro, _ = self._charm.get_cluster_endpoints(self.relation.name)
 
             if not ro:
                 logger.debug(f"Adding replica {cluster=} with {endpoint=}. Primary is the donor")
@@ -716,15 +716,16 @@ def _async_replication_credentials(self) -> dict[str, str]:
         secret = self._obtain_secret()
         return secret.peek_content()
 
-    def _get_endpoint(self) -> str:
+    def _get_endpoint(self) -> Optional[str]:
         """Get endpoint to be used by the primary cluster.
 
         This is the address in which the unit must be reachable from the primary cluster.
         Not necessarily the locally resolved address, but an ingress address.
         """
-        # TODO: devise method to inform the real address
-        # using unit informed address (fqdn or ip)
-        return self._charm.unit_address
+        if self.relation.name == RELATION_CONSUMER:
+            return self._charm.get_unit_address(self._charm.unit, RELATION_CONSUMER)
+        if self.relation.name == RELATION_OFFER:
+            return self._charm.get_unit_address(self._charm.unit, RELATION_OFFER)
 
     def _on_consumer_relation_created(self, event):
         """Handle the async_replica relation being created on the leader unit."""

lib/charms/mysql/v0/backups.py

@@ -95,6 +95,7 @@ def is_unit_blocked(self) -> bool:
 
 from constants import (
     MYSQL_DATA_DIR,
+    PEER,
     SERVER_CONFIG_PASSWORD_KEY,
     SERVER_CONFIG_USERNAME,
 )
@@ -112,21 +113,21 @@ def is_unit_blocked(self) -> bool:
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 13
+LIBPATCH = 14
 
 ANOTHER_S3_CLUSTER_REPOSITORY_ERROR_MESSAGE = "S3 repository claimed by another cluster"
 MOVE_RESTORED_CLUSTER_TO_ANOTHER_S3_REPOSITORY_ERROR = (
     "Move restored cluster to another S3 repository"
 )
 
 if typing.TYPE_CHECKING:
-    from charm import MySQLOperatorCharm
+    from mysql import MySQLCharmBase
 
 
 class MySQLBackups(Object):
     """Encapsulation of backups for MySQL."""
 
-    def __init__(self, charm: "MySQLOperatorCharm", s3_integrator: S3Requirer) -> None:
+    def __init__(self, charm: "MySQLCharmBase", s3_integrator: S3Requirer) -> None:
         super().__init__(charm, MYSQL_BACKUPS)
 
         self.charm = charm
@@ -715,7 +716,7 @@ def _pitr_restore(
         try:
             logger.info("Restoring point-in-time-recovery")
             stdout, stderr = self.charm._mysql.restore_pitr(
-                host=self.charm.get_unit_address(self.charm.unit),
+                host=self.charm.get_unit_address(self.charm.unit, PEER),
                 mysql_user=self.charm._mysql.server_config_user,
                 password=self.charm._mysql.server_config_password,
                 s3_parameters=s3_parameters,
@@ -744,8 +745,7 @@ def _post_restore(self) -> Tuple[bool, str]:
 
         try:
             logger.info("Creating cluster on restored node")
-            unit_label = self.charm.unit.name.replace("/", "-")
-            self.charm._mysql.create_cluster(unit_label)
+            self.charm._mysql.create_cluster(self.charm.unit_label)
             self.charm._mysql.create_cluster_set()
             self.charm._mysql.initialize_juju_units_operations_table()
 

lib/charms/mysql/v0/mysql.py

@@ -74,7 +74,6 @@ def wait_until_mysql_connection(self) -> None:
 import logging
 import os
 import re
-import socket
 import sys
 import time
 from abc import ABC, abstractmethod
@@ -134,7 +133,7 @@ def wait_until_mysql_connection(self) -> None:
 # Increment this major API version when introducing breaking changes
 LIBAPI = 0
 
-LIBPATCH = 87
+LIBPATCH = 88
 
 UNIT_TEARDOWN_LOCKNAME = "unit-teardown"
 UNIT_ADD_LOCKNAME = "unit-add"
@@ -495,11 +494,16 @@ def get_unit_hostname(self):
         raise NotImplementedError
 
     @abstractmethod
-    def get_unit_address(self, unit: Unit) -> str:
+    def get_unit_address(self, unit: Unit, relation_name: str) -> str:
         """Return unit address."""
         # each platform has its own way to get an arbitrary unit address
         raise NotImplementedError
 
+    @staticmethod
+    def get_unit_label(unit: Unit) -> str:
+        """Return unit label."""
+        return unit.name.replace("/", "-")
+
     def _on_get_password(self, event: ActionEvent) -> None:
         """Action used to retrieve the system user's password."""
         username = event.params.get("username") or ROOT_USERNAME
@@ -633,7 +637,7 @@ def cluster_initialized(self) -> bool:
         for unit in self.app_units:
             try:
                 if unit != self.unit and self._mysql.cluster_metadata_exists(
-                    self.get_unit_address(unit)
+                    self.get_unit_address(unit, PEER)
                 ):
                     return True
                 elif self._mysql.cluster_metadata_exists():
@@ -652,13 +656,14 @@ def only_one_cluster_node_thats_uninitialized(self) -> Optional[bool]:
         total_cluster_nodes = 0
         for unit in self.app_units:
             total_cluster_nodes += self._mysql.get_cluster_node_count(
-                from_instance=self.get_unit_address(unit)
+                from_instance=self.get_unit_address(unit, PEER)
             )
 
         total_online_cluster_nodes = 0
         for unit in self.app_units:
             total_online_cluster_nodes += self._mysql.get_cluster_node_count(
-                from_instance=self.get_unit_address(unit), node_status=MySQLMemberState["ONLINE"]
+                from_instance=self.get_unit_address(unit, PEER),
+                node_status=MySQLMemberState.ONLINE,
             )
 
         return total_cluster_nodes == 1 and total_online_cluster_nodes == 0
@@ -669,15 +674,16 @@ def cluster_fully_initialized(self) -> bool:
 
         Fully initialized means that all unit that can be joined are joined.
         """
-        return self._mysql.get_cluster_node_count(node_status=MySQLMemberState["ONLINE"]) == min(
+        return self._mysql.get_cluster_node_count(node_status=MySQLMemberState.ONLINE) == min(
             GR_MAX_MEMBERS, self.app.planned_units()
         )
 
     @property
     def unit_configured(self) -> bool:
         """Check if the unit is configured to be part of the cluster."""
         return self._mysql.is_instance_configured_for_innodb(
-            self.get_unit_address(self.unit), self.unit_label
+            self.get_unit_address(self.unit, PEER),
+            self.unit_label,
         )
 
     @property
@@ -707,7 +713,7 @@ def app_units(self) -> set[Unit]:
     @property
     def unit_label(self):
         """Return unit label."""
-        return self.unit.name.replace("/", "-")
+        return self.get_unit_label(self.unit)
 
     @property
     def _is_peer_data_set(self) -> bool:
@@ -774,6 +780,37 @@ def peer_relation_data(self, scope: Scopes) -> DataPeerData:
         elif scope == UNIT_SCOPE:
             return self.peer_relation_unit
 
+    def get_cluster_endpoints(self, relation_name: str) -> Tuple[str, str, str]:
+        """Return (rw, ro, offline) endpoints tuple names or IPs."""
+        repl_topology = self._mysql.get_cluster_topology()
+        repl_cluster = self._mysql.is_cluster_replica()
+
+        if not repl_topology:
+            raise MySQLGetClusterEndpointsError("Failed to get endpoints from cluster topology")
+
+        unit_labels = {self.get_unit_label(unit): unit for unit in self.app_units}
+
+        no_endpoints = set()
+        ro_endpoints = set()
+        rw_endpoints = set()
+
+        for k, v in repl_topology.items():
+            address = f"{self.get_unit_address(unit_labels[k], relation_name)}:3306"
+
+            if v["status"] != MySQLMemberState.ONLINE:
+                no_endpoints.add(address)
+            if v["status"] == MySQLMemberState.ONLINE and v["mode"] == "r/o":
+                ro_endpoints.add(address)
+            if v["status"] == MySQLMemberState.ONLINE and v["mode"] == "r/w" and not repl_cluster:
+                rw_endpoints.add(address)
+
+        # Replica return global primary address
+        if repl_cluster:
+            primary_address = f"{self._mysql.get_cluster_set_global_primary_address()}:3306"
+            rw_endpoints.add(primary_address)
+
+        return ",".join(rw_endpoints), ",".join(ro_endpoints), ",".join(no_endpoints)
+
     def get_secret(
         self,
         scope: Scopes,
@@ -2144,51 +2181,6 @@ def get_cluster_node_count(
 
         return int(matches.group(1)) if matches else 0
 
-    def get_cluster_endpoints(self, get_ips: bool = True) -> Tuple[str, str, str]:
-        """Return (rw, ro, ofline) endpoints tuple names or IPs."""
-        status = self.get_cluster_status()
-
-        if not status:
-            raise MySQLGetClusterEndpointsError("Failed to get endpoints from cluster status")
-
-        topology = status["defaultreplicaset"]["topology"]
-
-        def _get_host_ip(host: str) -> str:
-            try:
-                port = None
-                if ":" in host:
-                    host, port = host.split(":")
-
-                host_ip = socket.gethostbyname(host)
-                return f"{host_ip}:{port}" if port else host_ip
-            except socket.gaierror:
-                raise MySQLGetClusterEndpointsError(f"Failed to query IP for host {host}")
-
-        ro_endpoints = {
-            _get_host_ip(v["address"]) if get_ips else v["address"]
-            for v in topology.values()
-            if v["mode"] == "r/o" and v["status"] == MySQLMemberState.ONLINE
-        }
-
-        if self.is_cluster_replica():
-            # replica return global primary address
-            global_primary = self.get_cluster_set_global_primary_address()
-            if not global_primary:
-                raise MySQLGetClusterEndpointsError("Failed to get global primary address")
-            rw_endpoints = {_get_host_ip(global_primary) if get_ips else global_primary}
-        else:
-            rw_endpoints = {
-                _get_host_ip(v["address"]) if get_ips else v["address"]
-                for v in topology.values()
-                if v["mode"] == "r/w" and v["status"] == MySQLMemberState.ONLINE
-            }
-        # won't get offline endpoints to IP as they maybe unreachable
-        no_endpoints = {
-            v["address"] for v in topology.values() if v["status"] != MySQLMemberState.ONLINE
-        }
-
-        return ",".join(rw_endpoints), ",".join(ro_endpoints), ",".join(no_endpoints)
-
     def execute_remove_instance(
         self, connect_instance: Optional[str] = None, force: bool = False
     ) -> None:
@@ -2478,12 +2470,21 @@ def get_cluster_set_global_primary_address(
 
         return address
 
-    def get_primary_label(self) -> Optional[str]:
-        """Get the label of the cluster's primary."""
+    def get_cluster_topology(self) -> Optional[dict]:
+        """Get the cluster topology."""
         status = self.get_cluster_status()
         if not status:
             return None
-        for label, value in status["defaultreplicaset"]["topology"].items():
+
+        return status["defaultreplicaset"]["topology"]
+
+    def get_primary_label(self) -> Optional[str]:
+        """Get the label of the cluster's primary."""
+        topology = self.get_cluster_topology()
+        if not topology:
+            return None
+
+        for label, value in topology.items():
             if value["memberrole"] == "primary":
                 return label
 

src/charm.py

@@ -210,7 +210,7 @@ def tracing_endpoint(self) -> Optional[str]:
     def _mysql(self) -> MySQL:
         """Returns an instance of the MySQL object from mysql_k8s_helpers."""
         return MySQL(
-            self.get_unit_address(),
+            self.unit_address,
             self.app_peer_data["cluster-name"],
             self.app_peer_data["cluster-set-domain-name"],
             self.get_secret("app", ROOT_PASSWORD_KEY),  # pyright: ignore [reportArgumentType]
@@ -292,7 +292,7 @@ def restart_peers(self) -> Optional[ops.model.Relation]:
     @property
     def unit_address(self) -> str:
         """Return the address of this unit."""
-        return self.get_unit_address()
+        return self.get_unit_address(self.unit)
 
     @property
     def is_new_unit(self) -> bool:
@@ -344,14 +344,11 @@ def get_unit_hostname(self, unit_name: Optional[str] = None) -> str:
         return f"{unit_name.replace('/', '-')}.{self.app.name}-endpoints"
 
     @retry(reraise=True, stop=stop_after_delay(120), wait=wait_fixed(2))
-    def get_unit_address(self, unit: Optional[Unit] = None) -> str:
+    def get_unit_address(self, unit: Unit, relation_name: str = PEER) -> str:
         """Get fqdn/address for a unit.
 
         Translate juju unit name to resolvable hostname.
         """
-        if not unit:
-            unit = self.unit
-
         unit_hostname = self.get_unit_hostname(unit.name)
         unit_dns_domain = getfqdn(self.get_unit_hostname(unit.name))
 
@@ -411,7 +408,7 @@ def join_unit_to_cluster(self) -> None:
 
         Try to join the unit from the primary unit.
         """
-        instance_label = self.unit.name.replace("/", "-")
+        instance_label = self.get_unit_label(self.unit)
         instance_address = self.get_unit_address(self.unit)
 
         if not self._mysql.is_instance_in_cluster(instance_label):

src/k8s_helpers.py

@@ -41,7 +41,7 @@ def __init__(self, charm: "MySQLOperatorCharm"):
         Args:
             charm: a `CharmBase` parent object
         """
-        self.pod_name = charm.unit.name.replace("/", "-")
+        self.pod_name = charm.get_unit_label(charm.unit)
         self.namespace = charm.model.name
         self.app_name = charm.model.app.name
         self.cluster_name = charm.app_peer_data.get("cluster-name")

src/mysql_k8s_helpers.py

@@ -47,6 +47,7 @@
     MYSQLD_SERVICE,
     MYSQLD_SOCK_FILE,
     MYSQLSH_LOCATION,
+    PEER,
     ROOT_SYSTEM_USER,
     XTRABACKUP_PLUGIN_DIR,
 )
@@ -898,11 +899,11 @@ def is_data_dir_initialised(self) -> bool:
         except ExecError:
             return False
 
-    def update_endpoints(self) -> None:
+    def update_endpoints(self, relation_name: str) -> None:
         """Updates pod labels to reflect role of the unit."""
         logger.debug("Updating pod labels")
         try:
-            rw_endpoints, ro_endpoints, offline = self.get_cluster_endpoints(get_ips=False)
+            rw_endpoints, ro_endpoints, offline = self.charm.get_cluster_endpoints(relation_name)
 
             for endpoints, label in (
                 (rw_endpoints, "primary"),
@@ -920,7 +921,7 @@ def update_endpoints(self) -> None:
     def set_cluster_primary(self, new_primary_address: str) -> None:
         """Set the cluster primary and update pod labels."""
         super().set_cluster_primary(new_primary_address)
-        self.update_endpoints()
+        self.update_endpoints(PEER)
 
     def fetch_error_log(self) -> Optional[str]:
         """Fetch the MySQL error log."""
@@ -983,6 +984,6 @@ def get_cluster_members(self) -> list[str]:
 
         Returns: list of cluster members in MySQL MEMBER_HOST format.
         """
-        return [self.charm.get_unit_address(self.charm.unit)] + [
+        return [self.charm.unit_address] + [
             self.charm.get_unit_address(unit) for unit in self.charm.peers.units
         ]