diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index a345f45c..9be6edc1 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -139,10 +139,10 @@ PostgreSQL - **password**: replication password; 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 cluster restarts. - - **on\_role\_change**: run this script when the cluster is being promoted or demoted. - - **on\_start**: run this script when the cluster starts. - - **on\_stop**: run this script when the cluster stops. + - **on\_restart**: run this script when the postgres restarts (without changing role). + - **on\_role\_change**: run this script when the postgres is being promoted or demoted. + - **on\_start**: run this script when the postgres starts. + - **on\_stop**: run this script when the postgres stops. - **connect\_address**: IP address + port through which Postgres is accessible from other nodes and applications. - **create\_replica\_methods**: an ordered list of the create methods for turning a Patroni node into a new replica. "basebackup" is the default method; other methods are assumed to refer to scripts, each of which is configured as its diff --git a/features/environment.py b/features/environment.py index 0c020a77..718df86f 100644 --- a/features/environment.py +++ b/features/environment.py @@ -111,9 +111,9 @@ class PatroniController(AbstractController): with open(os.path.join(self._data_dir, 'label'), 'w') as f: f.write(content) - def read_label(self): + def read_label(self, label): try: - with open(os.path.join(self._data_dir, 'label'), 'r') as f: + with open(os.path.join(self._data_dir, label), 'r') as f: return f.read().strip() except IOError: return None diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature index bc6b3b7e..7e30d056 100644 --- a/features/standby_cluster.feature +++ b/features/standby_cluster.feature @@ -2,11 +2,11 @@ Feature: standby cluster Scenario: check permanent logical slots are preserved on failover/switchover Given I start postgres1 Then postgres1 is a leader after 10 seconds - And I sleep for 2 seconds + And I sleep for 3 seconds When I issue a PATCH request to http://127.0.0.1:8009/config with {"loop_wait": 2, "slots": {"pm_1": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}} Then I receive a response code 200 And Response on GET http://127.0.0.1:8009/config contains slots after 10 seconds - And I sleep for 2 seconds + And I sleep for 3 seconds When I issue a PATCH request to http://127.0.0.1:8009/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}} Then I receive a response code 200 When I start postgres0 with callback configured @@ -14,7 +14,7 @@ Feature: standby cluster And replication works from postgres1 to postgres0 after 15 seconds When I shut down postgres1 Then postgres0 is a leader after 10 seconds - And I sleep for 2 seconds + And "members/postgres0" key in DCS has role=master after 3 seconds When I issue a GET request to http://127.0.0.1:8008/ Then I receive a response code 200 And there is a label with "test_logical" in postgres0 data directory @@ -24,6 +24,12 @@ 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/master + Then I receive a response code 200 + When I issue a GET request to http://127.0.0.1:8009/standby_leader + Then I receive a response code 200 + And I receive a response role standby_leader + And there is a postgres1_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres1 data directory 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 @@ -31,4 +37,11 @@ Feature: standby cluster Scenario: check failover When I kill postgres1 And I kill postmaster on postgres1 - Then postgres2 is replicating from postgres0 after 20 seconds + Then postgres2 is replicating from postgres0 after 32 seconds + When I issue a GET request to http://127.0.0.1:8010/master + Then I receive a response code 200 + When I issue a GET request to http://127.0.0.1:8010/standby_leader + Then I receive a response code 200 + And I receive a response role standby_leader + And replication works from postgres0 to postgres2 after 15 seconds + And there is a postgres2_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres2 data directory diff --git a/features/steps/cascading_replication.py b/features/steps/cascading_replication.py index cea78f6a..35fb585b 100644 --- a/features/steps/cascading_replication.py +++ b/features/steps/cascading_replication.py @@ -9,9 +9,10 @@ def start_patroni_with_a_name_value_tag(context, name, tag_name, tag_value): return context.pctl.start(name, custom_config={'tags': {tag_name: tag_value}}) -@then('There is a label with "{content:w}" in {name:w} data directory') -def check_label(context, content, name): - label = context.pctl.read_label(name) +@then('There is a {label} with "{content}" in {name:w} data directory') +def check_label(context, label, content, name): + label = context.pctl.read_label(name, label) + label = label.replace('\n', '\\n') assert label == content, "{0} is not equal to {1}".format(label, content) diff --git a/features/steps/standby_cluster.py b/features/steps/standby_cluster.py index cdadcc62..0dc3a138 100644 --- a/features/steps/standby_cluster.py +++ b/features/steps/standby_cluster.py @@ -9,6 +9,8 @@ SELECT * FROM pg_catalog.pg_stat_replication WHERE application_name = '{0}' """ +callback = "bash -c 'echo \"${*: -3:1} ${*: -2:1} ${*: -1:1}\" >> data/$1/$1_cb.log' -- " + @step('I start {name:w} with callback configured') def start_patroni_with_callbacks(context, name): @@ -24,12 +26,15 @@ def start_patroni_with_callbacks(context, name): @step('I start {name:w} in a cluster {cluster_name:w}') def start_patroni(context, name, cluster_name): return context.pctl.start(name, custom_config={ - "scope": cluster_name + "scope": cluster_name, + "postgresql": { + "callbacks": {c: callback + name for c in ('on_start', 'on_stop', 'on_restart', 'on_role_change')} + } }) @step('I start {name:w} in a standby cluster {cluster_name:w} as a clone of {name2:w}') -def start_patroni_stanby_cluster(context, name, cluster_name, name2): +def start_patroni_standby_cluster(context, name, cluster_name, name2): # we need to remove patroni.dynamic.json in order to "bootstrap" standby cluster with existing PGDATA os.unlink(os.path.join(context.pctl._processes[name]._data_dir, 'patroni.dynamic.json')) port = context.pctl._processes[name2]._connkwargs.get('port') @@ -37,12 +42,18 @@ def start_patroni_stanby_cluster(context, name, cluster_name, name2): "scope": cluster_name, "bootstrap": { "dcs": { + "ttl": 20, + "loop_wait": 2, + "retry_timeout": 5, "standby_cluster": { "host": "localhost", "port": port, "primary_slot_name": "pm_1", } } + }, + "postgresql": { + "callbacks": {c: callback + name for c in ('on_start', 'on_stop', 'on_restart', 'on_role_change')} } }) return context.pctl.start(name) @@ -60,8 +71,8 @@ def check_replication_status(context, pg_name1, pg_name2, timeout): ) if cur and len(cur.fetchall()) != 0: - return True + break time.sleep(1) - - return False + else: + assert False, "{0} is not replicating from {1} after {2} seconds".format(pg_name1, pg_name2, timeout) diff --git a/patroni/api.py b/patroni/api.py index 08ffa176..e89d00c6 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -446,6 +446,9 @@ class RestApiHandler(BaseHTTPRequestHandler): }) } + if result['role'] == 'replica' and self.server.patroni.ha.is_standby_cluster(): + result['role'] = self.server.patroni.postgresql.role + if row[1] > 0: result['timeline'] = row[1] else: diff --git a/patroni/ha.py b/patroni/ha.py index dbeb7619..c242abb1 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -130,7 +130,7 @@ class Ha(object): self.old_cluster = cluster self.cluster = cluster - if self.cluster.is_unlocked() or self.cluster.leader.name != self.state_handler.name: + if not self.has_lock(False): self.set_is_leader(False) self._leader_timeline = None if cluster.is_unlocked() else cluster.leader.timeline @@ -154,9 +154,10 @@ class Ha(object): self.watchdog.keepalive() return ret - def has_lock(self): + def has_lock(self, info=True): lock_owner = self.cluster.leader and self.cluster.leader.name - logger.info('Lock owner: %s; I am %s', lock_owner, self.state_handler.name) + if info: + logger.info('Lock owner: %s; I am %s', lock_owner, self.state_handler.name) return lock_owner == self.state_handler.name def get_effective_tags(self): @@ -313,6 +314,7 @@ class Ha(object): self.load_cluster_from_dcs() + role = 'replica' if self.is_standby_cluster() or not self.has_lock(): if not self.state_handler.rewind_executed: self.state_handler.trigger_check_diverged_lsn() @@ -321,6 +323,7 @@ class Ha(object): if self.has_lock(): # in standby cluster msg = "starting as a standby leader because i had the session lock" + role = 'standby_leader' node_to_follow = self._get_node_to_follow(self.cluster) elif self.is_standby_cluster() and self.cluster.is_unlocked(): msg = "trying to follow a remote master because standby cluster is unhealthy" @@ -335,15 +338,13 @@ class Ha(object): self.recovering = True self._async_executor.schedule('restarting after failure') - self._async_executor.run_async(self.state_handler.follow, (node_to_follow, timeout)) + self._async_executor.run_async(self.state_handler.follow, (node_to_follow, role, timeout)) return msg def _get_node_to_follow(self, cluster): # determine the node to follow. If replicatefrom tag is set, # try to follow the node mentioned there, otherwise, follow the leader. - is_leader = self.cluster.leader and self.state_handler.name == self.cluster.leader.name - - if self.is_standby_cluster() and (is_leader or self.cluster.is_unlocked()): + if self.is_standby_cluster() and (self.cluster.is_unlocked() or self.has_lock(False)): node_to_follow = self.get_remote_master() elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name: node_to_follow = cluster.get_member(self.patroni.replicatefrom) @@ -375,9 +376,17 @@ class Ha(object): if self._handle_rewind_or_reinitialize(): return self._async_executor.scheduled_action - if not self.state_handler.check_recovery_conf(node_to_follow): + role = 'standby_leader' if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else 'replica' + # It might happen that leader key in the standby cluster references non-exiting member. + # In this case it is safe to continue running without changing recovery.conf + if self.is_standby_cluster() and role == 'replica' and not (node_to_follow and node_to_follow.conn_url): + return 'continue following the old known standby leader' + elif not self.state_handler.check_recovery_conf(node_to_follow): self._async_executor.schedule('changing primary_conninfo and restarting') - self._async_executor.run_async(self.state_handler.follow, (node_to_follow,)) + self._async_executor.run_async(self.state_handler.follow, (node_to_follow, role)) + elif role == 'standby_leader' and self.state_handler.role != role: + self.state_handler.set_role(role) + self.state_handler.call_nowait(ACTION_ON_ROLE_CHANGE) return follow_reason @@ -495,9 +504,7 @@ class Ha(object): self.dcs.set_history_value(json.dumps(history, separators=(',', ':'))) def enforce_follow_remote_master(self, message): - self.state_handler.set_role('standby_leader') demote_reason = 'cannot be a real master in standby cluster' - return self.follow(demote_reason, message) def enforce_master_role(self, message, promote_message): @@ -854,7 +861,7 @@ class Ha(object): # standby leader disappeared, and this is a healthiest # replica, so it should become a new standby leader. # This imply that we need to start following a remote master - msg = 'promoted self to a standby leader because i had the session lock' + msg = 'promoted self to a standby leader by acquiring session lock' return self.enforce_follow_remote_master(msg) else: return self.enforce_master_role( @@ -900,7 +907,9 @@ class Ha(object): # in case of standby cluster we don't really need to # enforce anything, since the leader is not a master. # So just remind the role. - msg = 'no action. i am the standby leader with the lock' + msg = 'no action. i am the standby leader with the lock' \ + if self.state_handler.role == 'standby_leader' else \ + 'promoted self to a standby leader because i had the session lock' return self.enforce_follow_remote_master(msg) else: return self.enforce_master_role( @@ -919,6 +928,9 @@ class Ha(object): return 'not promoting because failed to update leader lock in DCS' else: logger.info('does not have lock') + if self.is_standby_cluster(): + return self.follow('cannot be a real master in standby cluster', + 'no action. i am a secondary and i am following a standby leader', refresh=False) return self.follow('demoting self because i do not have the lock and i was a leader', 'no action. i am a secondary and i am following a leader', refresh=False) @@ -1050,7 +1062,7 @@ class Ha(object): if self.cluster.is_unlocked(): return 'Cluster has no leader, can not reinitialize' - if self.cluster.leader.name == self.state_handler.name: + if self.has_lock(False): return 'I am the leader, can not reinitialize' if force: @@ -1333,13 +1345,11 @@ class Ha(object): (" Leaving watchdog running." if self.watchdog.is_running else "")) def watch(self, timeout): - cluster = self.cluster # watch on leader key changes if the postgres is running and leader is known and current node is not lock owner - if not self._async_executor.busy and cluster and cluster.leader \ - and cluster.leader.name != self.state_handler.name: - leader_index = cluster.leader.index - else: + if self._async_executor.busy or self.cluster.is_unlocked() or self.has_lock(False): leader_index = None + else: + leader_index = self.cluster.leader.index return self.dcs.watch(leader_index, timeout) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index d487b926..6731faac 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -30,6 +30,7 @@ ACTION_ON_STOP = "on_stop" ACTION_ON_RESTART = "on_restart" ACTION_ON_RELOAD = "on_reload" ACTION_ON_ROLE_CHANGE = "on_role_change" +ACTION_NOOP = "noop" STATE_RUNNING = 'running' STATE_REJECT = 'rejecting connections' @@ -697,10 +698,11 @@ class Postgresql(object): # if basebackup succeeds, exit with success break else: - if not self.data_directory_empty() and not self.config.get(replica_method, {}).get('keep_data', False): - self.remove_data_directory() - else: - logger.info('Leaving data directory uncleaned') + if not self.data_directory_empty(): + if self.config.get(replica_method, {}).get('keep_data', False): + logger.info('Leaving data directory uncleaned') + else: + self.remove_data_directory() cmd = replica_method method_config = {} @@ -870,7 +872,7 @@ class Postgresql(object): self._pending_restart = True return effective_configuration - def start(self, timeout=None, block_callbacks=False, task=None): + def start(self, timeout=None, task=None, block_callbacks=False, role=None): """Start PostgreSQL Waits for postmaster to open ports or terminate so pg_isready can be used to check startup completion @@ -890,7 +892,7 @@ class Postgresql(object): if not block_callbacks: self.__cb_pending = ACTION_ON_START - self.set_role(self.get_postgres_role_from_data_directory()) + self.set_role(role or self.get_postgres_role_from_data_directory()) self.set_state('starting') self._pending_restart = False @@ -1091,7 +1093,7 @@ class Postgresql(object): return self.state == 'running' - def restart(self, timeout=None, task=None): + def restart(self, timeout=None, task=None, block_callbacks=False, role=None): """Restarts PostgreSQL. When timeout parameter is set the call will block either until PostgreSQL has started, failed to start or @@ -1100,8 +1102,9 @@ class Postgresql(object): :returns: True when restart was successful and timeout did not expire when waiting. """ self.set_state('restarting') - self.__cb_pending = ACTION_ON_RESTART - ret = self.stop(block_callbacks=True) and self.start(timeout=timeout, block_callbacks=True, task=task) + if not block_callbacks: + self.__cb_pending = ACTION_ON_RESTART + ret = self.stop(block_callbacks=True) and self.start(timeout, task, True, role) if not ret and not self.is_starting(): self.set_state('restart failed ({0})'.format(self.state)) return ret @@ -1436,7 +1439,7 @@ class Postgresql(object): def rewind_failed(self): return self._rewind_state == REWIND_STATUS.FAILED - def follow(self, member, timeout=None): + def follow(self, member, role='replica', timeout=None): is_remote_master = isinstance(member, RemoteMember) no_replication_slot = is_remote_master and member.no_replication_slot restore_command = is_remote_master and member.restore_command @@ -1444,7 +1447,8 @@ class Postgresql(object): archive_cleanup = is_remote_master and member.archive_cleanup_command primary_conninfo = self.primary_conninfo(member) - change_role = self.role in ('master', 'demoted') + change_role = self.cb_called and (self.role in ('master', 'demoted') or + not {'standby_leader', 'replica'} - {self.role, role}) recovery_params = self.config.get('recovery_conf', {}).copy() recovery_params.update({'standby_mode': 'on', 'recovery_target_timeline': 'latest'}) @@ -1463,11 +1467,17 @@ class Postgresql(object): self.write_recovery_conf(recovery_params) + # When we demoting the master or standby_leader to replica or promoting replica to a standby_leader + # and we know for sure that postgres was already running before, we will only execute on_role_change + # callback and prevent execution of on_restart/on_start callback. + # If the role remains the same (replica or standby_leader), we will execute on_start or on_restart + if change_role: + self.__cb_pending = ACTION_NOOP + if self.is_running(): - self.restart() + self.restart(block_callbacks=change_role, role=role) else: - self.start(timeout=timeout) - self.set_role('replica') + self.start(timeout=timeout, block_callbacks=change_role, role=role) if change_role: # TODO: postpone this until start completes, or maybe do even earlier @@ -1506,7 +1516,7 @@ class Postgresql(object): logger.exception('unable to restore configuration files from backup') def _wait_promote(self, wait_seconds): - for _ in polling_loop(wait_seconds - 1): + for _ in polling_loop(wait_seconds): data = self.controldata() if data.get('Database cluster state') == 'in production': return True diff --git a/tests/test_ha.py b/tests/test_ha.py index a03d78b8..282768c8 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -674,15 +674,17 @@ class TestHa(unittest.TestCase): self.p.is_leader = false self.p.name = 'leader' self.ha.cluster = get_standby_cluster_initialized_with_only_leader() - msg = 'no action. i am the standby leader with the lock' - self.assertEqual(self.ha.run_cycle(), msg) + self.p.check_recovery_conf = true + self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock') + self.assertEqual(self.ha.run_cycle(), 'no action. i am the standby leader with the lock') def test_process_healthy_standby_cluster_as_cascade_replica(self): self.p.is_leader = false self.p.name = 'replica' self.ha.cluster = get_standby_cluster_initialized_with_only_leader() - msg = 'no action. i am a secondary and i am following a leader' - self.assertEqual(self.ha.run_cycle(), msg) + self.assertEqual(self.ha.run_cycle(), 'no action. i am a secondary and i am following a standby leader') + with patch.object(Leader, 'conn_url', PropertyMock(return_value='')): + self.assertEqual(self.ha.run_cycle(), 'continue following the old known standby leader') @patch('requests.get', requests_get) def test_process_unhealthy_standby_cluster_as_standby_leader(self): @@ -692,8 +694,7 @@ class TestHa(unittest.TestCase): self.ha.cluster.is_unlocked = true self.ha.sysid_valid = true self.p._sysid = True - msg = 'promoted self to a standby leader because i had the session lock' - self.assertEqual(self.ha.run_cycle(), msg) + self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader by acquiring session lock') @patch.object(Postgresql, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)) @patch.object(Postgresql, 'can_rewind', PropertyMock(return_value=True)) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index c3ab1281..d2d061a9 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -118,6 +118,7 @@ class TestPatroni(unittest.TestCase): self.assertRaises(SystemExit, self.p.sigterm_handler) def test_schedule_next_run(self): + self.p.ha.cluster = Mock() self.p.ha.dcs.watch = Mock(return_value=True) self.p.schedule_next_run() self.p.next_run = time.time() - self.p.dcs.loop_wait - 1 diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index e42947fd..8a94d727 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -403,6 +403,7 @@ class TestPostgresql(unittest.TestCase): @patch.object(Postgresql, 'is_running', Mock(return_value=False)) @patch.object(Postgresql, 'start', Mock()) def test_follow(self): + self.p.call_nowait('on_start') m = RemoteMember('1', {'restore_command': '2', 'recovery_min_apply_delay': 3, 'archive_cleanup_command': '4'}) self.p.follow(m)