Release v3.0.4 (#2754)

- update release notes
- bump version
- bump pyright version

.github/workflows/tests.yaml

@@ -173,4 +173,4 @@ jobs:
 
     - uses: jakebailey/pyright-action@v1
       with:
-        version: 1.1.315
+        version: 1.1.317

.gitignore

@@ -57,3 +57,9 @@ docs/source/_templates/
 
 #VSCode IDE
 .vscode/
+
+# Virtual environment
+venv*/
+
+# Default test data directory
+data/

README.rst

@@ -8,7 +8,7 @@ You can find a version of this documentation that is searchable and also easier
 
 There are many ways to run high availability with PostgreSQL; for a list, see the `PostgreSQL Documentation <https://wiki.postgresql.org/wiki/Replication,_Clustering,_and_Connection_Pooling>`__.
 
-Patroni is a template for high availability (HA) PostgreSQL solutions using Python. For maximum accessibility, Patroni supports a variety of distributed configuration stores like `ZooKeeper <https://zookeeper.apache.org/>`__, `etcd <https://github.com/coreos/etcd>`__, `Consul <https://github.com/hashicorp/consul>`__ or `Kubernetes <https://kubernetes.io>`__. Database engineers, DBAs, DevOps engineers, and SREs who are looking to quickly deploy HA PostgreSQL in datacenters — or anywhere else — will hopefully find it useful.
+Patroni is a template for high availability (HA) PostgreSQL solutions using Python. For maximum accessibility, Patroni supports a variety of distributed configuration stores like `ZooKeeper <https://zookeeper.apache.org/>`__, `etcd <https://github.com/coreos/etcd>`__, `Consul <https://github.com/hashicorp/consul>`__ or `Kubernetes <https://kubernetes.io>`__. Database engineers, DBAs, DevOps engineers, and SREs who are looking to quickly deploy HA PostgreSQL in datacenters - or anywhere else - will hopefully find it useful.
 
 We call Patroni a "template" because it is far from being a one-size-fits-all or plug-and-play replication system. It will have its own caveats. Use wisely.
 

docs/rest_api.rst

@@ -141,8 +141,8 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e
 	# 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
+	# TYPE patroni_sync_standby gauge
-        patroni_sync_standby{scope="batman"} 0
+	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
@@ -155,6 +155,12 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e
 	# HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise.
 	# TYPE patroni_xlog_paused gauge
 	patroni_xlog_paused{scope="batman"} 0
+	# HELP patroni_postgres_streaming Value is 1 if Postgres is streaming, 0 otherwise.
+	# TYPE patroni_postgres_streaming gauge
+	patroni_postgres_streaming{scope="batman"} 1
+	# HELP patroni_postgres_in_archive_recovery Value is 1 if Postgres is replicating from archive, 0 otherwise.
+	# TYPE patroni_postgres_in_archive_recovery gauge
+	patroni_postgres_in_archive_recovery{scope="batman"} 0
 	# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.
 	# TYPE patroni_postgres_server_version gauge
 	patroni_postgres_server_version {scope="batman"} 140004

features/basic_replication.feature

@@ -72,14 +72,13 @@ Feature: basic replication
     Then table bar is present on postgres1 after 20 seconds
     And Response on GET http://127.0.0.1:8010/config contains master_start_timeout after 10 seconds
 
-  Scenario: check immediate failover when master_start_timeout=0
-    Given I kill postmaster on postgres2
-    Then postgres1 is a leader after 10 seconds
-    And postgres1 role is the primary after 10 seconds
-
   Scenario: check rejoin of the former primary with pg_rewind
     Given I add the table splitbrain to postgres0
     And I start postgres0
     Then postgres0 role is the secondary after 20 seconds
-    When I add the table buz to postgres1
+    When I add the table buz to postgres2
     Then table buz is present on postgres0 after 20 seconds
+
+  Scenario: check graceful rejection when two nodes have the same name
+    Given I start duplicate postgres0 on port 8011
+    Then there is a "Can't start; there is already a node named 'postgres0' running" CRITICAL in the dup-postgres0 patroni log

features/citus.feature

@@ -10,20 +10,21 @@ Feature: citus
     And I start postgres3 in citus group 1
     Then replication works from postgres0 to postgres1 after 15 seconds
     Then replication works from postgres2 to postgres3 after 15 seconds
-    And postgres0 is registered in the postgres0 as the worker in group 0
+    And postgres0 is registered in the postgres0 as the primary in group 0 after 5 seconds
-    And postgres2 is registered in the postgres0 as the worker in group 1
+    And postgres2 is registered in the postgres0 as the primary in group 1 after 5 seconds
 
   Scenario: coordinator failover updates pg_dist_node
     Given I run patronictl.py failover batman --group 0 --candidate postgres1 --force
     Then postgres1 role is the primary after 10 seconds
+    And "members/postgres0" key in a group 0 in DCS has state=running after 15 seconds
     And replication works from postgres1 to postgres0 after 15 seconds
+    And postgres1 is registered in the postgres2 as the primary in group 0 after 5 seconds
     And "sync" key in a group 0 in DCS has sync_standby=postgres0 after 15 seconds
-    And postgres1 is registered in the postgres2 as the worker in group 0
+    When I run patronictl.py switchover batman --group 0 --candidate postgres0 --force
-    When I run patronictl.py failover batman --group 0 --candidate postgres0 --force
     Then postgres0 role is the primary after 10 seconds
     And replication works from postgres0 to postgres1 after 15 seconds
+    And postgres0 is registered in the postgres2 as the primary in group 0 after 5 seconds
     And "sync" key in a group 0 in DCS has sync_standby=postgres1 after 15 seconds
-    And postgres0 is registered in the postgres2 as the worker in group 0
 
   Scenario: worker switchover doesn't break client queries on the coordinator
     Given I create a distributed table on postgres0
@@ -31,16 +32,17 @@ Feature: citus
     When I run patronictl.py switchover batman --group 1 --force
     Then I receive a response returncode 0
     And postgres3 role is the primary after 10 seconds
+    And "members/postgres2" key in a group 1 in DCS has state=running after 15 seconds
     And replication works from postgres3 to postgres2 after 15 seconds
+    And postgres3 is registered in the postgres0 as the primary in group 1 after 5 seconds
     And "sync" key in a group 1 in DCS has sync_standby=postgres2 after 15 seconds
-    And postgres3 is registered in the postgres0 as the worker in group 1
     And a thread is still alive
     When I run patronictl.py switchover batman --group 1 --force
     Then I receive a response returncode 0
     And postgres2 role is the primary after 10 seconds
     And replication works from postgres2 to postgres3 after 15 seconds
+    And postgres2 is registered in the postgres0 as the primary in group 1 after 5 seconds
     And "sync" key in a group 1 in DCS has sync_standby=postgres3 after 15 seconds
-    And postgres2 is registered in the postgres0 as the worker in group 1
     And a thread is still alive
     When I stop a thread
     Then a distributed table on postgres0 has expected rows
@@ -52,7 +54,7 @@ Feature: citus
     Then I receive a response returncode 0
     And postgres2 role is the primary after 10 seconds
     And replication works from postgres2 to postgres3 after 15 seconds
-    And postgres2 is registered in the postgres0 as the worker in group 1
+    And postgres2 is registered in the postgres0 as the primary in group 1 after 5 seconds
     And a thread is still alive
     When I stop a thread
     Then a distributed table on postgres0 has expected rows
@@ -64,8 +66,7 @@ Feature: citus
     When I run patronictl.py edit-config batman --group 2 -s ttl=20 --force
     Then I receive a response returncode 0
     And I receive a response output "+ttl: 20"
-    When I sleep for 2 seconds
+    Then postgres4 is registered in the postgres2 as the primary in group 2 after 5 seconds
-    Then postgres4 is registered in the postgres2 as the worker in group 2
     When I shut down postgres4
     Then There is a transaction in progress on postgres0 changing pg_dist_node
     When I run patronictl.py restart batman postgres2 --group 1 --force

features/environment.py

@@ -52,10 +52,9 @@ class AbstractController(abc.ABC):
         self._log = open(os.path.join(self._output_dir, self._name + '.log'), 'a')
         self._handle = self._start()
 
-        assert self._has_started(), "Process {0} is not running after being started".format(self._name)
-
         max_wait_limit *= self._context.timeout_multiplier
         for _ in range(max_wait_limit):
+            assert self._has_started(), "Process {0} is not running after being started".format(self._name)
             if self._is_accessible():
                 break
             time.sleep(1)
@@ -344,6 +343,13 @@ class PatroniController(AbstractController):
                         '--datadir=' + os.path.join(self._work_directory, dest),
                         '--dbname=' + self.backup_source])
 
+    def read_patroni_log(self, level):
+        try:
+            with open(str(os.path.join(self._output_dir or '', self._name + ".log"))) as f:
+                return [line for line in f.readlines() if line[24:24 + len(level)] == level]
+        except IOError:
+            return []
+
 
 class ProcessHang(object):
 
@@ -827,7 +833,7 @@ class PatroniPoolController(object):
 
     def __getattr__(self, func):
         if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to',
-                        'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup']:
+                        'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup', 'read_patroni_log']:
             raise AttributeError("PatroniPoolController instance has no attribute '{0}'".format(func))
 
         def wrapper(name, *args, **kwargs):

features/patroni_api.feature

@@ -35,21 +35,21 @@ Scenario: check local configuration reload
 	Then I receive a response code 202
 
 Scenario: check dynamic configuration change via DCS
-	Given I run patronictl.py edit-config -s 'ttl=10' -p 'max_connections=101' --force batman
+	Given I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "postgresql": {"parameters": {"max_connections": "101"}}}
-	Then I receive a response returncode 0
+	Then I receive a response code 200
-	And I receive a response output "+ttl: 10"
 	And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 11 seconds
 	When I issue a GET request to http://127.0.0.1:8008/config
 	Then I receive a response code 200
-	And I receive a response ttl 10
+	And I receive a response ttl 20
 	When I issue a GET request to http://127.0.0.1:8008/patroni
 	Then I receive a response code 200
 	And I receive a response tags {'new_tag': 'new_value'}
 	And I sleep for 4 seconds
 
 Scenario: check the scheduled restart
-	Given I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"superuser_reserved_connections": "6"}}}
+	Given I run patronictl.py edit-config -p 'superuser_reserved_connections=6' --force batman
-	Then I receive a response code 200
+	Then I receive a response returncode 0
+	And I receive a response output "+    superuser_reserved_connections: 6"
 	And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 5 seconds
 	Given I issue a scheduled restart at http://127.0.0.1:8008 in 5 seconds with {"role": "replica"}
 	Then I receive a response code 202

features/recovery.feature

@@ -0,0 +1,24 @@
+Feature: recovery
+  We want to check that crashed postgres is started back
+
+  Scenario: check that timeline is not incremented when primary is started after crash
+    Given I start postgres0
+    Then postgres0 is a leader after 10 seconds
+    And there is a non empty initialize key in DCS after 15 seconds
+    When I start postgres1
+    And I add the table foo to postgres0
+    Then table foo is present on postgres1 after 20 seconds
+    When I kill postmaster on postgres0
+    Then postgres0 role is the primary after 10 seconds
+    When I issue a GET request to http://127.0.0.1:8008/
+    Then I receive a response code 200
+    And I receive a response role master
+    And I receive a response timeline 1
+
+  Scenario: check immediate failover when master_start_timeout=0
+    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"master_start_timeout": 0}
+    Then I receive a response code 200
+    And Response on GET http://127.0.0.1:8008/config contains master_start_timeout after 10 seconds
+    When I kill postmaster on postgres0
+    Then postgres1 is a leader after 10 seconds
+    And postgres1 role is the primary after 10 seconds

features/standby_cluster.feature

@@ -13,6 +13,10 @@ Feature: standby cluster
     When I start postgres0
     Then "members/postgres0" key in DCS has state=running after 10 seconds
     And replication works from postgres1 to postgres0 after 15 seconds
+    When I issue a GET request to http://127.0.0.1:8008/patroni
+    Then I receive a response code 200
+    And I receive a response replication_state streaming
+    And "members/postgres0" key in DCS has replication_state=streaming after 10 seconds
 
   @slot-advance
   Scenario: check permanent logical slots are synced to the replica
@@ -34,6 +38,9 @@ Feature: standby cluster
     Then postgres1 is a leader of batman1 after 10 seconds
     When I add the table foo to postgres0
     Then table foo is present on postgres1 after 20 seconds
+    When I issue a GET request to http://127.0.0.1:8009/patroni
+    Then I receive a response code 200
+    And I receive a response replication_state streaming
     And I sleep for 3 seconds
     When I issue a GET request to http://127.0.0.1:8009/primary
     Then I receive a response code 503
@@ -44,6 +51,9 @@ Feature: standby cluster
     When I start postgres2 in a cluster batman1
     Then postgres2 role is the replica after 24 seconds
     And table foo is present on postgres2 after 20 seconds
+    When I issue a GET request to http://127.0.0.1:8010/patroni
+    Then I receive a response code 200
+    And I receive a response replication_state streaming
     And postgres1 does not have a logical replication slot named test_logical
 
   Scenario: check failover

features/steps/basic_replication.py

@@ -9,6 +9,22 @@ def start_patroni(context, name):
     return context.pctl.start(name)
 
 
+@step('I start duplicate {name:w} on port {port:d}')
+def start_duplicate_patroni(context, name, port):
+    config = {
+        "name": name,
+        "restapi": {
+            "listen": "127.0.0.1:{0}".format(port)
+        }
+    }
+    try:
+        context.pctl.start('dup-' + name, custom_config=config)
+        assert False, "Process was expected to fail"
+    except AssertionError as e:
+        assert 'is not running after being started' in str(e),\
+            "No error was raised by duplicate start of {0} ".format(name)
+
+
 @step('I shut down {name:w}')
 def stop_patroni(context, name):
     return context.pctl.stop(name, timeout=60)
@@ -90,3 +106,10 @@ def replication_works(context, primary, replica, time_limit):
         When I add the table test_{0} to {1}
         Then table test_{0} is present on {2} after {3} seconds
     """.format(int(time()), primary, replica, time_limit))
+
+
+@then('there is a "{message}" {level:w} in the {node} patroni log')
+def check_patroni_log(context, message, level, node):
+    messsages_of_level = context.pctl.read_patroni_log(node, level)
+    assert any(message in line for line in messsages_of_level),\
+        "There was no {0} {1} in the {2} patroni log".format(message, level, node)

features/steps/citus.py

@@ -44,12 +44,24 @@ def start_citus(context, name, group):
     return context.pctl.start(name, custom_config={"citus": {"database": "postgres", "group": int(group)}})
 
 
-@step('{name1:w} is registered in the {name2:w} as the worker in group {group:d}')
+@step('{name1:w} is registered in the {name2:w} as the {role:w} in group {group:d} after {time_limit:d} seconds')
-def check_registration(context, name1, name2, group):
+def check_registration(context, name1, name2, role, group, time_limit):
+    time_limit *= context.timeout_multiplier
+    max_time = time.time() + int(time_limit)
+
     worker_port = int(context.pctl.query(name1, "SHOW port").fetchone()[0])
-    r = context.pctl.query(name2, "SELECT nodeport FROM pg_catalog.pg_dist_node WHERE groupid = {0}".format(group))
+
-    assert worker_port == r.fetchone()[0],\
+    while time.time() < max_time:
-        "Worker {0} is not registered in pg_dist_node on the coordinator {1}".format(name1, name2)
+        try:
+            cur = context.pctl.query(name2, "SELECT nodeport, noderole"
+                                            " FROM pg_catalog.pg_dist_node WHERE groupid = {0}".format(group))
+            mapping = {r[0]: r[1] for r in cur}
+            if mapping.get(worker_port) == role:
+                return
+        except Exception:
+            pass
+        time.sleep(1)
+    assert False, "Node {0} is not registered in pg_dist_node on the node {1}".format(name1, name2)
 
 
 @step('I create a distributed table on {name:w}')

patroni/__main__.py

@@ -30,12 +30,15 @@ class Patroni(AbstractPatroniDaemon):
 
         self.version = __version__
         self.dcs = get_dcs(self.config)
+        self.request = PatroniRequest(self.config, True)
+
+        self.ensure_unique_name()
+
         self.watchdog = Watchdog(self.config)
         self.load_dynamic_configuration()
 
         self.postgresql = Postgresql(self.config['postgresql'])
         self.api = RestApiServer(self, self.config['restapi'])
-        self.request = PatroniRequest(self.config, True)
         self.ha = Ha(self)
 
         self.tags = self.get_tags()
@@ -60,6 +63,23 @@ class Patroni(AbstractPatroniDaemon):
                 logger.warning('Can not get cluster from dcs')
                 time.sleep(5)
 
+    def ensure_unique_name(self) -> None:
+        """A helper method to prevent splitbrain from operator naming error."""
+        from patroni.dcs import Member
+
+        cluster = self.dcs.get_cluster()
+        if not cluster:
+            return
+        member = cluster.get_member(self.config['name'], False)
+        if not isinstance(member, Member):
+            return
+        try:
+            _ = self.request(member, endpoint="/liveness")
+            logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
+            sys.exit(1)
+        except Exception:
+            return
+
     def get_tags(self) -> Dict[str, Any]:
         return {tag: value for tag, value in self.config.get('tags', {}).items()
                 if tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync') or value}

patroni/api.py

@@ -535,6 +535,18 @@ class RestApiHandler(BaseHTTPRequestHandler):
         metrics.append("patroni_xlog_paused{0} {1}"
                        .format(scope_label, int(postgres.get('xlog', {}).get('paused', False) is True)))
 
+        if postgres.get('server_version', 0) >= 90600:
+            metrics.append("# HELP patroni_postgres_streaming Value is 1 if Postgres is streaming, 0 otherwise.")
+            metrics.append("# TYPE patroni_postgres_streaming gauge")
+            metrics.append("patroni_postgres_streaming{0} {1}"
+                           .format(scope_label, int(postgres.get('replication_state') == 'streaming')))
+
+            metrics.append("# HELP patroni_postgres_in_archive_recovery Value is 1"
+                           " if Postgres is replicating from archive, 0 otherwise.")
+            metrics.append("# TYPE patroni_postgres_in_archive_recovery gauge")
+            metrics.append("patroni_postgres_in_archive_recovery{0} {1}"
+                           .format(scope_label, int(postgres.get('replication_state') == 'in archive recovery')))
+
         metrics.append("# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.")
         metrics.append("# TYPE patroni_postgres_server_version gauge")
         metrics.append("patroni_postgres_server_version {0} {1}".format(scope_label, postgres.get('server_version', 0)))
@@ -1151,8 +1163,11 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
             if postgresql.state not in ('running', 'restarting', 'starting'):
                 raise RetryFailedError('')
+            replication_state = ('(pg_catalog.pg_stat_get_wal_receiver()).status'
+                                 if postgresql.major_version >= 90600 else 'NULL') + ", " +\
+                ("pg_catalog.current_setting('restore_command')" if postgresql.major_version >= 120000 else "NULL")
             stmt = ("SELECT " + postgresql.POSTMASTER_START_TIME + ", " + postgresql.TL_LSN + ","
-                    " pg_catalog.pg_last_xact_replay_timestamp(),"
+                    " pg_catalog.pg_last_xact_replay_timestamp(), " + replication_state + ","
                     " pg_catalog.array_to_json(pg_catalog.array_agg(pg_catalog.row_to_json(ri))) "
                     "FROM (SELECT (SELECT rolname FROM pg_catalog.pg_authid WHERE oid = usesysid) AS usename,"
                     " application_name, client_addr, w.state, sync_state, sync_priority"
@@ -1188,8 +1203,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     if not cluster or cluster.is_unlocked() or not cluster.leader else cluster.leader.timeline
                 result['timeline'] = postgresql.replica_cached_timeline(leader_timeline)
 
-            if row[7]:
+            replication_state = postgresql.replication_state_from_parameters(row[1] > 0, row[7], row[8])
-                result['replication'] = row[7]
+            if replication_state:
+                result['replication_state'] = replication_state
+
+            if row[9]:
+                result['replication'] = row[9]
 
         except (psycopg.Error, RetryFailedError, PostgresConnectionException):
             state = postgresql.state
@@ -1541,11 +1560,11 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
         if self.__ssl_options.get('certfile'):
             import ssl
             try:
-                ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+                crt: Dict[str, Any] = ssl._ssl._test_decode_cert(self.__ssl_options['certfile'])  # pyright: ignore
-                crts = ctx.load_verify_locations(self.__ssl_options['certfile'])
+                if TYPE_CHECKING:  # pragma: no cover
-                if crts:
+                    assert isinstance(crt, dict)
-                    return crts[0].get('serialNumber')
+                return crt.get('serialNumber')
-            except Exception as e:
+            except ssl.SSLError as e:
                 logger.error('Failed to get serial number from certificate %s: %r', self.__ssl_options['certfile'], e)
 
     def reload_local_certificate(self) -> Optional[bool]:

patroni/ctl.py

@@ -258,7 +258,7 @@ option_insecure = click.option('-k', '--insecure', is_flag=True, help='Allow con
 role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 'standby', 'any', 'master'])
 
 
-@click.group()
+@click.group(cls=click.Group)
 @click.option('--config-file', '-c', help='Configuration file',
               envvar='PATRONICTL_CONFIG_FILE', default=CONFIG_FILE_PATH)
 @click.option('--dcs-url', '--dcs', '-d', 'dcs_url', help='The DCS connect url', envvar='DCS_URL')
@@ -1490,7 +1490,8 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str,
         * ``Role``: ``Leader``, ``Standby Leader``, ``Sync Standby`` or ``Replica``;
         * ``State``: ``stopping``, ``stopped``, ``stop failed``, ``crashed``, ``running``, ``starting``,
           ``start failed``, ``restarting``, ``restart failed``, ``initializing new cluster``, ``initdb failed``,
-          ``running custom bootstrap script``, ``custom bootstrap failed``, or ``creating replica``, and so on;
+          ``running custom bootstrap script``, ``custom bootstrap failed``, ``creating replica``, ``streaming``,
+          ``in archive recovery``, and so on;
         * ``TL``: current timeline in Postgres;
           ``Lag in MB``: replication lag.
 

patroni/dcs/__init__.py

@@ -239,23 +239,26 @@ class Member(NamedTuple):
 
 
 class RemoteMember(Member):
-    """Represents a remote member (typically a primary) for a standby cluster"""
+    """Represents a remote member (typically a primary) for a standby cluster.
+
+    :cvar ALLOWED_KEYS: Controls access to relevant key names that could be in stored :attr:`~RemoteMember.data`.
+    """
+
+    ALLOWED_KEYS: Tuple[str, ...] = (
+        'primary_slot_name',
+        'create_replica_methods',
+        'restore_command',
+        'archive_cleanup_command',
+        'recovery_min_apply_delay',
+        'no_replication_slot'
+    )
 
     @classmethod
     def from_name_and_data(cls, name: str, data: Dict[str, Any]) -> 'RemoteMember':
         return super(RemoteMember, cls).__new__(cls, -1, name, None, data)
 
-    @staticmethod
-    def allowed_keys() -> Tuple[str, ...]:
-        return ('primary_slot_name',
-                'create_replica_methods',
-                'restore_command',
-                'archive_cleanup_command',
-                'recovery_min_apply_delay',
-                'no_replication_slot')
-
     def __getattr__(self, name: str) -> Any:
-        if name in RemoteMember.allowed_keys():
+        if name in RemoteMember.ALLOWED_KEYS:
             return self.data.get(name)
 
 

patroni/dcs/consul.py

@@ -400,8 +400,12 @@ class Consul(AbstractDCS):
 
         return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
 
+    @property
+    def _consistency(self) -> str:
+        return 'consistent' if self._ctl else self._client.consistency
+
     def _cluster_loader(self, path: str) -> Cluster:
-        _, results = self.retry(self._client.kv.get, path, recurse=True)
+        _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency)
         if results is None:
             raise NotFound
         nodes = {}
@@ -412,7 +416,7 @@ class Consul(AbstractDCS):
         return self._cluster_from_nodes(nodes)
 
     def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
-        _, results = self.retry(self._client.kv.get, path, recurse=True)
+        _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency)
         clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict)
         for node in results or []:
             key = node['Key'][len(path):].split('/', 1)

patroni/dcs/etcd.py

@@ -99,7 +99,7 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
         self._dns_resolver = dns_resolver
         self.set_machines_cache_ttl(cache_ttl)
         self._machines_cache_updated = 0
-        kwargs = {p: config.get(p) for p in ('host', 'port', 'protocol', 'use_proxies',
+        kwargs = {p: config.get(p) for p in ('host', 'port', 'protocol', 'use_proxies', 'version_prefix',
                                              'username', 'password', 'cert', 'ca_cert') if config.get(p)}
         super(AbstractEtcdClientWithFailover, self).__init__(read_timeout=config['retry_timeout'], **kwargs)
         # For some reason python3-etcd on debian and ubuntu are not based on the latest version
@@ -443,6 +443,9 @@ class EtcdClient(AbstractEtcdClientWithFailover):
 
     ERROR_CLS = EtcdError
 
+    def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None:
+        super(EtcdClient, self).__init__({**config, 'version_prefix': None}, dns_resolver, cache_ttl)
+
     def __del__(self) -> None:
         try:
             self.http.clear()
@@ -722,13 +725,13 @@ class Etcd(AbstractEtcd):
         return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
 
     def _cluster_loader(self, path: str) -> Cluster:
-        result = self.retry(self._client.read, path, recursive=True)
+        result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
         nodes = {node.key[len(result.key):].lstrip('/'): node for node in result.leaves}
         return self._cluster_from_nodes(result.etcd_index, nodes)
 
     def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
         clusters: Dict[int, Dict[str, etcd.EtcdResult]] = defaultdict(dict)
-        result = self.retry(self._client.read, path, recursive=True)
+        result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
         for node in result.leaves:
             key = node.key[len(result.key):].lstrip('/').split('/', 1)
             if len(key) == 2 and citus_group_re.match(key[0]):

patroni/dcs/etcd3.py

@@ -206,8 +206,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
     def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None:
         self._token = None
         self._cluster_version: Tuple[int] = tuple()
-        self.version_prefix = '/v3beta'
+        super(Etcd3Client, self).__init__({**config, 'version_prefix': '/v3beta'}, dns_resolver, cache_ttl)
-        super(Etcd3Client, self).__init__(config, dns_resolver, cache_ttl)
 
         try:
             self.authenticate()
@@ -327,14 +326,14 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
             return retry(e)
 
     @_handle_auth_errors
-    def range(self, key: str, range_end: Union[bytes, str, None] = None,
+    def range(self, key: str, range_end: Union[bytes, str, None] = None, serializable: bool = True,
               retry: Optional[Retry] = None) -> Dict[str, Any]:
         params = build_range_request(key, range_end)
-        params['serializable'] = True  # For better performance. We can tolerate stale reads.
+        params['serializable'] = serializable  # For better performance. We can tolerate stale reads
         return self.call_rpc('/kv/range', params, retry)
 
-    def prefix(self, key: str, retry: Optional[Retry] = None) -> Dict[str, Any]:
+    def prefix(self, key: str, serializable: bool = True, retry: Optional[Retry] = None) -> Dict[str, Any]:
-        return self.range(key, prefix_range_end(key), retry)
+        return self.range(key, prefix_range_end(key), serializable, retry)
 
     @_handle_auth_errors
     def lease_grant(self, ttl: int, retry: Optional[Retry] = None) -> str:
@@ -595,7 +594,8 @@ class PatroniEtcd3Client(Etcd3Client):
                 self._wait_cache(self.read_timeout)
                 ret = self._kv_cache.copy()
         else:
-            ret = self._etcd3.retry(self.prefix, path).get('kvs', [])
+            serializable = not getattr(self._etcd3, '_ctl')  # use linearizable for patronictl
+            ret = self._etcd3.retry(self.prefix, path, serializable).get('kvs', [])
         for node in ret:
             node.update({'key': base64_decode(node['key']),
                          'value': base64_decode(node.get('value', '')),

patroni/dcs/kubernetes.py

@@ -766,7 +766,7 @@ class Kubernetes(AbstractDCS):
             k8s_config.load_kube_config(context=config.get('context', 'kind-kind'))
 
         pod_ip = config.get('pod_ip')
-        self.__ips: List[str] = [] if config.get('patronictl') or not isinstance(pod_ip, str) else [pod_ip]
+        self.__ips: List[str] = [] if self._ctl or not isinstance(pod_ip, str) else [pod_ip]
         self.__ports: List[K8sObject] = []
         ports: List[Dict[str, Any]] = config.get('ports', [{}])
         for p in ports:
@@ -774,7 +774,7 @@ class Kubernetes(AbstractDCS):
             port.update({n: p[n] for n in ('name', 'protocol') if p.get(n)})
             self.__ports.append(k8s_client.V1EndpointPort(**port))
 
-        bypass_api_service = not config.get('patronictl') and config.get('bypass_api_service')
+        bypass_api_service = not self._ctl and config.get('bypass_api_service')
         self._api = CoreV1ApiProxy(config.get('use_endpoints'), bypass_api_service)
         self._should_create_config_service = self._api.use_endpoints
         self.reload_config(config)

patroni/ha.py

@@ -149,7 +149,6 @@ class Ha(object):
         self._leader_timeline = None
         self.recovering = False
         self._async_response = CriticalTask()
-        self._crash_recovery_executed = False
         self._crash_recovery_started = 0
         self._start_timeout = None
         self._async_executor = AsyncExecutor(self.state_handler.cancellable, self.wakeup)
@@ -307,10 +306,13 @@ class Ha(object):
             if self._async_executor.scheduled_action in (None, 'promote') \
                     and data['state'] in ['running', 'restarting', 'starting']:
                 try:
-                    timeline: Optional[int]
                     timeline, wal_position, pg_control_timeline = self.state_handler.timeline_wal_position()
                     data['xlog_location'] = wal_position
-                    if not timeline:  # try pg_stat_wal_receiver to get the timeline
+                    if not timeline:  # running as a standby
+                        replication_state = self.state_handler.replication_state()
+                        if replication_state:
+                            data['replication_state'] = replication_state
+                        # try pg_stat_wal_receiver to get the timeline
                         timeline = self.state_handler.received_timeline()
                     if not timeline:
                         # So far the only way to get the current timeline on the standby is from
@@ -411,8 +413,7 @@ class Ha(object):
         return result
 
     def _handle_crash_recovery(self) -> Optional[str]:
-        if not self._crash_recovery_executed and (self.cluster.is_unlocked() or self._rewind.can_rewind):
+        if self._crash_recovery_started == 0 and (self.cluster.is_unlocked() or self._rewind.can_rewind):
-            self._crash_recovery_executed = True
             self._crash_recovery_started = time.time()
             msg = 'doing crash recovery in a single user mode'
             return self._async_executor.try_run_async(msg, self._rewind.ensure_clean_shutdown) or msg
@@ -438,15 +439,29 @@ class Ha(object):
             return self._async_executor.try_run_async(msg, self._do_reinitialize, args=(self.cluster,)) or msg
 
     def recover(self) -> str:
-        # Postgres is not running and we will restart in standby mode. Watchdog is not needed until we promote.
+        """Handle the case when postgres isn't running.
-        self.watchdog.disable()
 
+        Depending on the state of Patroni, DCS cluster view, and pg_controldata the following could happen:
+        - if ``primary_start_timeout`` is 0 and this node owns the leader lock, the lock
+          will be voluntarily released if there are healthy replicas to take it over.
+        - if postgres was running as a ``primary`` and this node owns the leader lock, postgres is started as primary.
+        - crash recover in a single-user mode is executed in the following cases:
+          - postgres was running as ``primary`` wasn't ``shut down`` cleanly and there is no leader in DCS
+          - postgres was running as ``replica`` wasn't ``shut down in recovery`` (cleanly)
+            and we need to run ``pg_rewind`` to join back to the cluster.
+        - ``pg_rewind`` is executed if it is necessary, or optinally, the data directory could
+           be removed if it is allowed by configuration.
+        - after ``crash recovery`` and/or ``pg_rewind`` are executed, postgres is started in recovery.
+
+        :returns: action message, describing what was performed.
+        """
         if self.has_lock() and self.update_lock():
             timeout = self.global_config.primary_start_timeout
             if timeout == 0:
                 # We are requested to prefer failing over to restarting primary. But see first if there
                 # is anyone to fail over to.
                 if self.is_failover_possible(self.cluster.members):
+                    self.watchdog.disable()
                     logger.info("Primary crashed. Failing over.")
                     self.demote('immediate')
                     return 'stopped PostgreSQL to fail over after a crash'
@@ -455,6 +470,23 @@ class Ha(object):
 
         data = self.state_handler.controldata()
         logger.info('pg_controldata:\n%s\n', '\n'.join('  {0}: {1}'.format(k, v) for k, v in data.items()))
+
+        # timeout > 0 indicates that we still have the leader lock, and it was just updated
+        if timeout\
+                and data.get('Database cluster state') in ('in production', 'shutting down', 'shut down')\
+                and self.state_handler.state == 'crashed'\
+                and self.state_handler.role in ('primary', 'master')\
+                and not self.state_handler.config.recovery_conf_exists():
+            # We know 100% that we were running as a primary a few moments ago, therefore could just start postgres
+            msg = 'starting primary after failure'
+            if self._async_executor.try_run_async(msg, self.state_handler.start,
+                                                  args=(timeout, self._async_executor.critical_task)) is None:
+                self.recovering = True
+                return msg
+
+        # Postgres is not running, and we will restart in standby mode. Watchdog is not needed until we promote.
+        self.watchdog.disable()
+
         if data.get('Database cluster state') in ('in production', 'shutting down', 'in crash recovery'):
             msg = self._handle_crash_recovery()
             if msg:
@@ -965,9 +997,6 @@ class Ha(object):
             if ret is not None:  # continue if we just deleted the stale failover key as a leader
                 return ret
 
-        if self.state_handler.is_starting():  # postgresql still starting up is unhealthy
-            return False
-
         if self.state_handler.is_leader():
             # in pause leader is the healthiest only when no initialize or sysid matches with initialize!
             return not self.is_paused() or not self.cluster.initialize\
@@ -1451,7 +1480,7 @@ class Ha(object):
 
             if self.state_handler.role in ('master', 'primary'):
                 logger.info('Demoting primary during %s', self._async_executor.scheduled_action)
-                if self._async_executor.scheduled_action == 'restart':
+                if self._async_executor.scheduled_action in ('restart', 'starting primary after failure'):
                     # Restart needs a special interlocking cancel because postmaster may be just started in a
                     # background thread and has not even written a pid file yet.
                     with self._async_executor.critical_task as task:
@@ -1515,6 +1544,9 @@ class Ha(object):
         self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':')))
         self.dcs.take_leader()
         self.set_is_leader(True)
+        if self.is_synchronous_mode():
+            self.state_handler.sync_handler.set_synchronous_standby_names(
+                CaseInsensitiveSet('*') if self.global_config.is_synchronous_mode_strict else CaseInsensitiveSet())
         self.state_handler.call_nowait(CallbackAction.ON_START)
         self.load_cluster_from_dcs()
 
@@ -1616,7 +1648,7 @@ class Ha(object):
                         return msg
 
                 # Reset some states after postgres successfully started up
-                self._crash_recovery_executed = False
+                self._crash_recovery_started = 0
                 if self._rewind.executed and not self._rewind.failed:
                     self._rewind.reset_state()
 
@@ -1708,16 +1740,21 @@ class Ha(object):
                 msg = self.process_healthy_cluster()
                 ret = self.evaluate_scheduled_restart() or msg
 
-            # we might not have a valid PostgreSQL connection here if another thread
+            # We might not have a valid PostgreSQL connection here if AsyncExecutor is doing
-            # stops PostgreSQL, therefore, we only reload replication slots if no
+            # something with PostgreSQL. Therefore we will sync replication slots only if no
-            # asynchronous processes are running (should be always the case for the primary)
+            # asynchronous processes are running or we know that this is a standby being promoted.
-            if not self._async_executor.busy and not self.state_handler.is_starting():
+            # But, we don't want to run pg_rewind checks or copy logical slots from itself,
+            # therefore we have a couple additional `not is_promoting` checks.
+            is_promoting = self._async_executor.scheduled_action == 'promote'
+            if (not self._async_executor.busy or is_promoting) and not self.state_handler.is_starting():
                 create_slots = self._sync_replication_slots(False)
+
                 if not self.state_handler.cb_called:
-                    if not self.state_handler.is_leader():
+                    if not is_promoting and not self.state_handler.is_leader():
                         self._rewind.trigger_check_diverged_lsn()
                     self.state_handler.call_nowait(CallbackAction.ON_START)
-                if create_slots and self.cluster.leader:
+
+                if not is_promoting and create_slots and self.cluster.leader:
                     err = self._async_executor.try_run_async('copy_logical_slots',
                                                              self.state_handler.slots_handler.copy_logical_slots,
                                                              args=(self.cluster, create_slots))
@@ -1862,7 +1899,7 @@ class Ha(object):
         cluster_params = self.global_config.get_standby_cluster_config()
 
         if cluster_params:
-            data.update({k: v for k, v in cluster_params.items() if k in RemoteMember.allowed_keys()})
+            data.update({k: v for k, v in cluster_params.items() if k in RemoteMember.ALLOWED_KEYS})
             data['no_replication_slot'] = 'primary_slot_name' not in cluster_params
             conn_kwargs = member.conn_kwargs() if member else \
                 {k: cluster_params[k] for k in ('host', 'port') if k in cluster_params}

patroni/postgresql/__init__.py

@@ -197,18 +197,19 @@ class Postgresql(object):
                         and self.role in ('master', 'primary', 'promoted') else "'on', '', NULL")
 
         if self._major_version >= 90600:
-            extra = ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, "
+            extra = ("pg_catalog.current_setting('restore_command')" if self._major_version >= 120000 else "NULL") +\
-                     "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"
+                ", " + ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, "
-                     " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
+                        "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"
-                     if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra
+                        " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
+                        if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra
             extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END,"
-                     " slot_name, conninfo, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra)
+                     " slot_name, conninfo, status, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra)
             if self.role == 'standby_leader':
                 extra = "timeline_id" + extra + ", pg_catalog.pg_control_checkpoint()"
             else:
                 extra = "0" + extra
         else:
-            extra = "0, NULL, NULL, NULL, NULL" + extra
+            extra = "0, NULL, NULL, NULL, NULL, NULL, NULL" + extra
 
         return ("SELECT " + self.TL_LSN + ", {2}").format(self.wal_name, self.lsn_name, extra)
 
@@ -426,7 +427,8 @@ class Postgresql(object):
                 result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone()
                 cluster_info_state = dict(zip(['timeline', 'wal_position', 'replayed_location',
                                                'received_location', 'replay_paused', 'pg_control_timeline',
-                                               'received_tli', 'slot_name', 'conninfo', 'slots', 'synchronous_commit',
+                                               'received_tli', 'slot_name', 'conninfo', 'receiver_state',
+                                               'restore_command', 'slots', 'synchronous_commit',
                                                'synchronous_standby_names', 'pg_stat_replication'], result))
                 if self._has_permanent_logical_slots:
                     cluster_info_state['slots'] =\
@@ -472,6 +474,41 @@ class Postgresql(object):
         """:returns: a result set of 'SELECT * FROM pg_stat_replication'."""
         return self._cluster_info_state_get('pg_stat_replication') or []
 
+    def replication_state_from_parameters(self, is_leader: bool, receiver_state: Optional[str],
+                                          restore_command: Optional[str]) -> Optional[str]:
+        """Figure out the replication state from input parameters.
+
+        .. note::
+            This method could be only called when Postgres is up, running and queries are successfuly executed.
+
+        :is_leader: `True` is postgres is not running in recovery
+        :receiver_state: value from `pg_stat_get_wal_receiver.state` or None if Postgres is older than 9.6
+        :restore_command: value of ``restore_command`` GUC for PostgreSQL 12+ or
+                          `postgresql.recovery_conf.restore_command` if it is set in Patroni configuration
+
+        :returns: - `None` for the primary and for Postgres older than 9.6;
+                  - 'streaming' if replica is streaming according to the `pg_stat_wal_receiver` view;
+                  - 'in archive recovery' if replica isn't streaming and there is a `restore_command`
+        """
+        if self._major_version >= 90600 and not is_leader:
+            if receiver_state == 'streaming':
+                return 'streaming'
+            # For Postgres older than 12 we get `restore_command` from Patroni config, otherwise we check GUC
+            if self._major_version < 120000 and self.config.restore_command() or restore_command:
+                return 'in archive recovery'
+
+    def replication_state(self) -> Optional[str]:
+        """Checks replication state from `pg_stat_get_wal_receiver()`.
+
+        .. note::
+            Available only since 9.6
+
+        :returns: ``streaming``, ``in archive recovery``, or ``None``
+        """
+        return self.replication_state_from_parameters(self.is_leader(),
+                                                      self._cluster_info_state_get('receiver_state'),
+                                                      self._cluster_info_state_get('restore_command'))
+
     def is_leader(self) -> bool:
         try:
             return bool(self._cluster_info_state_get('timeline'))

patroni/postgresql/bootstrap.py

@@ -185,7 +185,7 @@ class Bootstrap(object):
                 r['host'] = 'localhost'  # set it to localhost to write into pgpass
 
             env = self._postgresql.config.write_pgpass(r)
-            env['PGOPTIONS'] = '-c synchronous_commit=local'
+            env['PGOPTIONS'] = '-c synchronous_commit=local -c statement_timeout=0'
 
             try:
                 ret = self._postgresql.cancellable.call(shlex.split(cmd) + [connstring], env=env)

patroni/postgresql/config.py

@@ -1177,3 +1177,6 @@ class ConfigHandler(object):
 
     def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
         return self._config.get(key, default)
+
+    def restore_command(self) -> Optional[str]:
+        return (self.get('recovery_conf') or {}).get('restore_command')

patroni/postgresql/slots.py

@@ -21,8 +21,24 @@ logger = logging.getLogger(__name__)
 
 
 def compare_slots(s1: Dict[str, Any], s2: Dict[str, Any], dbid: str = 'database') -> bool:
-    return s1['type'] == s2['type'] and (s1['type'] == 'physical'
+    """Compare 2 replication slot objects for equality.
-                                         or s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin'])
+
+    ..note ::
+        If the first argument is a ``physical`` replication slot then only the `type` of the second slot is compared.
+        If the first argument is another ``type`` (e.g. ``logical``) then *dbid* and ``plugin`` are compared.
+
+    :param s1: First slot dictionary to be compared.
+    :param s2: Second slot dictionary to be compared.
+    :param dbid: Optional attribute to be compared when comparing ``logical`` replication slots.
+
+    :return: ``True`` if the slot ``type`` of *s1* and *s2* is matches, and the ``type`` of *s1* is ``physical``,
+             OR the ``types`` match AND the *dbid* and ``plugin`` attributes are equal.
+
+    """
+    return (s1['type'] == s2['type']
+            and (s1['type'] == 'physical'
+                 or s1.get(dbid) == s2.get(dbid)
+                 and s1['plugin'] == s2['plugin']))
 
 
 class SlotsAdvanceThread(Thread):

patroni/postgresql/sync.py

@@ -193,7 +193,12 @@ class SyncHandler(object):
 
         # Newly connected replicas will be counted as sync only when reached self._primary_flush_lsn
         self._primary_flush_lsn = self._postgresql.last_operation()
-        self._postgresql.query('SELECT pg_catalog.txid_current()')  # Ensure some WAL traffic to move replication
+        # Ensure some WAL traffic to move replication
+        self._postgresql.query("""DO $$
+BEGIN
+    SET local synchronous_commit = 'off';
+    PERFORM * FROM pg_catalog.txid_current();
+END;$$""")
         self._postgresql.reset_cluster_info_state(None)  # Reset internal cache to query fresh values
 
     def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]:
@@ -289,6 +294,6 @@ class SyncHandler(object):
         # Reset internal cache to query fresh values
         self._postgresql.reset_cluster_info_state(None)
 
-        # timeline == 0 -- indicates that this is the replica, shoudn't ever happen
+        # timeline == 0 -- indicates that this is the replica
         if self._postgresql.get_primary_timeline() > 0:
             self._handle_synchronous_standby_names_change()

patroni/utils.py

@@ -326,6 +326,21 @@ def parse_int(value: Any, base_unit: Optional[str] = None) -> Optional[int]:
         >>> parse_int('1TB', 'GB') is None
         True
 
+        >>> parse_int(50, None) == 50
+        True
+
+        >>> parse_int("51", None) == 51
+        True
+
+        >>> parse_int("nonsense", None) == None
+        True
+
+        >>> parse_int("nonsense", "kB") == None
+        True
+
+        >>> parse_int("nonsense") == None
+        True
+
         >>> parse_int(0) == 0
         True
 
@@ -758,7 +773,8 @@ def cluster_as_json(cluster: 'Cluster', global_config: Optional['GlobalConfig']
         else:
             role = 'replica'
 
-        member = {'name': m.name, 'role': role, 'state': m.data.get('state', ''), 'api_url': m.api_url}
+        state = (m.data.get('replication_state', '') if role != 'leader' else '') or m.data.get('state', '')
+        member = {'name': m.name, 'role': role, 'state': state, 'api_url': m.api_url}
         conn_kwargs = m.conn_kwargs()
         if conn_kwargs.get('host'):
             member['host'] = conn_kwargs['host']

patroni/validator.py

@@ -792,8 +792,7 @@ class IntValidator(object):
         :param value: value to be checked against the rules defined for this :class:`IntValidator` instance.
         :returns: ``True`` if *value* is valid and within the expected range.
         """
-        if self.base_unit:
+        value = parse_int(value, self.base_unit) or ""
-            value = parse_int(value, self.base_unit) or ""
         ret = isinstance(value, int)\
             and (self.min is None or value >= self.min)\
             and (self.max is None or value <= self.max)

patroni/version.py

@@ -2,4 +2,4 @@
 
 :var __version__: the current Patroni version.
 """
-__version__ = '3.0.3'
+__version__ = '3.0.4'

setup.py

@@ -88,7 +88,7 @@ class Flake8(_Command):
                 yield package_directory
 
     def targets(self):
-        return [package for package in self.package_files()] + ['tests', 'setup.py']
+        return [package for package in self.package_files()] + ['tests', 'features', 'setup.py']
 
     def run(self):
         from flake8.main.cli import main
@@ -116,7 +116,7 @@ class PyTest(_Command):
 
 
 def read(fname):
-    with open(os.path.join(__location__, fname)) as fd:
+    with open(os.path.join(__location__, fname), encoding='utf-8') as fd:
         return fd.read()
 
 

tests/__init__.py

@@ -108,7 +108,7 @@ class MockCursor(object):
         elif sql.startswith('WITH slots AS (SELECT slot_name, active'):
             self.results = [(False, True)] if self.rowcount == 1 else [None]
         elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'):
-            self.results = [(1, 2, 1, 0, False, 1, 1, None, None,
+            self.results = [(1, 2, 1, 0, False, 1, 1, None, None, 'streaming', '',
                              [{"slot_name": "ls", "confirmed_flush_lsn": 12345}],
                              'on', 'n1', None)]
         elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'):
@@ -117,7 +117,7 @@ class MockCursor(object):
             replication_info = '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' +\
                                '"state":"streaming","sync_state":"async","sync_priority":0}]'
             now = datetime.datetime.now(tzutc)
-            self.results = [(now, 0, '', 0, '', False, now, replication_info)]
+            self.results = [(now, 0, '', 0, '', False, now, 'streaming', None, replication_info)]
         elif sql.startswith('SELECT name, setting'):
             self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
                             ('wal_block_size', '8192', None, 'integer', 'internal'),

tests/test_api.py

@@ -29,7 +29,8 @@ class MockPostgresql(object):
     name = 'test'
     state = 'running'
     role = 'primary'
-    server_version = '999999'
+    server_version = 90625
+    major_version = 90600
     sysid = 'dummysysid'
     scope = 'dummy'
     pending_restart = True
@@ -55,6 +56,10 @@ class MockPostgresql(object):
     def is_running():
         return True
 
+    @staticmethod
+    def replication_state_from_parameters(*args):
+        return 'streaming'
+
 
 class MockWatchdog(object):
     is_healthy = False
@@ -180,7 +185,6 @@ class MockRestApiServer(RestApiServer):
 
 @patch('ssl.SSLContext.load_cert_chain', Mock())
 @patch('ssl.SSLContext.wrap_socket', Mock(return_value=0))
-@patch('ssl.SSLContext.load_verify_locations', Mock(return_value=[Mock()]))
 @patch.object(HTTPServer, '__init__', Mock())
 class TestRestApiHandler(unittest.TestCase):
 
@@ -220,7 +224,7 @@ class TestRestApiHandler(unittest.TestCase):
         with patch.object(MockHa, 'restart_scheduled', Mock(return_value=True)):
             MockRestApiServer(RestApiHandler, 'GET /primary')
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /primary'))
-        with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, '')])):
+        with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, None, None, '')])):
             self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni'))
         with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)),\
                 patch.object(GlobalConfig, 'is_paused', Mock(return_value=True)):
@@ -589,7 +593,6 @@ class TestRestApiServer(unittest.TestCase):
     @patch('ssl.SSLContext.load_cert_chain', Mock())
     @patch('ssl.SSLContext.set_ciphers', Mock())
     @patch('ssl.SSLContext.wrap_socket', Mock(return_value=0))
-    @patch('ssl.SSLContext.load_verify_locations', Mock(return_value=[Mock()]))
     @patch.object(HTTPServer, '__init__', Mock())
     def setUp(self):
         self.srv = MockRestApiServer(Mock(), '', {'listen': '*:8008', 'certfile': 'a', 'verify_client': 'required',
@@ -652,10 +655,9 @@ class TestRestApiServer(unittest.TestCase):
         mock_get_request.return_value = (self.__create_socket(), ('127.0.0.1', 55555))
         self.srv._handle_request_noblock()
 
-    @patch('ssl.SSLContext.load_verify_locations', Mock(return_value=[Mock()]))
+    @patch('ssl._ssl._test_decode_cert', Mock())
     def test_reload_local_certificate(self):
         self.assertTrue(self.srv.reload_local_certificate())
 
-    @patch('ssl.SSLContext.load_verify_locations', Mock(side_effect=Exception))
     def test_get_certificate_serial_number(self):
         self.assertIsNone(self.srv.get_certificate_serial_number())

tests/test_ha.py

@@ -223,9 +223,12 @@ class TestHa(PostgresInit):
 
     @patch.object(Postgresql, 'received_timeline', Mock(return_value=None))
     def test_touch_member(self):
+        self.p._major_version = 110000
+        self.p.is_leader = false
         self.p.timeline_wal_position = Mock(return_value=(0, 1, 0))
         self.p.replica_cached_timeline = Mock(side_effect=Exception)
-        self.ha.touch_member()
+        with patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value='streaming')):
+            self.ha.touch_member()
         self.p.timeline_wal_position = Mock(return_value=(0, 1, 1))
         self.p.set_role('standby_leader')
         self.ha.touch_member()
@@ -282,11 +285,20 @@ class TestHa(PostgresInit):
         self.p.follow = false
         self.p.is_running = false
         self.p.name = 'leader'
-        self.p.set_role('primary')
+        self.p.set_role('demoted')
         self.p.controldata = lambda: {'Database cluster state': 'shut down', 'Database system identifier': SYSID}
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.assertEqual(self.ha.run_cycle(), 'starting as readonly because i had the session lock')
 
+    def test_start_primary_after_failure(self):
+        self.p.start = false
+        self.p.is_running = false
+        self.p.name = 'leader'
+        self.p.set_role('primary')
+        self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.assertEqual(self.ha.run_cycle(), 'starting primary after failure')
+
     @patch.object(Rewind, 'ensure_clean_shutdown', Mock())
     def test_crash_recovery(self):
         self.ha.has_lock = true
@@ -574,6 +586,7 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'waiting for end of recovery after bootstrap')
         self.p.is_leader = true
+        self.ha.is_synchronous_mode = true
         self.assertEqual(self.ha.run_cycle(), 'running post_bootstrap')
         self.assertEqual(self.ha.run_cycle(), 'initialized a new cluster')
 
@@ -837,8 +850,6 @@ class TestHa(PostgresInit):
         self.ha.dcs._last_failsafe = None
         with patch.object(Watchdog, 'is_healthy', PropertyMock(return_value=False)):
             self.assertFalse(self.ha.is_healthiest_node())
-        with patch('patroni.postgresql.Postgresql.is_starting', return_value=True):
-            self.assertFalse(self.ha.is_healthiest_node())
         self.ha.is_paused = true
         self.assertFalse(self.ha.is_healthiest_node())
 

tests/test_patroni.py

@@ -10,6 +10,7 @@ from http.server import HTTPServer
 from mock import Mock, PropertyMock, patch
 from patroni.api import RestApiServer
 from patroni.async_executor import AsyncExecutor
+from patroni.dcs import Cluster, Member
 from patroni.dcs.etcd import AbstractEtcdClientWithFailover
 from patroni.exceptions import DCSError
 from patroni.postgresql import Postgresql
@@ -202,3 +203,36 @@ class TestPatroni(unittest.TestCase):
             self.assertRaises(SystemExit, check_psycopg)
         with patch('builtins.__import__', mock_import):
             self.assertRaises(SystemExit, check_psycopg)
+
+    def test_ensure_unique_name(self):
+        # None/empty cluster implies unique name
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=None)):
+            self.assertIsNone(self.p.ensure_unique_name())
+        empty_cluster = Cluster.empty()
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=empty_cluster)):
+            self.assertIsNone(self.p.ensure_unique_name())
+        without_members = empty_cluster._asdict()
+        del without_members['members']
+
+        # Cluster with members with different names implies unique name
+        okay_cluster = Cluster(
+            members=[Member(version=1, name="distinct", session=1, data={})],
+            **without_members
+        )
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=okay_cluster)):
+            self.assertIsNone(self.p.ensure_unique_name())
+
+        # Cluster with a member with the same name that is running
+        bad_cluster = Cluster(
+            members=[Member(version=1, name="postgresql0", session=1, data={
+                "api_url": "https://127.0.0.1:8008",
+            })],
+            **without_members
+        )
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=bad_cluster)):
+            # If the api of the running node cannot be reached, this implies unique name
+            with patch.object(self.p, 'request', Mock(side_effect=ConnectionError)):
+                self.assertIsNone(self.p.ensure_unique_name())
+            # Only if the api of the running node is reachable do we throw an error
+            with patch.object(self.p, 'request', Mock()):
+                self.assertRaises(SystemExit, self.p.ensure_unique_name)

tests/test_rewind.py

@@ -91,9 +91,10 @@ class TestRewind(BaseTestPostgresql):
                                              'Latest checkpoint location': '0/'})):
             self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
-        with patch.object(Postgresql, 'is_running', Mock(return_value=True)):
+        with patch.object(Postgresql, 'is_running', Mock(return_value=True)),\
-            with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None), Exception])):
+                patch.object(MockCursor, 'fetchone',
-                self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
+                             Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None, None, None), Exception])):
+            self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
     @patch.object(CancellableSubprocess, 'call', mock_cancellable_call)
     @patch.object(Postgresql, 'checkpoint', side_effect=['', '1'],)

tests/test_slots.py

@@ -77,14 +77,14 @@ class TestSlotsHandler(BaseTestPostgresql):
         with patch.object(Postgresql, '_query') as mock_query:
             self.p.reset_cluster_info_state(None)
             mock_query.return_value.fetchone.return_value = (
-                1, 0, 0, 0, 0, 0, 0, 0, 0,
+                1, 0, 0, 0, 0, 0, 0, 0, 0, None, None,
                 [{"slot_name": "ls", "type": "logical", "datoid": 5, "plugin": "b",
                   "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])
             self.assertEqual(self.p.slots(), {'ls': 12345})
 
             self.p.reset_cluster_info_state(None)
             mock_query.return_value.fetchone.return_value = (
-                1, 0, 0, 0, 0, 0, 0, 0, 0,
+                1, 0, 0, 0, 0, 0, 0, 0, 0, None, None,
                 [{"slot_name": "ls", "type": "logical", "datoid": 6, "plugin": "b",
                   "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])
             self.assertEqual(self.p.slots(), {})