Merge branch 'master' of github.com:zalando/patroni into feature/quorum-commit

features/patroni_api.feature

@@ -68,6 +68,7 @@ Scenario: check API requests for the primary-replica pair in the pause mode
 	When I kill postmaster on postgres1
 	And I issue a GET request to http://127.0.0.1:8009/replica
 	Then I receive a response code 503
+	And "members/postgres1" key in DCS has state=stopped after 10 seconds
 	When I run patronictl.py restart batman postgres1 --force
 	Then I receive a response returncode 0
 	Then replication works from postgres0 to postgres1 after 20 seconds
@@ -76,7 +77,7 @@ Scenario: check API requests for the primary-replica pair in the pause mode
 	Then I receive a response code 200
 	And I receive a response state running
 	And I receive a response role replica
-	When I run patronictl.py reinit batman postgres1 --force
+	When I run patronictl.py reinit batman postgres1 --force --wait
 	Then I receive a response returncode 0
 	And I receive a response output "Success: reinitialize for member postgres1"
 	And postgres1 role is the secondary after 30 seconds

patroni/__main__.py

@@ -106,6 +106,8 @@ class Patroni(AbstractPatroniDaemon, Tags):
 
     def ensure_unique_name(self) -> None:
         """A helper method to prevent splitbrain from operator naming error."""
+        from urllib.parse import urlparse
+        from urllib3.connection import HTTPConnection
         from patroni.dcs import Member
 
         cluster = self.dcs.get_cluster()
@@ -115,9 +117,12 @@ class Patroni(AbstractPatroniDaemon, Tags):
         if not isinstance(member, Member):
             return
         try:
-            _ = self.request(member, endpoint="/liveness", timeout=3)
-            logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
-            sys.exit(1)
+            parts = urlparse(member.api_url)
+            if isinstance(parts.hostname, str):
+                connection = HTTPConnection(parts.hostname, port=parts.port or 80, timeout=3)
+                connection.connect()
+                logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
+                sys.exit(1)
         except Exception:
             return
 

patroni/config.py

@@ -495,7 +495,8 @@ class Config(object):
             elif not is_local:
                 validator = ConfigHandler.CMDLINE_OPTIONS[name][1]
                 if validator(value):
-                    pg_params[name] = int(value) if isinstance(validator, IntValidator) else value
+                    int_val = parse_int(value) if isinstance(validator, IntValidator) else None
+                    pg_params[name] = int_val if isinstance(int_val, int) else value
                 else:
                     logger.warning("postgresql parameter %s=%s failed validation, defaulting to %s",
                                    name, value, ConfigHandler.CMDLINE_OPTIONS[name][0])

patroni/config_generator.py

@@ -287,7 +287,7 @@ class RunningClusterConfigGenerator(AbstractConfigGenerator):
 
         :param cur: connection cursor to use.
         """
-        cur.execute("SELECT name, current_setting(name) FROM pg_settings "
+        cur.execute("SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings "
                     "WHERE context <> 'internal' "
                     "AND source IN ('configuration file', 'command line', 'environment variable') "
                     "AND category <> 'Write-Ahead Log / Recovery Target' "

patroni/dcs/zookeeper.py

@@ -379,21 +379,28 @@ class ZooKeeper(AbstractDCS):
         cluster = self.cluster
         member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
         member_data = self.__last_member_data or member and member.data
-        #  We want to notify leader if some important fields in the member key changed by removing ZNode
-        if member and (self._client.client_id is not None and member.session != self._client.client_id[0]
-                       or not (member_data and deep_compare(member_data.get('tags', {}), data.get('tags', {}))
-                               and (member_data.get('state') == data.get('state')
-                                    or 'running' not in (member_data.get('state'), data.get('state')))
-                               and member_data.get('version') == data.get('version')
-                               and member_data.get('checkpoint_after_promote')
-                               == data.get('checkpoint_after_promote'))):
-            try:
-                self._client.delete_async(self.member_path).get(timeout=1)
-            except NoNodeError:
-                pass
-            except Exception:
-                return False
-            member = None
+        if member and member_data:
+            is_leader = data.get('role') in ('master', 'primary', 'standby_leader')
+            checkpoint_after_promote_changed = member_data.get('checkpoint_after_promote') \
+                != data.get('checkpoint_after_promote')
+            state_running_changed = member_data.get('state') != data.get('state') \
+                and 'running' in (member_data.get('state'), data.get('state'))
+            tags_changed = not deep_compare(member_data.get('tags', {}), data.get('tags', {}))
+
+            # We want delete the member ZNode if:
+            # - our session doesn't match with session id on our member key; or
+            # - we want to notify leader if some important fields in the member key changed; or
+            # - if we are the leader and want to notify replicas about checkpoint_after_promote;
+            if self._client.client_id is not None and member.session != self._client.client_id[0] \
+                    or is_leader and checkpoint_after_promote_changed \
+                    or not is_leader and (state_running_changed or tags_changed):
+                try:
+                    self._client.delete_async(self.member_path).get(timeout=1)
+                except NoNodeError:
+                    pass
+                except Exception:
+                    return False
+                member = None
 
         encoded_data = json.dumps(data, separators=(',', ':')).encode('utf-8')
         if member and member_data:

patroni/ha.py

@@ -789,6 +789,14 @@ class Ha(object):
         allow_promote = current_state.sync
         voters = CaseInsensitiveSet(sync.voters)
 
+        if picked == voters and voters != allow_promote:
+            logger.warning('Inconsistent state between synchronous_standby_names = %s and /sync = %s key '
+                           'detected, updating synchronous replication key...', list(allow_promote), list(voters))
+            sync = self.dcs.write_sync_state(self.state_handler.name, allow_promote, 0, version=sync.version)
+            if not sync:
+                return logger.warning("Updating sync state failed")
+            voters = CaseInsensitiveSet(sync.voters)
+
         if picked == voters:
             return
 

patroni/postgresql/__init__.py

@@ -119,17 +119,28 @@ class Postgresql(object):
         # Last known running process
         self._postmaster_proc = None
 
-        if self.is_running():  # we are "joining" already running postgres
-            self.set_state('running')
+        if self.is_running():
+            # If we found postmaster process we need to figure out whether postgres is accepting connections
+            self.set_state('starting')
+            self.check_startup_state_changed()
+
+        if self.state == 'running':  # we are "joining" already running postgres
+            # we know that PostgreSQL is accepting connections and can read some GUC's from pg_settings
+            self.config.load_current_server_parameters()
+
             self.set_role('master' if self.is_primary() else 'replica')
-            # postpone writing postgresql.conf for 12+ because recovery parameters are not yet known
-            if self.major_version < 120000 or self.is_primary():
-                self.config.write_postgresql_conf()
+
             hba_saved = self.config.replace_pg_hba()
             ident_saved = self.config.replace_pg_ident()
-            if hba_saved or ident_saved:
+
+            if self.major_version < 120000 or self.role in ('master', 'primary'):
+                # If PostgreSQL is running as a primary or we run PostgreSQL that is older than 12 we can
+                # call reload_config() once again (the first call happened in the ConfigHandler constructor),
+                # so that it can figure out if config files should be updated and pg_ctl reload executed.
+                self.config.reload_config(config, sighup=bool(hba_saved or ident_saved))
+            elif hba_saved or ident_saved:
                 self.reload()
-        elif self.role in ('master', 'primary'):
+        elif not self.is_running() and self.role in ('master', 'primary'):
             self.set_role('demoted')
 
     @property

patroni/postgresql/config.py

@@ -326,14 +326,22 @@ class ConfigHandler(object):
                                         .format(self._pgpass))
         self._passfile = None
         self._passfile_mtime = None
-        self._synchronous_standby_names = None
         self._postmaster_ctime = None
         self._current_recovery_params: Optional[CaseInsensitiveDict] = None
         self._config = {}
         self._recovery_params = CaseInsensitiveDict()
-        self._server_parameters: CaseInsensitiveDict
+        self._server_parameters: CaseInsensitiveDict = CaseInsensitiveDict()
         self.reload_config(config)
 
+    def load_current_server_parameters(self) -> None:
+        """Read GUC's values from ``pg_settings`` when Patroni is joining the the postgres that is already running."""
+        exclude = [name.lower() for name, value in self.CMDLINE_OPTIONS.items() if value[1] == _false_validator] \
+            + [name.lower() for name in self._RECOVERY_PARAMETERS]
+        self._server_parameters = CaseInsensitiveDict({r[0]: r[1] for r in self._postgresql.query(
+            "SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings"
+            " WHERE (source IN ('command line', 'environment variable') OR sourcefile = %s)"
+            " AND pg_catalog.lower(name) != ALL(%s)", self._postgresql_conf, exclude)})
+
     def setup_server_parameters(self) -> None:
         self._server_parameters = self.get_server_parameters(self._config)
         self._adjust_recovery_parameters()
@@ -922,14 +930,15 @@ class ConfigHandler(object):
         listen_addresses, port = split_host_port(config['listen'], 5432)
         parameters.update(cluster_name=self._postgresql.scope, listen_addresses=listen_addresses, port=str(port))
         if not self._postgresql.global_config or self._postgresql.global_config.is_synchronous_mode:
-            if self._synchronous_standby_names is None:
+            synchronous_standby_names = self._server_parameters.get('synchronous_standby_names')
+            if synchronous_standby_names is None:
                 if self._postgresql.global_config and self._postgresql.global_config.is_synchronous_mode_strict\
                         and self._postgresql.role in ('master', 'primary', 'promoted'):
                     parameters['synchronous_standby_names'] = '*'
                 else:
                     parameters.pop('synchronous_standby_names', None)
             else:
-                parameters['synchronous_standby_names'] = self._synchronous_standby_names
+                parameters['synchronous_standby_names'] = synchronous_standby_names
 
         # Handle hot_standby <-> replica rename
         if parameters.get('wal_level') == ('hot_standby' if self._postgresql.major_version >= 90600 else 'replica'):
@@ -1150,12 +1159,11 @@ class ConfigHandler(object):
     def set_synchronous_standby_names(self, value: Optional[str]) -> Optional[bool]:
         """Updates synchronous_standby_names and reloads if necessary.
         :returns: True if value was updated."""
-        if value != self._synchronous_standby_names:
+        if value != self._server_parameters.get('synchronous_standby_names'):
             if value is None:
                 self._server_parameters.pop('synchronous_standby_names', None)
             else:
                 self._server_parameters['synchronous_standby_names'] = value
-            self._synchronous_standby_names = value
             if self._postgresql.state == 'running':
                 self.write_postgresql_conf()
                 self._postgresql.reload()

tests/__init__.py

@@ -117,7 +117,7 @@ class MockCursor(object):
             self.results = [(False, 2)]
         elif sql.startswith('SELECT pg_catalog.pg_postmaster_start_time'):
             self.results = [(datetime.datetime.now(tzutc),)]
-        elif sql.startswith('SELECT name, current_setting(name) FROM pg_settings'):
+        elif sql.startswith('SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings'):
             self.results = [('data_directory', 'data'),
                             ('hba_file', os.path.join('data', 'pg_hba.conf')),
                             ('ident_file', os.path.join('data', 'pg_ident.conf')),
@@ -137,6 +137,11 @@ class MockCursor(object):
                             ('wal_block_size', '8192', None, 'integer', 'internal'),
                             ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'),
                             ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'),
+                            ('max_connections', '100', None, 'integer', 'postmaster'),
+                            ('max_prepared_transactions', '0', None, 'integer', 'postmaster'),
+                            ('max_worker_processes', '8', None, 'integer', 'postmaster'),
+                            ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'),
+                            ('max_wal_senders', '5', None, 'integer', 'postmaster'),
                             ('search_path', 'public', None, 'string', 'user'),
                             ('port', '5433', None, 'integer', 'postmaster'),
                             ('listen_addresses', '*', None, 'string', 'postmaster'),
@@ -248,6 +253,7 @@ class PostgresInit(unittest.TestCase):
 
 class BaseTestPostgresql(PostgresInit):
 
+    @patch('time.sleep', Mock())
     def setUp(self):
         super(BaseTestPostgresql, self).setUp()
 

tests/test_config.py

@@ -155,6 +155,7 @@ class TestConfig(unittest.TestCase):
         expected_params = {
             'f.oo': 'bar',  # not in ConfigHandler.CMDLINE_OPTIONS
             'max_connections': 100,  # IntValidator
+            'wal_keep_size': '128MB',  # IntValidator
             'wal_level': 'hot_standby',  # EnumValidator
         }
         input_params = deepcopy(expected_params)

tests/test_ha.py

@@ -1486,6 +1486,25 @@ class TestHa(PostgresInit):
             self.ha.run_cycle()
             self.assertEqual(mock_logger.call_args[0][0], 'Updating sync state failed')
 
+    @patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
+    def test_inconsistent_synchronous_state(self):
+        self.ha.is_synchronous_mode = true
+        self.ha.has_lock = true
+        self.p.name = 'leader'
+        self.ha.cluster = get_cluster_initialized_without_leader(sync=('leader', 'a'))
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, 0,
+                                                                         CaseInsensitiveSet(), CaseInsensitiveSet('a')))
+        self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
+        mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock()
+        with patch('patroni.ha.logger.warning') as mock_logger:
+            self.ha.run_cycle()
+            mock_set_sync.assert_called_once()
+            self.assertTrue(mock_logger.call_args_list[0][0][0].startswith('Inconsistent state between '))
+        self.ha.dcs.write_sync_state = Mock(return_value=None)
+        with patch('patroni.ha.logger.warning') as mock_logger:
+            self.ha.run_cycle()
+            self.assertEqual(mock_logger.call_args[0][0], 'Updating sync state failed')
+
     def test_effective_tags(self):
         self.ha._disable_sync = True
         self.assertEqual(self.ha.get_effective_tags(), {'foo': 'bar', 'nosync': True})

tests/test_patroni.py

@@ -40,7 +40,7 @@ class MockFrozenImporter(object):
 @patch('time.sleep', Mock())
 @patch('subprocess.call', Mock(return_value=0))
 @patch('patroni.psycopg.connect', psycopg_connect)
-@patch('urllib3.PoolManager.request', Mock(side_effect=Exception))
+@patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=Exception))
 @patch.object(ConfigHandler, 'append_pg_hba', Mock())
 @patch.object(ConfigHandler, 'write_postgresql_conf', Mock())
 @patch.object(ConfigHandler, 'write_recovery_conf', Mock())
@@ -64,7 +64,7 @@ class TestPatroni(unittest.TestCase):
             self.assertRaises(SystemExit, _main)
 
     @patch('pkgutil.iter_importers', Mock(return_value=[MockFrozenImporter()]))
-    @patch('urllib3.PoolManager.request', Mock(side_effect=Exception))
+    @patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=Exception))
     @patch('sys.frozen', Mock(return_value=True), create=True)
     @patch.object(HTTPServer, '__init__', Mock())
     @patch.object(etcd.Client, 'read', etcd_read)
@@ -108,6 +108,7 @@ class TestPatroni(unittest.TestCase):
     @patch('os.getpid')
     @patch('multiprocessing.Process')
     @patch('patroni.__main__.patroni_main', Mock())
+    @patch('sys.argv', ['patroni.py', 'postgres0.yml'])
     def test_patroni_main(self, mock_process, mock_getpid):
         mock_getpid.return_value = 2
         _main()
@@ -233,8 +234,8 @@ class TestPatroni(unittest.TestCase):
         )
         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)):
+            with patch('urllib3.connection.HTTPConnection.connect', 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()):
+            with patch('urllib3.connection.HTTPConnection.connect', Mock()):
                 self.assertRaises(SystemExit, self.p.ensure_unique_name)

tests/test_postgresql.py

@@ -721,6 +721,7 @@ class TestPostgresql(BaseTestPostgresql):
         self.assertEqual(self.p.get_primary_timeline(), 1)
 
     @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='replica'))
+    @patch.object(Postgresql, 'is_running', Mock(return_value=False))
     @patch.object(Bootstrap, 'running_custom_bootstrap', PropertyMock(return_value=True))
     @patch.object(Postgresql, 'controldata', Mock(return_value={'max_connections setting': '200',
                                                                 'max_worker_processes setting': '20',
@@ -964,6 +965,7 @@ class TestPostgresql2(BaseTestPostgresql):
     @patch('patroni.postgresql.CallbackExecutor', Mock())
     @patch.object(Postgresql, 'get_major_version', Mock(return_value=140000))
     @patch.object(Postgresql, 'is_running', Mock(return_value=True))
+    @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def setUp(self):
         super(TestPostgresql2, self).setUp()
 

tests/test_zookeeper.py

@@ -276,6 +276,7 @@ class TestZooKeeper(unittest.TestCase):
         self.assertTrue(self.zk.delete_cluster())
 
     def test_watch(self):
+        self.zk.event.wait = Mock()
         self.zk.watch(None, 0)
         self.zk.event.is_set = Mock(return_value=True)
         self.zk._fetch_status = False