diff --git a/patroni/__init__.py b/patroni/__init__.py index 78811752..045d0d59 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -18,12 +18,17 @@ class Patroni: def __init__(self, config): self.nap_time = config['loop_wait'] + self.tags = config.get('tags', dict()) self.postgresql = Postgresql(config['postgresql']) self.dcs = self.get_dcs(self.postgresql.name, config) self.api = RestApiServer(self, config['restapi']) self.ha = Ha(self) self.next_run = time.time() + @property + def nofailover(self): + return self.tags.get('nofailover', False) + @staticmethod def get_dcs(name, config): if 'etcd' in config: diff --git a/patroni/api.py b/patroni/api.py index 36319c48..466d7e32 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -60,6 +60,7 @@ class RestApiHandler(BaseHTTPRequestHandler): path = '/master' if self.path == '/' else self.path response = self.get_postgresql_status() + response.update(self.get_tags()) patroni = self.server.patroni cluster = patroni.dcs.cluster @@ -90,6 +91,7 @@ class RestApiHandler(BaseHTTPRequestHandler): def do_GET_patroni(self): response = self.get_postgresql_status(True) + response.update(self.get_tags()) self.send_response(200) self.send_header('Content-Type', 'application/json') @@ -160,8 +162,8 @@ class RestApiHandler(BaseHTTPRequestHandler): members = [m for m in cluster.members if m.name != cluster.leader.name and m.api_url] if not members: return b'failover is not possible: cluster does not have members except leader' - for member, reachable, in_recovery, xlog_location in self.server.patroni.ha.fetch_nodes_statuses(members): - if reachable: + for member, reachable, in_recovery, xlog_location, tags in self.server.patroni.ha.fetch_nodes_statuses(members): + if reachable and not tags.get('nofailover', False): return None return b'failover is not possible: no good candidates have been found' @@ -245,6 +247,9 @@ class RestApiHandler(BaseHTTPRequestHandler): state = 'unknown' if state == 'running' else state return {'state': state} + def get_tags(self): + return {'tags': self.server.patroni.tags} + class RestApiServer(ThreadingMixIn, HTTPServer, Thread): diff --git a/patroni/dcs.py b/patroni/dcs.py index bb56fecc..f640787c 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -63,6 +63,10 @@ class Member(namedtuple('Member', 'index,name,session,data')): def api_url(self): return self.data.get('api_url', None) + @property + def nofailover(self): + return self.data.get('tags', {}).get('nofailover', False) + class Leader(namedtuple('Leader', 'index,session,member')): diff --git a/patroni/ha.py b/patroni/ha.py index 19bf395a..3a8d46d7 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -51,7 +51,8 @@ class Ha: '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, + 'tags': self.patroni.tags } if data['state'] in ['running', 'restarting', 'starting']: try: @@ -73,7 +74,7 @@ class Ha: self._async_executor.schedule('bootstrap from leader') self._async_executor.run_async(self.copy_backup_from_leader, args=(self.cluster.leader, )) return 'trying to bootstrap from leader' - elif not self.cluster.initialize: # no initialize key + elif not self.cluster.initialize and not self.patroni.nofailover: # no initialize key if self.dcs.initialize(create_new=True): # race for initialization try: self.state_handler.bootstrap() @@ -142,7 +143,9 @@ class Ha: reachable - `!False` if the node is not reachable or is not responding with correct JSON in_recovery - `!True` if pg_is_in_recovery() == true - xlog_location - value of `replayed_location` or `location` from JSON, dependin on its role.""" + xlog_location - value of `replayed_location` or `location` from JSON, dependin on its role. + tags - dictionary with values of different tags (i.e. nofailover) + """ try: response = requests.get(member.api_url, timeout=2, verify=False) @@ -150,10 +153,11 @@ class Ha: json = response.json() is_master = json['role'] == 'master' xlog_location = json['xlog']['location' if is_master else 'replayed_location'] - return (member, True, not is_master, xlog_location) + tags = json.get('tags', dict()) + return (member, True, not is_master, xlog_location, tags) except: logging.exception('request failed: GET %s', member.api_url) - return (member, False, None, 0) + return (member, False, None, 0, {}) def fetch_nodes_statuses(self, members): pool = ThreadPool(len(members)) @@ -168,16 +172,19 @@ class Ha: if self.state_handler.is_leader(): return True + if self.patroni.nofailover is True: + return False + if check_replication_lag and not self.state_handler.check_replication_lag(self.cluster.last_leader_operation): return False # Too far behind last reported xlog location on master # Prepare list of nodes to run check against - members = [m for m in members if m.name != self.state_handler.name and m.api_url] + members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url] if members: my_xlog_location = self.state_handler.xlog_position() - for member, reachable, in_recovery, xlog_location in self.fetch_nodes_statuses(members): - if reachable: # If the node is unreachable it's not healhy + for member, reachable, in_recovery, xlog_location, tags in self.fetch_nodes_statuses(members): + if reachable and not tags.get('nofailover', False): # If the node is unreachable it's not healhy if not in_recovery: logger.warning('Master (%s) is still alive', member.name) return False @@ -187,13 +194,15 @@ class Ha: def is_failover_possible(self, members): ret = False - members = [m for m in members if m.name != self.state_handler.name and m.api_url] + members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url] if members: - for member, reachable, in_recovery, xlog_location in self.fetch_nodes_statuses(members): - if reachable: + for member, reachable, in_recovery, xlog_location, tags in self.fetch_nodes_statuses(members): + if reachable and not tags.get('nofailover', False): ret = True # TODO: check xlog_location - else: + elif not reachable: logger.info('Member %s is not reachable', member.name) + elif tags.get('nofailover', False): + logger.info('Member %s is not allowed to promote', member.name) else: logger.warning('manual failover: members list is empty') return ret @@ -207,12 +216,15 @@ class Ha: # find specific node and check that it is healthy members = [m for m in self.cluster.members if m.name == failover.member] if members: - member, reachable, in_recovery, xlog_location = self.fetch_node_status(members[0]) - if reachable: # node is healthy + member, reachable, in_recovery, xlog_location, tags = self.fetch_node_status(members[0]) + if reachable and not tags.get('nofailover', False): # node is healthy logger.info('manual failover: to %s, i am %s', member.name, self.state_handler.name) return False # we wanted to failover to specific member but it is not healthy - logger.warning('manual failover: member %s is unhealthy', member.name) + if not reachable: + logger.warning('manual failover: member %s is unhealthy', member.name) + elif tags.get('nofailover', False): + logger.warning('manual failover: member %s is not allowed to promote', member.name) # at this point we should consider all members as a candidates for failover # i.e. we assume that failover.member is None @@ -221,7 +233,7 @@ class Ha: if failover.leader: if self.state_handler.name == failover.leader: # I was the leader # exclude me and desired member which is unhealthy (failover.member can be None) - members = [m for m in self.cluster.members if m.name != failover.member] + members = [m for m in self.cluster.members if m.name not in (failover.member, failover.leader)] if self.is_failover_possible(members): # check that there are healthy members return False else: # I was the leader and it looks like currently I am the only healthy member @@ -234,6 +246,13 @@ class Ha: return self._is_healthiest_node(members, check_replication_lag=False) def is_healthiest_node(self): + + if self.state_handler.is_leader(): # leader is always the healthiest + return True + + if self.patroni.nofailover: # nofailover tag makes node always unhealthy + return False + if self.cluster.failover: return self.manual_failover_process_no_leader() @@ -282,6 +301,9 @@ class Ha: return self.follow_the_leader('demoted self due after trying and failing to obtain lock', 'following new leader after trying and failing to obtain lock') else: + if self.patroni.nofailover: + return self.follow_the_leader('demoting self because I am not allowed to become master', + 'following a different leader because I am not allowed to promote') return self.follow_the_leader('demoting self because i am not the healthiest node', 'following a different leader because i am not the healthiest node') diff --git a/postgres0.yml b/postgres0.yml index fa83765f..2ef74276 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -85,3 +85,8 @@ postgresql: max_replication_slots: 5 hot_standby: "on" wal_log_hints: "on" +tags: + nofailover: False + noloadbalance: False + clonefrom: False + replicatefrom: 127.0.0.1 diff --git a/postgres1.yml b/postgres1.yml index 34058ef7..2d28ccaa 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -85,3 +85,8 @@ postgresql: max_replication_slots: 5 hot_standby: "on" wal_log_hints: "on" +tags: + nofailover: False + noloadbalance: False + clonefrom: False + replicatefrom: 127.0.0.1 diff --git a/tests/test_api.py b/tests/test_api.py index 1fb06d6a..013e5065 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -42,7 +42,7 @@ class MockHa(Mock): return False def fetch_nodes_statuses(self, members): - return [[None, True, None, None]] + return [[None, True, None, None, {}]] class MockPatroni: @@ -50,6 +50,7 @@ class MockPatroni: postgresql = MockPostgresql() ha = MockHa() dcs = Mock() + tags = {} class MockRequest: diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 6ffa59af..915d5f9d 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -50,7 +50,7 @@ def requests_get(url, **kwargs): if url.startswith('http://local'): raise requests.exceptions.RequestException() elif ':8011/patroni' in url: - response.content = '{"role": "replica", "xlog": {"replayed_location": 0}}' + response.content = '{"role": "replica", "xlog": {"replayed_location": 0}, "tags": {}}' elif url.endswith('/members'): if url.startswith('http://error'): response.content = '[{}]' diff --git a/tests/test_ha.py b/tests/test_ha.py index e34f9b8e..43d4cded 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -82,6 +82,8 @@ class MockPatroni: self.postgresql = p self.dcs = d self.api = Mock() + self.tags = {} + self.nofailover = None self.api.connection_string = 'http://127.0.0.1:8008' @@ -272,6 +274,11 @@ class TestHa(unittest.TestCase): f = Failover(0, MockPostgresql.name, '') self.ha.cluster = get_cluster_initialized_with_leader(f) self.assertEquals(self.ha.run_cycle(), 'manual failover: demoting myself') + self.ha.fetch_node_status = lambda e: (e, True, True, 0, {'nofailover': 'True'}) + self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') + # manual failover from the previous leader to us won't happen if we hold the nofailover flag + self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', MockPostgresql.name)) + self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') @patch('requests.get', requests_get) def test_manual_failover_process_no_leader(self): @@ -280,24 +287,41 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader')) self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') - self.ha.fetch_node_status = lambda e: (e, True, True, 0) # accessible, in_recovery + self.ha.fetch_node_status = lambda e: (e, True, True, 0, {}) # accessible, in_recovery self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node') self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, MockPostgresql.name, '')) self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node') - self.ha.fetch_node_status = lambda e: (e, False, True, 0) # accessible, in_recovery + self.ha.fetch_node_status = lambda e: (e, False, True, 0, {}) # inaccessible, in_recovery self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') + # set failover flag to True for all members of the cluster + # this should elect the current member, as we are not going to call the API for it. + self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other')) + self.ha.fetch_node_status = lambda e: (e, True, True, 0, {'nofailover': 'True'}) # accessible, in_recovery + self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') + # same as previous, but set the current member to nofailover. In no case it should be elected as a leader + self.ha.patroni.nofailover = True + self.assertEquals(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote') + + def test_is_healthiest_node(self): + self.ha.state_handler.is_leader = false + self.ha.patroni.nofailover = False + self.ha.fetch_node_status = lambda e: (e, True, True, 0, {}) + self.assertTrue(self.ha.is_healthiest_node()) def test__is_healthiest_node(self): self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members)) self.p.is_leader = false - self.ha.fetch_node_status = lambda e: (e, True, True, 0) # accessible, in_recovery + self.ha.fetch_node_status = lambda e: (e, True, True, 0, {}) # accessible, in_recovery self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members)) - self.ha.fetch_node_status = lambda e: (e, True, False, 0) # accessible, not in_recovery + self.ha.fetch_node_status = lambda e: (e, True, False, 0, {}) # accessible, not in_recovery self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) - self.ha.fetch_node_status = lambda e: (e, True, True, 1) # accessible, in_recovery, xlog location ahead + self.ha.fetch_node_status = lambda e: (e, True, True, 1, {}) # accessible, in_recovery, xlog location ahead self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) self.p.check_replication_lag = false self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) + self.ha.patroni.nofailover = True + self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) + self.ha.patroni.nofailover = False @patch('requests.get', requests_get) def test_fetch_node_status(self): diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 52b10d7a..18f5c14b 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -74,3 +74,9 @@ class TestPatroni(unittest.TestCase): self.p.schedule_next_run() self.p.next_run = time.time() - self.p.nap_time - 1 self.p.schedule_next_run() + + def test_nofailover(self): + self.p.tags['nofailover'] = True + self.assertTrue(self.p.nofailover) + self.p.tags['nofailover'] = None + self.assertFalse(self.p.nofailover)