From bba9066315f8dda23e0a28909790a5a3d48d1842 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 2 May 2019 14:07:26 +0200 Subject: [PATCH] Make it possible to run pg_rewind without superuser on pg11+ (#1035) * expose the current patroni version in DCS * expose `checkpoint_after_promote` flag in DCS as an indicator that pg_rewind could be safely executed * other nodes will wait until this flag is set instead of connecting as superuser and issuing the CHECKPOINT * define `postgresql.authention.rewind` with credentials for pg_rewind in patroni configuration files. * create user for pg_rewind if postgres is 11+ * grant execute on functions required for pg_rewind to rewind user --- docs/SETTINGS.rst | 3 +++ patroni/dcs/__init__.py | 13 ++++++++++ patroni/dcs/zookeeper.py | 5 +++- patroni/ha.py | 8 ++++++- patroni/postgresql.py | 51 +++++++++++++++++++++++++++++++++------- postgres0.yml | 14 ++++++----- postgres1.yml | 14 ++++++----- postgres2.yml | 14 ++++++----- tests/test_ha.py | 1 + tests/test_postgresql.py | 21 +++++++++++++---- 10 files changed, 111 insertions(+), 33 deletions(-) diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index a1280e28..de1844a3 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -137,6 +137,9 @@ PostgreSQL - **replication**: - **username**: replication username; the user will be created during initialization. Replicas will use this user to access master via streaming replication - **password**: replication password; the user will be created during initialization. + - **rewind**: + - **username**: name for the user for ``pg_rewind``; the user will be created during initialization of postgres 11+ and all necessary `permissions `__ will be granted. + - **password**: password for the user for ``pg_rewind``; the user will be created during initialization. - **callbacks**: callback scripts to run on certain actions. Patroni will pass the action, role and cluster name. (See scripts/aws.py as an example of how to write them.) - **on\_reload**: run this script when configuration reload is triggered. - **on\_restart**: run this script when the postgres restarts (without changing role). diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 050a78b2..e1f0d760 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -239,6 +239,19 @@ class Leader(namedtuple('Leader', 'index,session,member')): def timeline(self): return self.member.data.get('timeline') + @property + def checkpoint_after_promote(self): + """ + >>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote + """ + version = self.member.data.get('version') + if version: + try: + if tuple(map(int, version.split('.'))) >= (1, 5, 6): + return bool(self.member.data.get('checkpoint_after_promote')) + except Exception: + logger.debug('Failed to parse Patroni version %s', version) + class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')): diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index d71f70a5..33433d7a 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -274,7 +274,10 @@ class ZooKeeper(AbstractDCS): cluster = self.cluster member = cluster and cluster.get_member(self._name, fallback_to_leader=False) encoded_data = json.dumps(data, separators=(',', ':')).encode('utf-8') - if member and self._client.client_id is not None and member.session != self._client.client_id[0]: + if member and (self._client.client_id is not None and member.session != self._client.client_id[0] or + not (deep_compare(member.data.get('tags', {}), data.get('tags', {})) 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: diff --git a/patroni/ha.py b/patroni/ha.py index 826e9759..dfa9b405 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -174,12 +174,15 @@ class Ha(object): 'conn_url': self.state_handler.connection_string, 'api_url': self.patroni.api.connection_string, 'state': self.state_handler.state, - 'role': self.state_handler.role + 'role': self.state_handler.role, + 'version': self.patroni.version } # following two lines are mainly necessary for consul, to avoid creation of master service if data['role'] == 'master' and not self.is_leader(): data['role'] = 'promoted' + if self.is_leader(): + data['checkpoint_after_promote'] = self.state_handler.checkpoint_after_promote() tags = self.get_effective_tags() if tags: data['tags'] = tags @@ -904,6 +907,9 @@ class Ha(object): if msg is not None: return msg + # check if the node is ready to be used by pg_rewind + self.state_handler.check_for_checkpoint_after_promote() + if self.is_standby_cluster(): # in case of standby cluster we don't really need to # enforce anything, since the leader is not a master. diff --git a/patroni/postgresql.py b/patroni/postgresql.py index c6eac738..2c56a0f4 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -38,7 +38,8 @@ STATE_NO_RESPONSE = 'not responding' STATE_UNKNOWN = 'unknown' STOP_POLLING_INTERVAL = 1 -REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECK': 1, 'NEED': 2, 'NOT_NEED': 3, 'SUCCESS': 4, 'FAILED': 5}) +REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECKPOINT': 1, 'CHECK': 2, 'NEED': 3, + 'NOT_NEED': 4, 'SUCCESS': 5, 'FAILED': 6}) sync_standby_name_re = re.compile(r'^[A-Za-z_][A-Za-z_0-9\$]*$') cluster_info_query = ("SELECT CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 " @@ -202,6 +203,11 @@ class Postgresql(object): def _replication(self): return self.config['authentication']['replication'] + @property + def _rewind_credentials(self): + return self.config['authentication'].get('rewind', self._superuser) \ + if self._major_version >= 110000 else self._superuser + @property def callback(self): return self.config.get('callbacks') or {} @@ -1055,6 +1061,18 @@ class Postgresql(object): self.call_nowait(ACTION_ON_RELOAD) return ret + def check_for_checkpoint_after_promote(self): + if self._rewind_state == REWIND_STATUS.INITIAL and self.is_leader(): + try: + timeline = int(self.controldata().get("Latest checkpoint's TimeLineID")) + if self.get_master_timeline() == timeline: + self._rewind_state = REWIND_STATUS.CHECKPOINT + except (TypeError, ValueError): + logger.exception('Failed to parse timeline from pg_controldata output') + + def checkpoint_after_promote(self): + return self._rewind_state == REWIND_STATUS.CHECKPOINT + def check_for_startup(self): """Checks PostgreSQL status and returns if PostgreSQL is in the middle of startup.""" return self.is_starting() and not self.check_startup_state_changed() @@ -1361,7 +1379,7 @@ class Postgresql(object): if leader.member.data.get('role') != 'master': return # standby cluster - elif not self.check_leader_is_not_in_recovery(**leader.conn_kwargs(self._superuser)): + elif not self.check_leader_is_not_in_recovery(**leader.conn_kwargs(self._replication)): return history = need_rewind = None @@ -1425,14 +1443,21 @@ class Postgresql(object): return logger.warning('Can not run pg_rewind because postgres is still running') # prepare pg_rewind connection - r = leader.conn_kwargs(self._superuser) + r = leader.conn_kwargs(self._rewind_credentials) - # first make sure that we are really trying to rewind - # from the master and run a checkpoint on it in order to - # make it store the new timeline (5540277D.8020309@iki.fi) - leader_status = self.checkpoint(r) - if leader_status: - return logger.warning('Can not use %s for rewind: %s', leader.name, leader_status) + # 1. make sure that we are really trying to rewind from the master + # 2. make sure that pg_control contains the new timeline by: + # running a checkpoint or + # waiting until Patroni on the master will expose checkpoint_after_promote=True + checkpoint_status = leader.checkpoint_after_promote if isinstance(leader, Leader) else None + if checkpoint_status is None: # master still runs the old Patroni + leader_status = self.checkpoint(leader.conn_kwargs(self._superuser)) + if leader_status: + return logger.warning('Can not use %s for rewind: %s', leader.name, leader_status) + elif not checkpoint_status: + return logger.info('Waiting for checkpoint on %s before rewind', leader.name) + elif not self.check_leader_is_not_in_recovery(**r): + return if self.pg_rewind(r): self._rewind_state = REWIND_STATUS.SUCCESS @@ -1733,6 +1758,14 @@ END;$$""".format(name, ' '.join(options)) if task.result: self.create_or_update_role(self._replication['username'], self._replication.get('password'), ['REPLICATION']) + + if self._major_version >= 110000 and 'rewind' in self.config['authentication']: + rewind = self.config['authentication']['rewind'] + self.create_or_update_role(rewind['username'], rewind.get('password'), []) + for f in ('pg_ls_dir(text, boolean, boolean)', 'pg_stat_file(text, boolean)', + 'pg_read_binary_file(text)', 'pg_read_binary_file(text, bigint, bigint, boolean)'): + self.query('GRANT EXECUTE ON function pg_catalog.{0} TO "{1}"'.format(f, rewind['username'])) + for name, value in (config.get('users') or {}).items(): if name not in (self._superuser.get('username'), self._replication['username']): self.create_or_update_role(name, value.get('password'), value.get('options', [])) diff --git a/postgres0.yml b/postgres0.yml index fb5cf2d2..6b429ff0 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -74,12 +74,6 @@ bootstrap: - createdb postgresql: - # Fully qualified kerberos ticket file for the running user - # same as KRB5CCNAME used by the GSS -# krb_server_keyfile: /var/spool/keytabs/postgres - - # Server side kerberos spn -# krbsrvname: postgres listen: 127.0.0.1:5432 connect_address: 127.0.0.1:5432 data_dir: data/postgresql0 @@ -93,7 +87,15 @@ postgresql: superuser: username: postgres password: zalando + rewind: # Has no effect on postgres 10 and lower + username: rewind_user + password: rewind_password + # Server side kerberos spn +# krbsrvname: postgres parameters: + # Fully qualified kerberos ticket file for the running user + # same as KRB5CCNAME used by the GSS +# krb_server_keyfile: /var/spool/keytabs/postgres unix_socket_directories: '.' #watchdog: diff --git a/postgres1.yml b/postgres1.yml index a4b2ca1b..7c35a459 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -68,12 +68,6 @@ bootstrap: - createdb postgresql: - # Fully qualified kerberos ticket file for the running user - # same as KRB5CCNAME used by the GSS -# krb_server_keyfile: /var/spool/keytabs/postgres - - # Server side kerberos spn -# krbsrvname: postgres listen: 127.0.0.1:5433 connect_address: 127.0.0.1:5433 data_dir: data/postgresql1 @@ -87,7 +81,15 @@ postgresql: superuser: username: postgres password: zalando + rewind: # Has no effect on postgres 10 and lower + username: rewind_user + password: rewind_password + # Server side kerberos spn +# krbsrvname: postgres parameters: + # Fully qualified kerberos ticket file for the running user + # same as KRB5CCNAME used by the GSS +# krb_server_keyfile: /var/spool/keytabs/postgres unix_socket_directories: '.' basebackup: - verbose diff --git a/postgres2.yml b/postgres2.yml index 68e125c5..7187aac6 100644 --- a/postgres2.yml +++ b/postgres2.yml @@ -65,12 +65,6 @@ bootstrap: - createdb postgresql: - # Fully qualified kerberos ticket file for the running user - # same as KRB5CCNAME used by the GSS -# krb_server_keyfile: /var/spool/keytabs/postgres - - # Server side kerberos spn -# krbsrvname: postgres listen: 127.0.0.1:5434 connect_address: 127.0.0.1:5434 data_dir: data/postgresql2 @@ -84,7 +78,15 @@ postgresql: superuser: username: postgres password: zalando + rewind: # Has no effect on postgres 10 and lower + username: rewind_user + password: rewind_password + # Server side kerberos spn +# krbsrvname: postgres parameters: + # Fully qualified kerberos ticket file for the running user + # same as KRB5CCNAME used by the GSS +# krb_server_keyfile: /var/spool/keytabs/postgres unix_socket_directories: '.' tags: nofailover: false diff --git a/tests/test_ha.py b/tests/test_ha.py index ca28b926..9a74755c 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -119,6 +119,7 @@ zookeeper: self.config = Config() self.config.set_dynamic_configuration({'maximum_lag_on_failover': 5}) + self.version = '1.5.7' self.postgresql = p self.dcs = d self.api = Mock() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index d72072cd..85ef2f48 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -191,7 +191,7 @@ class TestPostgresql(unittest.TestCase): os.makedirs(self.data_dir) self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': self.data_dir, 'config_dir': self.config_dir, 'retry_timeout': 10, - 'pgpass': os.path.join(gettempdir(), 'pgpass0'), + 'krbsrvname': 'postgres', 'pgpass': os.path.join(gettempdir(), 'pgpass0'), 'listen': '127.0.0.2, 127.0.0.3:5432', 'connect_address': '127.0.0.2:5432', 'authentication': {'superuser': {'username': 'test', 'password': 'test'}, 'replication': {'username': 'replicator', 'password': 'rep-pass'}}, @@ -398,6 +398,14 @@ class TestPostgresql(unittest.TestCase): self.p.config['remove_data_directory_on_rewind_failure'] = False self.p.trigger_check_diverged_lsn() self.p.rewind(self.leader) + + self.leader.member.data.update(version='1.5.7', checkpoint_after_promote=False) + self.assertIsNone(self.p.rewind(self.leader)) + + self.leader.member.data['checkpoint_after_promote'] = True + with patch.object(Postgresql, 'check_leader_is_not_in_recovery', Mock(return_value=False)): + self.assertIsNone(self.p.rewind(self.leader)) + with patch.object(Postgresql, 'is_running', Mock(return_value=True)): self.p.rewind(self.leader) self.p.is_leader = Mock(return_value=False) @@ -663,7 +671,7 @@ class TestPostgresql(unittest.TestCase): @patch.object(Postgresql, 'run_bootstrap_post_init', Mock(return_value=True)) @patch.object(Postgresql, '_custom_bootstrap', Mock(return_value=True)) @patch.object(Postgresql, 'start', Mock(return_value=True)) - @patch.object(Postgresql, 'get_major_version', Mock(return_value=90600)) + @patch.object(Postgresql, 'get_major_version', Mock(return_value=110000)) def test_post_bootstrap(self): config = {'method': 'foo', 'foo': {'command': 'bar'}} self.p.bootstrap(config) @@ -686,7 +694,8 @@ class TestPostgresql(unittest.TestCase): self.p.bootstrap(config) self.p.set_state('stopped') self.p.reload_config({'authentication': {'superuser': {'username': 'p', 'password': 'p'}, - 'replication': {'username': 'r', 'password': 'r'}}, + 'replication': {'username': 'r', 'password': 'r'}, + 'rewind': {'username': 'rw', 'password': 'rw'}}, 'listen': '*', 'retry_timeout': 10, 'parameters': {'wal_level': '', 'hba_file': 'foo'}}) with patch.object(Postgresql, 'restart', Mock()) as mock_restart: self.p.post_bootstrap({}, task) @@ -787,7 +796,7 @@ class TestPostgresql(unittest.TestCase): parameters = self._PARAMETERS.copy() parameters.pop('f.oo') config = {'pg_hba': [''], 'pg_ident': [''], 'use_unix_socket': True, 'authentication': {}, - 'retry_timeout': 10, 'listen': '*', 'parameters': parameters} + 'retry_timeout': 10, 'listen': '*', 'krbsrvname': 'postgres', 'parameters': parameters} self.p.reload_config(config) parameters['b.ar'] = 'bar' self.p.reload_config(config) @@ -1040,3 +1049,7 @@ class TestPostgresql(unittest.TestCase): self.p.cancel() self.assertFalse(self.p.start()) self.assertTrue(self.p.pending_restart) + + @patch.object(Postgresql, 'controldata', Mock(return_value={"Latest checkpoint's TimeLineID": 1})) + def test_check_for_checkpoint_after_promote(self): + self.p.check_for_checkpoint_after_promote()