Don't try to remove leader key when running ctl on the leader node (#302)

patroni/config.py

@@ -270,9 +270,7 @@ class Config(object):
 
         # restapi server expects to get restapi.auth = 'username:password'
         if 'authentication' in config['restapi']:
-            restapi = config['restapi']
-            auth = restapi['authentication']
-            restapi['auth'] = '{0}:{1}'.format(auth['username'], auth['password'])
+            config['restapi']['auth'] = '{username}:{password}'.format(**config['restapi']['authentication'])
 
         # special treatment for old config
 

patroni/ctl.py

@@ -106,7 +106,7 @@ def ctl(ctx):
 
 
 def get_dcs(config, scope):
-    config['scope'] = scope
+    config.update({'scope': scope, 'patronictl': True})
     config.setdefault('name', scope)
     try:
         return _get_dcs(config)

patroni/dcs/__init__.py

@@ -60,8 +60,8 @@ def get_dcs(config):
                 available_implementations.add(name)
                 if name in config:  # which has configuration section in the config file
                     # propagate some parameters
-                    config[name].update({p: config[p] for p in ('namespace', 'name', 'scope',
-                                         'loop_wait', 'ttl', 'retry_timeout') if p in config})
+                    config[name].update({p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait',
+                                         'patronictl', 'ttl', 'retry_timeout') if p in config})
                     return value(config[name])
     raise PatroniException("""Can not find suitable configuration of distributed configuration store
 Available implementations: """ + ', '.join(available_implementations))
@@ -272,6 +272,7 @@ class AbstractDCS(object):
         self._base_path = '/'.join([self._namespace, config['scope']])
         self._set_loop_wait(config.get('loop_wait', 10))
 
+        self._ctl = bool(config.get('patronictl', False))
         self._cluster = None
         self._cluster_thread_lock = Lock()
         self.event = Event()

patroni/dcs/consul.py

@@ -79,7 +79,8 @@ class Consul(AbstractDCS):
         self._client = ConsulClient(host=host, port=port)
         self._client.http.patch_default_timeout(config['retry_timeout']/2.0)
         self._scope = config['scope']
-        self.create_session()
+        if not self._ctl:
+            self.create_session()
         self.__do_not_watch = False
 
     def create_session(self):
@@ -154,7 +155,8 @@ class Consul(AbstractDCS):
 
             # get leader
             leader = nodes.get(self._LEADER)
-            if leader and leader['Value'] == self._name and self._session != leader.get('Session', 'x'):
+            if not self._ctl and leader and leader['Value'] == self._name \
+                    and self._session != leader.get('Session', 'x'):
                 logger.info('I am leader but not owner of the session. Removing leader node')
                 self._client.kv.delete(self.leader_path, cas=leader['ModifyIndex'])
                 leader = None

patroni/dcs/zookeeper.py

@@ -65,7 +65,6 @@ class ZooKeeper(AbstractDCS):
         self._client.start()
 
     def _kazoo_connect(self, host, port):
-
         """Kazoo is using Ping's to determine health of connection to zookeeper. If there is no
         response on Ping after Ping interval (1/2 from read_timeout) it will consider current
         connection dead and try to connect to another node. Without this "magic" it was taking
@@ -164,7 +163,8 @@ class ZooKeeper(AbstractDCS):
         leader = self.get_node(self.leader_path) if self._LEADER in nodes else None
         if leader:
             client_id = self._client.client_id
-            if leader[0] == self._name and client_id is not None and client_id[0] != leader[1].ephemeralOwner:
+            if not self._ctl and leader[0] == self._name and client_id is not None \
+                    and client_id[0] != leader[1].ephemeralOwner:
                 logger.info('I am leader but not owner of the session. Removing leader node')
                 self._client.delete(self.leader_path)
                 leader = None

tests/test_ha.py

@@ -12,6 +12,7 @@ from patroni.exceptions import DCSError, PostgresException
 from patroni.ha import Ha
 from patroni.postgresql import Postgresql
 from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get
+from test_postgresql import psycopg2_connect
 
 
 def true(*args, **kwargs):
@@ -115,6 +116,7 @@ def run_async(self, func, args=()):
 class TestHa(unittest.TestCase):
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
+    @patch('psycopg2.connect', psycopg2_connect)
     @patch.object(etcd.Client, 'read', etcd_read)
     def setUp(self):
         with patch.object(Client, 'machines') as mock_machines:

tests/test_postgresql.py

@@ -11,7 +11,6 @@ from patroni.exceptions import PostgresException, PostgresConnectionException
 from patroni.postgresql import Postgresql
 from patroni.utils import RetryFailedError
 from six.moves import builtins
-from test_ha import false
 
 
 class MockCursor(object):
@@ -225,7 +224,7 @@ class TestPostgresql(unittest.TestCase):
             self.assertTrue(self.p.stop())
 
     def test_restart(self):
-        self.p.start = false
+        self.p.start = Mock(return_value=False)
         self.assertFalse(self.p.restart())
         self.assertEquals(self.p.state, 'restart failed (restarting)')