Repository: zalando/patroni
Pull request #2610 (2023-03-21 12:46:32)

Add docstrings and type annotations to `patroni/daemon.py`
References: PAT-38

barthisrael: Thanks for the review, all changes applied!
coveralls: ## Pull Request Test Coverage Report for [Build 4482002007](https://coveralls.io/builds/58107619)

* **18** of **18**   **(100.0%)**  changed or added relevant lines in **1** file are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+6.0e**-05%**) to **99.851%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/58107619/badge)](https://coveralls.io/builds/58107619) |
| :-- | --: |
| Change from base [Build 4417069895](https://coveralls.io/builds/57844528): |  6.0e-05% |
| Covered Lines: | 12092 |
| Relevant Lines: | 12110 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/docs/rest_api.rst b/docs/rest_api.rst
index be8249fb..e5323b82 100644
--- a/docs/rest_api.rst
+++ b/docs/rest_api.rst
@@ -139,6 +139,9 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e
 	# HELP patroni_replica Value is 1 if this node is a replica, 0 otherwise.
 	# TYPE patroni_replica gauge
 	patroni_replica{scope="batman"} 0
+	# HELP patroni_sync_standby Value is 1 if this node is a sync standby replica, 0 otherwise.
+        # TYPE patroni_sync_standby gauge
+        patroni_sync_standby{scope="batman"} 0
 	# HELP patroni_xlog_received_location Current location of the received Postgres transaction log, 0 if this node is not a replica.
 	# TYPE patroni_xlog_received_location counter
 	patroni_xlog_received_location{scope="batman"} 0
diff --git a/patroni/api.py b/patroni/api.py
index 7a049bae..ebbdee9a 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -269,6 +269,10 @@ class RestApiHandler(BaseHTTPRequestHandler):
         metrics.append("# TYPE patroni_replica gauge")
         metrics.append("patroni_replica{0} {1}".format(scope_label, int(postgres['role'] == 'replica')))
 
+        metrics.append("# HELP patroni_sync_standby Value is 1 if this node is a sync standby replica, 0 otherwise.")
+        metrics.append("# TYPE patroni_sync_standby gauge")
+        metrics.append("patroni_sync_standby{0} {1}".format(scope_label, int(postgres.get('sync_standby', False))))
+
         metrics.append("# HELP patroni_xlog_received_location Current location of the received"
                        " Postgres transaction log, 0 if this node is not a replica.")
         metrics.append("# TYPE patroni_xlog_received_location counter")
@@ -686,6 +690,10 @@ class RestApiHandler(BaseHTTPRequestHandler):
             if result['role'] == 'replica' and self.server.patroni.ha.is_standby_cluster():
                 result['role'] = postgresql.role
 
+            if result['role'] == 'replica' and cluster and cluster.is_synchronous_mode()\
+                    and cluster.sync and postgresql.name in cluster.sync.members:
+                result['sync_standby'] = True
+
             if row[1] > 0:
                 result['timeline'] = row[1]
             else:
diff --git a/patroni/daemon.py b/patroni/daemon.py
index 6bac504b..404e4061 100644
--- a/patroni/daemon.py
+++ b/patroni/daemon.py
@@ -1,3 +1,8 @@
+"""Daemon processes abstraction module.
+
+This module implements abstraction classes and functions for creating and managing daemon processes in Patroni.
+Currently it is only used for the main "Thread" of ``patroni`` and ``patroni_raft_controller`` commands.
+"""
 from __future__ import print_function
 
 import abc
@@ -6,11 +11,30 @@ import signal
 import sys
 
 from threading import Lock
+from typing import Any, Optional, Type
+
+from .config import Config
+from .validator import Schema
 
 
 class AbstractPatroniDaemon(abc.ABC):
+    """A Patroni daemon process.
+
+    .. note::
+
+        When inheriting from :class:`AbstractPatroniDaemon` you are expected to define the methods :func:`_run_cycle`
+        to determine what it should do in each execution cycle, and :func:`_shutdown` to determine what it should do
+        when shutting down.
+
+    :ivar logger: log handler used by this daemon.
+    :ivar config: configuration options for this daemon.
+    """
 
-    def __init__(self, config):
+    def __init__(self, config: Config) -> None:
+        """Set up signal handlers, logging handler and configuration.
+
+        :param config: configuration options for this daemon.
+        """
         from patroni.log import PatroniLogger
 
         self.setup_signal_handlers()
@@ -19,20 +43,44 @@ class AbstractPatroniDaemon(abc.ABC):
         self.config = config
         AbstractPatroniDaemon.reload_config(self, local=True)
 
-    def sighup_handler(self, *args):
+    def sighup_handler(self, *_: Any) -> None:
+        """Handle SIGHUP signals.
+
+        Flag the daemon as "SIGHUP received".
+        """
         self._received_sighup = True
 
-    def api_sigterm(self):
+    def api_sigterm(self) -> bool:
+        """Guarantee only a single SIGTERM is being processed.
+
+        Flag the daemon as "SIGTERM received" with a lock-based approach.
+
+        :returns: ``True`` if the daemon was flagged as "SIGTERM received".
+        """
+        ret = False
         with self._sigterm_lock:
             if not self._received_sigterm:
                 self._received_sigterm = True
-                return True
+                ret = True
+        return ret
+
+    def sigterm_handler(self, *_: Any) -> None:
+        """Handle SIGTERM signals.
 
-    def sigterm_handler(self, *args):
+        Terminate the daemon process through :func:`api_sigterm`.
+        """
         if self.api_sigterm():
             sys.exit()
 
-    def setup_signal_handlers(self):
+    def setup_signal_handlers(self) -> None:
+        """Set up daemon signal handlers.
+
+        Set up SIGHUP and SIGTERM signal handlers.
+
+        .. note::
+
+            SIGHUP is only handled in non-Windows environments.
+        """
         self._received_sighup = False
         self._sigterm_lock = Lock()
         self._received_sigterm = False
@@ -41,19 +89,34 @@ class AbstractPatroniDaemon(abc.ABC):
         signal.signal(signal.SIGTERM, self.sigterm_handler)
 
     @property
-    def received_sigterm(self):
+    def received_sigterm(self) -> bool:
+        """If daemon was signaled with SIGTERM."""
         with self._sigterm_lock:
             return self._received_sigterm
 
-    def reload_config(self, sighup=False, local=False):
+    def reload_config(self, sighup: Optional[bool] = False, local: Optional[bool] = False) -> None:
+        """Reload configuration.
+
+        :param sighup: if it is related to a SIGHUP signal.
+                       The sighup parameter could be used in the method overridden in a child class.
+        :param local: will be ``True`` if there are changes in the local configuration file.
+        """
         if local:
             self.logger.reload_config(self.config.get('log', {}))
 
     @abc.abstractmethod
-    def _run_cycle(self):
-        """_run_cycle"""
+    def _run_cycle(self) -> None:
+        """Define what the daemon should do in each execution cycle.
+
+        Keep being called in the daemon's main loop until the daemon is eventually terminated.
+        """
 
-    def run(self):
+    def run(self) -> None:
+        """Run the daemon process.
+
+        Start the logger thread and keep running execution cycles until a SIGTERM is eventually received. Also reload
+        configuration uppon receiving SIGHUP.
+        """
         self.logger.start()
         while not self.received_sigterm:
             if self._received_sighup:
@@ -63,17 +126,29 @@ class AbstractPatroniDaemon(abc.ABC):
             self._run_cycle()
 
     @abc.abstractmethod
-    def _shutdown(self):
-        """_shutdown"""
+    def _shutdown(self) -> None:
+        """Define what the daemon should do when shutting down."""
+
+    def shutdown(self) -> None:
+        """Shut the daemon down when a SIGTERM is received.
 
-    def shutdown(self):
+        Shut down the daemon process and the logger thread.
+        """
         with self._sigterm_lock:
             self._received_sigterm = True
         self._shutdown()
         self.logger.shutdown()
 
 
-def abstract_main(cls, validator=None):
+def abstract_main(cls: Type[AbstractPatroniDaemon], validator: Optional[Schema] = None) -> None:
+    """Create the main entry point of a given daemon process.
+
+    Expose a basic argument parser, parse the command-line arguments, and run the given daemon process.
+
+    :param cls: a class that should inherit from :class:`AbstractPatroniDaemon`.
+    :param validator: used to validate the daemon configuration schema, if requested by the user through
+        ``--validate-config`` CLI option.
+    """
     import argparse
 
     from .config import Config, ConfigParseError
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 09d4be66..4ae26150 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -14,6 +14,7 @@ from collections import defaultdict, namedtuple
 from copy import deepcopy
 from random import randint
 from threading import Event, Lock
+from typing import Any, Dict, List, Optional, Union
 from urllib.parse import urlparse, urlunparse, parse_qsl
 
 from ..exceptions import PatroniFatalException
@@ -374,7 +375,7 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')):
     """
 
     @staticmethod
-    def from_node(index, value):
+    def from_node(index: Union[str, int], value: Union[str, Dict[str, Any]]) -> 'SyncState':
         """
         >>> SyncState.from_node(1, None).leader is None
         True
@@ -389,27 +390,31 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')):
         >>> SyncState.from_node(1, {"leader": "leader"}).leader == "leader"
         True
         """
-        if isinstance(value, dict):
-            data = value
-        elif value:
-            try:
-                data = json.loads(value)
-                if not isinstance(data, dict):
-                    data = {}
-            except (TypeError, ValueError):
-                data = {}
-        else:
-            data = {}
-        return SyncState(index, data.get('leader'), data.get('sync_standby'))
+        try:
+            if value and isinstance(value, str):
+                value = json.loads(value)
+            if not isinstance(value, dict):
+                return SyncState.empty(index)
+            return SyncState(index, value.get('leader'), value.get('sync_standby'))
+        except (TypeError, ValueError):
+            return SyncState.empty(index)
+
+    @staticmethod
+    def empty(index: Optional[Union[str, int]] = '') -> 'SyncState':
+        return SyncState(index, None, '')
 
     @property
-    def members(self):
-        """ Returns sync_standby in list """
-        return self.sync_standby and self.sync_standby.split(',') or []
+    def is_empty(self) -> bool:
+        """:returns: True if /sync key doesn't have a leader"""
+        return self.leader is None
 
-    def matches(self, name):
-        """
-        Returns if a node name matches one of the nodes in the sync state
+    @property
+    def members(self) -> List[str]:
+        """:returns: sync_standby as list"""
+        return list(filter(lambda a: a, [s.strip() for s in self.sync_standby.split(',')])) if self.sync_standby else []
+
+    def matches(self, name: str) -> bool:
+        """:returns: True if a node name matches one of the nodes in the sync state (including leader)
 
         >>> s = SyncState(1, 'foo', 'bar,zoo')
         >>> s.matches('foo')
@@ -471,6 +476,10 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,'
             args = args + ({},)
         return super(Cluster, cls).__new__(cls, *args)
 
+    @staticmethod
+    def empty():
+        return Cluster(None, None, None, 0, [], None, SyncState.empty(), None, None, None)
+
     @property
     def leader_name(self):
         return self.leader and self.leader.name
@@ -812,8 +821,8 @@ class AbstractDCS(abc.ABC):
         if isinstance(groups, Cluster):  # Zookeeper could return a cached version
             cluster = groups
         else:
-            cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID,
-                                 Cluster(None, None, None, None, [], None, None, None, None, None))
+            assert isinstance(groups, dict)
+            cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty())
             cluster.workers.update(groups)
         return cluster
 
diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index 5404c961..5c988833 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -409,7 +409,7 @@ class Consul(AbstractDCS):
         try:
             return loader(path)
         except NotFound:
-            return Cluster(None, None, None, None, [], None, None, None, None, None)
+            return Cluster.empty()
         except Exception:
             logger.exception('get_cluster')
             raise ConsulError('Consul is not responding properly')
diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py
index ca862df9..154a7de0 100644
--- a/patroni/dcs/etcd.py
+++ b/patroni/dcs/etcd.py
@@ -704,7 +704,7 @@ class Etcd(AbstractEtcd):
         try:
             cluster = loader(path)
         except etcd.EtcdKeyNotFound:
-            cluster = Cluster(None, None, None, None, [], None, None, None, None, None)
+            cluster = Cluster.empty()
         except Exception as e:
             self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly'))
         self._has_failed = False
diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py
index 0d0e4568..e09b3290 100644
--- a/patroni/dcs/raft.py
+++ b/patroni/dcs/raft.py
@@ -383,7 +383,7 @@ class Raft(AbstractDCS):
     def _cluster_loader(self, path):
         response = self._sync_obj.get(path, recursive=True)
         if not response:
-            return Cluster(None, None, None, None, [], None, None, None, None, None)
+            return Cluster.empty()
         nodes = {key[len(path):]: value for key, value in response.items()}
         return self._cluster_from_nodes(nodes)
 
diff --git a/patroni/ha.py b/patroni/ha.py
index 6eda0a3e..53f6d419 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -590,7 +590,7 @@ class Ha(object):
         """
         if self.is_synchronous_mode():
             sync_node_count = self.patroni.config['synchronous_node_count']
-            current = self.cluster.sync.leader and self.cluster.sync.members or []
+            current = [] if self.cluster.sync.is_empty else self.cluster.sync.members
             picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, sync_node_count,
                                                                                   self.patroni.config[
                                                                                       'maximum_lag_on_syncnode'])
@@ -625,7 +625,7 @@ class Ha(object):
                         cluster = self.dcs.get_cluster()
                     except DCSError:
                         return logger.warning("Could not get cluster state from DCS during process_sync_replication()")
-                    if cluster.sync.leader and cluster.sync.leader != self.state_handler.name:
+                    if not cluster.sync.is_empty and cluster.sync.leader != self.state_handler.name:
                         logger.info("Synchronous replication key updated by someone else")
                         return
                     if not self.dcs.write_sync_state(self.state_handler.name, allow_promote, index=cluster.sync.index):
@@ -633,7 +633,7 @@ class Ha(object):
                         return
                     logger.info("Synchronous standby status assigned to %s", allow_promote)
         else:
-            if self.cluster.sync.leader and self.dcs.delete_sync_state(index=self.cluster.sync.index):
+            if not self.cluster.sync.is_empty and self.dcs.delete_sync_state(index=self.cluster.sync.index):
                 logger.info("Disabled synchronous replication")
             self.state_handler.sync_handler.set_synchronous_standby_names([])
 
@@ -994,7 +994,7 @@ class Ha(object):
         all_known_members += self.cluster.members
 
         # When in sync mode, only last known primary and sync standby are allowed to promote automatically.
-        if self.is_synchronous_mode() and self.cluster.sync and self.cluster.sync.leader:
+        if self.is_synchronous_mode() and not self.cluster.sync.is_empty:
             if not self.cluster.sync.matches(self.state_handler.name):
                 return False
             # pick between synchronous candidates so we minimize unnecessary failovers/demotions
diff --git a/patroni/version.py b/patroni/version.py
index b7a55314..7962f44c 100644
--- a/patroni/version.py
+++ b/patroni/version.py
@@ -1,1 +1,5 @@
+"""This module specifies the current Patroni version.
+
+:var __version__: the current Patroni version.
+"""
 __version__ = '3.0.1'


Test patch:
diff --git a/tests/test_api.py b/tests/test_api.py
index ab2e8f04..43e185f9 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -184,6 +184,7 @@ class TestRestApiHandler(unittest.TestCase):
 
     def test_do_GET(self):
         MockPatroni.dcs.cluster.last_lsn = 20
+        MockPatroni.dcs.cluster.sync.members = [MockPostgresql.name]
         MockRestApiServer(RestApiHandler, 'GET /replica')
         MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M')
         MockRestApiServer(RestApiHandler, 'GET /replica?lag=10MB')
@@ -196,7 +197,6 @@ class TestRestApiHandler(unittest.TestCase):
         with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'state': 'running'})):
             MockRestApiServer(RestApiHandler, 'GET /health')
         MockRestApiServer(RestApiHandler, 'GET /leader')
-        MockPatroni.dcs.cluster.sync.members = [MockPostgresql.name]
         MockPatroni.dcs.cluster.is_synchronous_mode = Mock(return_value=True)
         with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})):
             MockRestApiServer(RestApiHandler, 'GET /synchronous')
diff --git a/tests/test_ha.py b/tests/test_ha.py
index d30689bf..70e465c6 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -42,11 +42,11 @@ def get_cluster(initialize, leader, members, failover, sync, cluster_config=None
 
 
 def get_cluster_not_initialized_without_leader(cluster_config=None):
-    return get_cluster(None, None, [], None, SyncState(None, None, None), cluster_config)
+    return get_cluster(None, None, [], None, SyncState.empty(), cluster_config)
 
 
 def get_cluster_bootstrapping_without_leader(cluster_config=None):
-    return get_cluster("", None, [], None, SyncState(None, None, None), cluster_config)
+    return get_cluster("", None, [], None, SyncState.empty(), cluster_config)
 
 
 def get_cluster_initialized_without_leader(leader=False, failover=None, sync=None, cluster_config=None, failsafe=False):
@@ -72,7 +72,7 @@ def get_cluster_initialized_with_leader(failover=None, sync=None):
 
 def get_cluster_initialized_with_only_leader(failover=None, cluster_config=None):
     leader = get_cluster_initialized_without_leader(leader=True, failover=failover).leader
-    return get_cluster(True, leader, [leader.member], failover, None, cluster_config)
+    return get_cluster(True, leader, [leader.member], failover, SyncState.empty(), cluster_config)
 
 
 def get_standby_cluster_initialized_with_only_leader(failover=None, sync=None):
diff --git a/tests/test_slots.py b/tests/test_slots.py
index 95e55851..f88cd881 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -7,7 +7,7 @@ from mock import Mock, PropertyMock, patch
 from threading import Thread
 
 from patroni import psycopg
-from patroni.dcs import Cluster, ClusterConfig, Member
+from patroni.dcs import Cluster, ClusterConfig, Member, SyncState
 from patroni.postgresql import Postgresql
 from patroni.postgresql.misc import fsync_dir
 from patroni.postgresql.slots import SlotsAdvanceThread, SlotsHandler
@@ -31,15 +31,15 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.s = self.p.slots_handler
         self.p.start()
         config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
-        self.cluster = Cluster(True, config, self.leader, 0,
-                               [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345}, None)
+        self.cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem],
+                               None, SyncState.empty(), None, {'ls': 12345}, None)
 
     def test_sync_replication_slots(self):
         config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'},
                                              'A': 0, 'ls': 0, 'b': {'type': 'logical', 'plugin': '1'}},
                                    'ignore_slots': [{'name': 'blabla'}]}, 1)
-        cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, {'test_3': 10}, None)
+        cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem],
+                          None, SyncState.empty(), None, {'test_3': 10}, None)
         with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg.OperationalError)):
             self.s.sync_replication_slots(cluster, False)
         self.p.set_role('standby_leader')
@@ -70,8 +70,8 @@ class TestSlotsHandler(BaseTestPostgresql):
     def test_process_permanent_slots(self):
         config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}},
                                    'ignore_slots': [{'name': 'blabla'}]}, 1)
-        cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, None, None)
+        cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem],
+                          None, SyncState.empty(), None, None, None)
 
         self.s.sync_replication_slots(cluster, False)
         with patch.object(Postgresql, '_query') as mock_query: