Improve incomplete failover is a paused mode.

Instead of empying the stale failover key as a master and bailing
out, continue with the healthiest node evaluation. This should make
the actual master acquire the leader key faster. Emit the warning
message as well and add unit tests.

patroni/ha.py

@@ -238,10 +238,12 @@ class Ha(object):
                 return True
             elif self.is_paused():
                 # Remove failover key if the node to failover has terminated to avoid waiting for it indefinitely
-                # In order to avoid race conditions only the master is allowed to do so.
+                # In order to avoid attempts to delete this key from all nodes only the master is allowed to do it.
                 if (not self.cluster.get_member(failover.candidate, fallback_to_leader=False) and
-                   (self.state_handler.is_leader() or self.state_handler.role == 'master')):
+                   self.state_handler.is_leader()):
+                    logger.warning("manual failover: removing failover key because failover candidate is not running")
                     self.dcs.manual_failover('', '', index=self.cluster.failover.index)
+                    return None
                 return False
 
             # find specific node and check that it is healthy
@@ -281,7 +283,9 @@ class Ha(object):
     def is_healthiest_node(self):
         if self.is_paused() and not self.patroni.nofailover and \
                 self.cluster.failover and not self.cluster.failover.scheduled_at:
-            return self.manual_failover_process_no_leader()
+            ret = self.manual_failover_process_no_leader()
+            if ret is not None:  # continue if we just deleted the stale failover key as a master
+                return ret
 
         if self.state_handler.is_leader():  # leader is always the healthiest
             return True
@@ -377,6 +381,7 @@ class Ha(object):
 
     def process_unhealthy_cluster(self):
         """Cluster has no leader key"""
+
         if self.is_healthiest_node():
             if self.acquire_lock():
                 if self.cluster.failover:

tests/test_ha.py

@@ -394,6 +394,9 @@ class TestHa(unittest.TestCase):
         self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock')
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', '', None))
         self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock')
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'blabla', None))
+        self.p.is_leader = true
+        self.assertEquals('PAUSE: acquired session lock as a leader', self.ha.run_cycle())
 
     def test_is_healthiest_node(self):
         self.ha.state_handler.is_leader = false