Merge pull request #80 from zalando/feature/nofailover

Feature/nofailover

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:

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):
 

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')):
 

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')
 

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

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

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:

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 = '[{}]'

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):

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)