diff --git a/patroni/api.py b/patroni/api.py index eea02cbb..71f5545a 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -87,7 +87,7 @@ class RestApiHandler(BaseHTTPRequestHandler): replica_status_code = 200 if not patroni.noloadbalance and response.get('role') == 'replica' else 503 status_code = 503 - if patroni.config.is_standby_cluster and ('standby_leader' in path or 'standby-leader' in path): + if patroni.ha.is_standby_cluster() and ('standby_leader' in path or 'standby-leader' in path): status_code = 200 if patroni.ha.is_leader() else 503 elif 'master' in path or 'leader' in path or 'primary' in path: status_code = 200 if patroni.ha.is_leader() else 503 diff --git a/patroni/config.py b/patroni/config.py index 44db51d3..4ccf6075 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -9,7 +9,7 @@ import yaml from collections import defaultdict from copy import deepcopy -from patroni.dcs import ClusterConfig, is_standby_cluster +from patroni.dcs import ClusterConfig from patroni.postgresql import Postgresql from patroni.utils import deep_compare, parse_bool, parse_int, patch_config from requests.structures import CaseInsensitiveDict @@ -99,10 +99,6 @@ class Config(object): def dynamic_configuration(self): return deepcopy(self._dynamic_configuration) - @property - def is_standby_cluster(self): - return is_standby_cluster(self._dynamic_configuration.get('standby_cluster')) - def check_mode(self, mode): return bool(parse_bool(self._dynamic_configuration.get(mode))) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 068ec279..b15e7b21 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -189,13 +189,12 @@ class RemoteMember(Member): 'create_replica_methods', 'restore_command', 'archive_cleanup_command', - 'recovery_min_apply_delay') + 'recovery_min_apply_delay', + 'no_replication_slot') def __getattr__(self, name): - if name not in RemoteMember.allowed_keys(): - return - - return self.data.get(name) + if name in RemoteMember.allowed_keys(): + return self.data.get(name) class Leader(namedtuple('Leader', 'index,session,member')): @@ -398,9 +397,6 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat def is_synchronous_mode(self): return self.check_mode('synchronous_mode') - def is_standby_cluster(self): - return is_standby_cluster(self.config and self.config.data.get('standby_cluster')) - @six.add_metaclass(abc.ABCMeta) class AbstractDCS(object): @@ -654,14 +650,3 @@ class AbstractDCS(object): self.event.wait(timeout) return self.event.isSet() - - -def is_standby_cluster(config): - """ Check whether or not provided configuration describes a standby cluster. - Config can be both patroni config or cluster.config.data - """ - return isinstance(config, dict) and ( - config.get('host') or - config.get('port') or - config.get('restore_command') - ) diff --git a/patroni/ha.py b/patroni/ha.py index e2f4d4d8..165b0aa2 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -91,6 +91,18 @@ class Ha(object): def is_paused(self): return self.check_mode('pause') + def get_standby_cluster_config(self): + if self.cluster and self.cluster.config and self.cluster.config.modify_index: + config = self.cluster.config.data + else: + config = self.patroni.config.dynamic_configuration + return config.get('standby_cluster') + + def is_standby_cluster(self): + config = self.get_standby_cluster_config() + # Check whether or not provided configuration describes a standby cluster + return isinstance(config, dict) and (config.get('host') or config.get('port') or config.get('restore_command')) + def is_leader(self): with self._is_leader_lock: return self._is_leader @@ -202,7 +214,7 @@ class Ha(object): self.state_handler.bootstrapping = True self._post_bootstrap_task = CriticalTask() - if self.patroni.config.is_standby_cluster: + if self.is_standby_cluster(): self._async_executor.schedule('bootstrap_standby_leader') self._async_executor.run_async(self.bootstrap_standby_leader) return 'trying to bootstrap a new standby leader' @@ -228,8 +240,7 @@ class Ha(object): not a real master, but a 'standby leader', that will take base backup from a remote master and start follow it. """ - patroni_config = self.patroni.config.dynamic_configuration - clone_source = self.get_remote_master(patroni_config) + clone_source = self.get_remote_master() msg = 'clone from remote master {0}'.format(clone_source.conn_url) result = self.clone(clone_source, msg) self._post_bootstrap_task.complete(result) @@ -239,9 +250,10 @@ class Ha(object): return result def _handle_rewind(self): - if self.state_handler.rewind_needed_and_possible(self.cluster.leader): - self._async_executor.schedule('running pg_rewind from ' + self.cluster.leader.name) - self._async_executor.run_async(self.state_handler.rewind, (self.cluster.leader,)) + leader = self.get_remote_master() if self.is_standby_cluster() else self.cluster.leader + if self.state_handler.rewind_needed_and_possible(leader): + self._async_executor.schedule('running pg_rewind from ' + leader.name) + self._async_executor.run_async(self.state_handler.rewind, (leader,)) return True def recover(self): @@ -273,16 +285,24 @@ class Ha(object): self.load_cluster_from_dcs() - if self.has_lock(): - msg = "starting as readonly because i had the session lock" - node_to_follow = None - else: + if self.is_standby_cluster() or not self.has_lock(): if not self.state_handler.rewind_executed: self.state_handler.trigger_check_diverged_lsn() if self._handle_rewind(): return self._async_executor.scheduled_action - msg = "starting as a secondary" - node_to_follow = self._get_node_to_follow(self.cluster) + + if self.has_lock(): # in standby cluster + msg = "starting as a standby leader because i had the session lock" + 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" + node_to_follow = self.get_remote_master() + else: + msg = "starting as a secondary" + node_to_follow = self._get_node_to_follow(self.cluster) + elif self.has_lock(): + msg = "starting as readonly because i had the session lock" + node_to_follow = None self.recovering = True @@ -295,8 +315,8 @@ class Ha(object): # 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.cluster.is_standby_cluster() and is_leader: - node_to_follow = self.get_remote_master(cluster.config.data) + if self.is_standby_cluster() and (is_leader or self.cluster.is_unlocked()): + 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) else: @@ -780,7 +800,7 @@ class Ha(object): self.dcs.manual_failover('', '') self.load_cluster_from_dcs() - if self.cluster.is_standby_cluster(): + if self.is_standby_cluster(): # 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 @@ -822,7 +842,7 @@ class Ha(object): if msg is not None: return msg - if self.cluster.is_standby_cluster(): + 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. # So just remind the role. @@ -1276,15 +1296,14 @@ class Ha(object): This usually happens on the master or if the node is running async action""" self.dcs.event.set() - def get_remote_master(self, config): + def get_remote_master(self): """ In case of standby cluster this will tel us from which remote master to stream. Config can be both patroni config or cluster.config.data """ - config = config or (self.config is not None and self.config.data) + cluster_params = self.get_standby_cluster_config() - if config and config.get('standby_cluster'): - cluster_params = config.get('standby_cluster') + if cluster_params: unique_name = 'remote_master:{}'.format(uuid.uuid1()) data = { 'conn_kwargs': { diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 1e86b958..8077e2b5 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -1214,7 +1214,7 @@ class Postgresql(object): ('user', r.get('user')), ('host', r.get('host')), ('port', r.get('port')), - ('dbname', r.get('database')), + ('dbname', r.get('database') or self._database), ('sslmode', 'prefer'), ('sslcompression', '1'), ] diff --git a/tests/test_api.py b/tests/test_api.py index e164b60e..b60a2cc0 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -95,6 +95,10 @@ class MockHa(object): def is_paused(): return True + @staticmethod + def is_standby_cluster(): + return False + class MockPatroni(object): @@ -167,8 +171,8 @@ class TestRestApiHandler(unittest.TestCase): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /master')) with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, '')])): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni')) - MockPatroni.config.is_standby_cluster = PropertyMock(return_value=True) - MockRestApiServer(RestApiHandler, 'GET /standby_leader') + with patch.object(MockHa, 'is_standby_cluster', Mock(return_value=True)): + MockRestApiServer(RestApiHandler, 'GET /standby_leader') def test_do_OPTIONS(self): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'OPTIONS / HTTP/1.0')) diff --git a/tests/test_ha.py b/tests/test_ha.py index ce8834a6..af1ef62b 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -5,7 +5,6 @@ import unittest import sys from mock import Mock, MagicMock, PropertyMock, patch -from patroni.async_executor import CriticalTask from patroni.config import Config from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState, TimelineHistory from patroni.dcs.etcd import Client @@ -62,17 +61,6 @@ def get_cluster_initialized_with_only_leader(failover=None, cluster_config=None) return get_cluster(True, leader, [leader], failover, None, cluster_config) -def get_cluster_not_initialized_standby(failover=None, sync=None): - return get_cluster_not_initialized_without_leader( - cluster_config=ClusterConfig(1, { - "standby_cluster": { - "host": "localhost", - "port": 5432, - "primary_slot_name": "", - }}, 1) - ) - - def get_standby_cluster_initialized_with_only_leader(failover=None, sync=None): return get_cluster_initialized_with_only_leader( cluster_config=ClusterConfig(1, { @@ -214,17 +202,10 @@ class TestHa(unittest.TestCase): @patch('patroni.dcs.etcd.Etcd.initialize', return_value=True) def test_start_as_standby_leader(self, initialize): self.p.data_directory_empty = true - self.ha.cluster = get_cluster_not_initialized_standby() + self.ha.cluster = get_cluster_not_initialized_without_leader(cluster_config=ClusterConfig(0, {}, 0)) self.ha.cluster.is_unlocked = true - self.ha.patroni.config._dynamic_configuration = {"standby_cluster": { - "host": "localhost", - "port": 5432, - "primary_slot_name": "", - }} - self.assertEqual( - self.ha.run_cycle(), - 'trying to bootstrap a new standby leader' - ) + self.ha.patroni.config._dynamic_configuration = {"standby_cluster": {"port": 5432}} + self.assertEqual(self.ha.run_cycle(), 'trying to bootstrap a new standby leader') @patch.object(Cluster, 'get_clone_member', Mock(return_value=Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni'}))) @@ -233,27 +214,7 @@ class TestHa(unittest.TestCase): self.p.data_directory_empty = true self.ha.cluster = get_standby_cluster_initialized_with_only_leader() self.ha.cluster.is_unlocked = false - self.ha.patroni.config._dynamic_configuration = {"standby_cluster": { - "host": "localhost", - "port": 5432, - "primary_slot_name": "", - }} - self.assertEqual( - self.ha.run_cycle(), - "trying to bootstrap from replica 'test'" - ) - - @patch.object(Postgresql, 'create_replica', Mock(return_value=0)) - def test_bootstrap_standby_leader(self): - self.ha.cluster = get_cluster_not_initialized_standby() - self.ha.cluster.is_unlocked = true - self.ha.patroni.config._dynamic_configuration = {"standby_cluster": { - "host": "localhost", - "port": 5432, - "primary_slot_name": "", - }} - self.ha._post_bootstrap_task = CriticalTask() - self.assertEqual(self.ha.bootstrap_standby_leader(), True) + self.assertEqual(self.ha.run_cycle(), "trying to bootstrap from replica 'test'") def test_recover_replica_failed(self): self.p.controldata = lambda: {'Database cluster state': 'in recovery', 'Database system identifier': SYSID} @@ -692,11 +653,6 @@ class TestHa(unittest.TestCase): def test_process_healthy_standby_cluster_as_standby_leader(self): self.p.is_leader = false self.p.name = 'leader' - self.ha.patroni.config._dynamic_configuration = {"standby_cluster": { - "host": "localhost", - "port": 5432, - "primary_slot_name": "", - }} 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) @@ -704,24 +660,13 @@ class TestHa(unittest.TestCase): def test_process_healthy_standby_cluster_as_cascade_replica(self): self.p.is_leader = false self.p.name = 'replica' - self.ha.patroni.config._dynamic_configuration = {"standby_cluster": { - "host": "localhost", - "port": 5432, - "primary_slot_name": "", - }} 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) - @patch('patroni.dcs.etcd.Etcd.initialize', return_value=True) - def test_process_unhealthy_standby_cluster_as_standby_leader(self, initialize): + def test_process_unhealthy_standby_cluster_as_standby_leader(self): self.p.is_leader = false self.p.name = 'leader' - self.ha.patroni.config._dynamic_configuration = {"standby_cluster": { - "host": "localhost", - "port": 5432, - "primary_slot_name": "", - }} self.ha.cluster = get_standby_cluster_initialized_with_only_leader() self.ha.cluster.is_unlocked = true self.ha.sysid_valid = true @@ -730,19 +675,30 @@ class TestHa(unittest.TestCase): self.assertEqual(self.ha.run_cycle(), msg) @patch.object(Postgresql, 'rewind_needed_and_possible', Mock(return_value=True)) - @patch('patroni.dcs.etcd.Etcd.initialize', return_value=True) - def test_process_unhealthy_standby_cluster_as_cascade_replica(self, initialize): + def test_process_unhealthy_standby_cluster_as_cascade_replica(self): self.p.is_leader = false self.p.name = 'replica' - self.ha.patroni.config._dynamic_configuration = {"standby_cluster": { - "host": "localhost", - "port": 5432, - "primary_slot_name": "", - }} self.ha.cluster = get_standby_cluster_initialized_with_only_leader() self.ha.is_unlocked = true - msg = 'running pg_rewind from leader' - self.assertEqual(self.ha.run_cycle(), msg) + self.assertTrue(self.ha.run_cycle().startswith('running pg_rewind from remote_master:')) + + def test_recover_unhealthy_leader_in_standby_cluster(self): + self.p.is_leader = false + self.p.name = 'leader' + self.p.is_running = false + self.p.follow = false + self.ha.cluster = get_standby_cluster_initialized_with_only_leader() + self.assertEqual(self.ha.run_cycle(), 'starting as a standby leader because i had the session lock') + + def test_recover_unhealthy_unlocked_standby_cluster(self): + self.p.is_leader = false + self.p.name = 'leader' + self.p.is_running = false + self.p.follow = false + self.ha.cluster = get_standby_cluster_initialized_with_only_leader() + self.ha.cluster.is_unlocked = true + self.ha.has_lock = false + self.assertEqual(self.ha.run_cycle(), 'trying to follow a remote master because standby cluster is unhealthy') def test_failed_to_update_lock_in_pause(self): self.ha.update_lock = false