[MISC] Converge cluster status calls (#1014)

* Switch most calls to the parallel cluster call

* Remove dead code

* Use cached get primary

* Revert cached primary call

Author: dragomirp

src/cluster.py

@@ -224,17 +224,10 @@ def _change_owner(self, path: str) -> None:
         os.chown(path, uid=user_database.pw_uid, gid=user_database.pw_gid)
 
     @property
-    @retry(stop=stop_after_attempt(10), wait=wait_exponential(multiplier=1, min=2, max=10))
     def cluster_members(self) -> set:
         """Get the current cluster members."""
         # Request info from cluster endpoint (which returns all members of the cluster).
-        cluster_status = requests.get(
-            f"{self._patroni_url}/{PATRONI_CLUSTER_STATUS_ENDPOINT}",
-            verify=self.verify,
-            timeout=API_REQUEST_TIMEOUT,
-            auth=self._patroni_auth,
-        )
-        return {member["name"] for member in cluster_status.json()["members"]}
+        return {member["name"] for member in self.cached_cluster_status}
 
     def _create_directory(self, path: str, mode: int) -> None:
         """Creates a directory.
@@ -256,6 +249,11 @@ def get_postgresql_version(self) -> str:
             if snp["name"] == charm_refresh.snap_name():
                 return snp["version"]
 
+    @property
+    def cached_cluster_status(self):
+        """Cached cluster status."""
+        return self.cluster_status()
+
     def cluster_status(self, alternative_endpoints: list | None = None) -> list[ClusterMember]:
         """Query the cluster status."""
         # Request info from cluster endpoint (which returns all members of the cluster).
@@ -411,27 +409,22 @@ def are_all_members_ready(self) -> bool:
         # Request info from cluster endpoint
         # (which returns all members of the cluster and their states).
         try:
-            for attempt in Retrying(stop=stop_after_delay(10), wait=wait_fixed(3)):
-                with attempt:
-                    cluster_status = requests.get(
-                        f"{self._patroni_url}/{PATRONI_CLUSTER_STATUS_ENDPOINT}",
-                        verify=self.verify,
-                        timeout=API_REQUEST_TIMEOUT,
-                        auth=self._patroni_auth,
-                    )
+            members = self.cluster_status()
         except RetryError:
             return False
 
         # Check if all members are running and one of them is a leader (primary) or
         # a standby leader, because sometimes there may exist (for some period of time)
         # only replicas after a failed switchover.
-        return all(
-            member["state"] in STARTED_STATES for member in cluster_status.json()["members"]
-        ) and any(
-            member["role"] in ["leader", "standby_leader"]
-            for member in cluster_status.json()["members"]
+        return all(member["state"] in STARTED_STATES for member in members) and any(
+            member["role"] in ["leader", "standby_leader"] for member in members
         )
 
+    @property
+    def cached_patroni_health(self) -> dict[str, str]:
+        """Cached local unit health."""
+        return self.get_patroni_health()
+
     def get_patroni_health(self) -> dict[str, str]:
         """Gets, retires and parses the Patroni health endpoint."""
         for attempt in Retrying(stop=stop_after_delay(60), wait=wait_fixed(7)):
@@ -452,20 +445,12 @@ def is_creating_backup(self) -> bool:
         # cluster member; the "is_creating_backup" tag means that the member is creating
         # a backup).
         try:
-            for attempt in Retrying(stop=stop_after_delay(10), wait=wait_fixed(3)):
-                with attempt:
-                    r = requests.get(
-                        f"{self._patroni_url}/cluster",
-                        verify=self.verify,
-                        auth=self._patroni_auth,
-                        timeout=PATRONI_TIMEOUT,
-                    )
+            members = self.cached_cluster_status
         except RetryError:
             return False
 
         return any(
-            "tags" in member and member["tags"].get("is_creating_backup")
-            for member in r.json()["members"]
+            "tags" in member and member["tags"].get("is_creating_backup") for member in members
         )
 
     def is_replication_healthy(self) -> bool:
@@ -509,7 +494,7 @@ def member_started(self) -> bool:
         if not self.is_patroni_running():
             return False
         try:
-            response = self.get_patroni_health()
+            response = self.cached_patroni_health
         except RetryError:
             return False
 
@@ -524,7 +509,7 @@ def member_inactive(self) -> bool:
             seconds times to allow server time to start up.
         """
         try:
-            response = self.get_patroni_health()
+            response = self.cached_patroni_health
         except RetryError:
             return True
 
@@ -535,27 +520,6 @@ def member_inactive(self) -> bool:
             "restarting",
         ]
 
-    @property
-    def member_replication_lag(self) -> str:
-        """Member replication lag."""
-        try:
-            for attempt in Retrying(stop=stop_after_delay(60), wait=wait_fixed(3)):
-                with attempt:
-                    cluster_status = requests.get(
-                        f"{self._patroni_url}/{PATRONI_CLUSTER_STATUS_ENDPOINT}",
-                        verify=self.verify,
-                        timeout=API_REQUEST_TIMEOUT,
-                        auth=self._patroni_auth,
-                    )
-        except RetryError:
-            return "unknown"
-
-        for member in cluster_status.json()["members"]:
-            if member["name"] == self.member_name:
-                return member.get("lag", "unknown")
-
-        return "unknown"
-
     @property
     def is_member_isolated(self) -> bool:
         """Returns whether the unit is isolated from the cluster."""
@@ -589,16 +553,8 @@ def online_cluster_members(self) -> list[ClusterMember]:
     def are_replicas_up(self) -> dict[str, bool] | None:
         """Check if cluster members are running or streaming."""
         try:
-            response = requests.get(
-                f"{self._patroni_url}/cluster",
-                verify=self.verify,
-                auth=self._patroni_auth,
-                timeout=PATRONI_TIMEOUT,
-            )
-            return {
-                member["host"]: member["state"] in ["running", "streaming"]
-                for member in response.json()["members"]
-            }
+            members = self.cluster_status()
+            return {member["host"]: member["state"] in STARTED_STATES for member in members}
         except Exception:
             logger.exception("Unable to get the state of the cluster")
             return
@@ -909,15 +865,8 @@ def reinitialise_raft_data(self) -> None:
     def get_running_cluster_members(self) -> list[str]:
         """List running patroni members."""
         try:
-            members = requests.get(
-                f"{self._patroni_url}/{PATRONI_CLUSTER_STATUS_ENDPOINT}",
-                verify=self.verify,
-                timeout=API_REQUEST_TIMEOUT,
-                auth=self._patroni_auth,
-            ).json()["members"]
-            return [
-                member["name"] for member in members if member["state"] in ("streaming", "running")
-            ]
+            members = self.cluster_status()
+            return [member["name"] for member in members if member["state"] in STARTED_STATES]
         except Exception:
             return []
 

src/cluster_topology_observer.py

@@ -121,9 +121,3 @@ def stop_observer(self):
             self._charm._peers.data[self._charm.unit].update({"observer-pid": ""})
         except OSError:
             pass
-
-    @property
-    def unit_tag(self):
-        """Juju-style tag identifying the unit being run by this charm."""
-        unit_num = self._charm.unit.name.split("/")[-1]
-        return f"unit-{self._charm.app.name}-{unit_num}"

tests/unit/test_charm.py

@@ -803,9 +803,6 @@ def test_on_update_status(harness):
         ) as _set_primary_status_message,
         patch("charm.Patroni.restart_patroni") as _restart_patroni,
         patch("charm.Patroni.is_member_isolated") as _is_member_isolated,
-        patch(
-            "charm.Patroni.member_replication_lag", new_callable=PropertyMock
-        ) as _member_replication_lag,
         patch("charm.Patroni.member_started", new_callable=PropertyMock) as _member_started,
         patch(
             "charm.PostgresqlOperatorCharm.is_standby_leader", new_callable=PropertyMock
@@ -1383,9 +1380,6 @@ def test_on_peer_relation_changed(harness):
             "backups.PostgreSQLBackups.start_stop_pgbackrest_service"
         ) as _start_stop_pgbackrest_service,
         patch("charm.Patroni.reinitialize_postgresql") as _reinitialize_postgresql,
-        patch(
-            "charm.Patroni.member_replication_lag", new_callable=PropertyMock
-        ) as _member_replication_lag,
         patch("charm.PostgresqlOperatorCharm.is_standby_leader") as _is_standby_leader,
         patch("charm.PostgresqlOperatorCharm.is_primary") as _is_primary,
         patch("charm.Patroni.member_started", new_callable=PropertyMock) as _member_started,
@@ -1466,15 +1460,14 @@ def test_on_peer_relation_changed(harness):
         # huge or unknown lag.
         relation = harness.model.get_relation(PEER, rel_id)
         _member_started.return_value = True
-        for values in itertools.product([True, False], ["0", "1000", "1001", "unknown"]):
+        for values in [True, False]:
             _defer.reset_mock()
             _start_stop_pgbackrest_service.reset_mock()
-            _is_primary.return_value = values[0]
-            _is_standby_leader.return_value = values[0]
-            _member_replication_lag.return_value = values[1]
+            _is_primary.return_value = values
+            _is_standby_leader.return_value = values
             harness.charm.unit.status = ActiveStatus()
             harness.charm.on.database_peers_relation_changed.emit(relation)
-            if _is_primary.return_value == values[0] or int(values[1]) <= 1000:
+            if _is_primary.return_value == values:
                 _defer.assert_not_called()
                 _start_stop_pgbackrest_service.assert_called_once()
                 assert isinstance(harness.charm.unit.status, ActiveStatus)

tests/unit/test_cluster.py

@@ -12,7 +12,6 @@
 from pysyncobj.utility import UtilityException
 from tenacity import (
     RetryError,
-    Retrying,
     stop_after_delay,
     wait_fixed,
 )
@@ -92,11 +91,9 @@ def patroni(harness, peers_ips):
 
 
 def test_get_member_ip(peers_ips, patroni):
-    with (
-        patch(
-            "charm.Patroni.parallel_patroni_get_request", return_value=None
-        ) as _parallel_patroni_get_request,
-    ):
+    with patch(
+        "charm.Patroni.parallel_patroni_get_request", return_value=None
+    ) as _parallel_patroni_get_request:
         # No IP if no members
         assert patroni.get_member_ip(patroni.member_name) is None
 
@@ -201,21 +198,16 @@ def test_get_primary(peers_ips, patroni):
 
 
 def test_is_creating_backup(peers_ips, patroni):
-    with patch("requests.get") as _get:
+    with patch("charm.Patroni.cluster_status") as _cluster_status:
         # Test when one member is creating a backup.
-        response = _get.return_value
-        response.json.return_value = {
-            "members": [
-                {"name": "postgresql-0"},
-                {"name": "postgresql-1", "tags": {"is_creating_backup": True}},
-            ]
-        }
+        _cluster_status.return_value = [
+            {"name": "postgresql-0"},
+            {"name": "postgresql-1", "tags": {"is_creating_backup": True}},
+        ]
         assert patroni.is_creating_backup
 
         # Test when no member is creating a backup.
-        response.json.return_value = {
-            "members": [{"name": "postgresql-0"}, {"name": "postgresql-1"}]
-        }
+        _cluster_status.return_value = [{"name": "postgresql-0"}, {"name": "postgresql-1"}]
         assert not patroni.is_creating_backup
 
 
@@ -411,28 +403,6 @@ def test_stop_patroni(peers_ips, patroni):
         assert not patroni.stop_patroni()
 
 
-def test_member_replication_lag(peers_ips, patroni):
-    with (
-        patch("requests.get", side_effect=mocked_requests_get) as _get,
-        patch("charm.Patroni._patroni_url", new_callable=PropertyMock) as _patroni_url,
-    ):
-        # Test when the cluster member has a value for the lag field.
-        _patroni_url.return_value = "http://server1"
-        lag = patroni.member_replication_lag
-        assert lag == "1"
-
-        # Test when the cluster member doesn't have a value for the lag field.
-        patroni.member_name = "postgresql-1"
-        lag = patroni.member_replication_lag
-        assert lag == "unknown"
-
-        # Test when the API call fails.
-        _patroni_url.return_value = "http://server2"
-        with patch.object(Retrying, "iter", Mock(side_effect=RetryError(None))):
-            lag = patroni.member_replication_lag
-            assert lag == "unknown"
-
-
 def test_reinitialize_postgresql(peers_ips, patroni):
     with patch("requests.post") as _post:
         patroni.reinitialize_postgresql()
@@ -888,18 +858,14 @@ def test_reinitialise_raft_data(patroni):
 
 
 def test_are_replicas_up(patroni):
-    with (
-        patch("requests.get") as _get,
-    ):
-        _get.return_value.json.return_value = {
-            "members": [
-                {"host": "1.1.1.1", "state": "running"},
-                {"host": "2.2.2.2", "state": "streaming"},
-                {"host": "3.3.3.3", "state": "other state"},
-            ]
-        }
+    with patch("charm.Patroni.cluster_status") as _cluster_status:
+        _cluster_status.return_value = [
+            {"host": "1.1.1.1", "state": "running"},
+            {"host": "2.2.2.2", "state": "streaming"},
+            {"host": "3.3.3.3", "state": "other state"},
+        ]
         assert patroni.are_replicas_up() == {"1.1.1.1": True, "2.2.2.2": True, "3.3.3.3": False}
 
         # Return None on error
-        _get.side_effect = Exception
+        _cluster_status.side_effect = Exception
         assert patroni.are_replicas_up() is None