diff --git a/patroni/config.py b/patroni/config.py index aa80cf40..5cc5b3c6 100644 --- a/patroni/config.py +++ b/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 diff --git a/patroni/ctl.py b/patroni/ctl.py index 305cb8a8..dd876ac4 100644 --- a/patroni/ctl.py +++ b/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) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 32c3c8d2..9bcc7038 100644 --- a/patroni/dcs/__init__.py +++ b/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() diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 36a93536..84efa88a 100644 --- a/patroni/dcs/consul.py +++ b/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 diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 68f106a3..0db58ddd 100644 --- a/patroni/dcs/zookeeper.py +++ b/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 diff --git a/tests/test_ha.py b/tests/test_ha.py index da58eec4..61d65718 100644 --- a/tests/test_ha.py +++ b/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: diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 436c7d62..faf34f47 100644 --- a/tests/test_postgresql.py +++ b/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)')