Keep node name in AbstractDCS class

It eliminates need to pass this name into most of the methods

governor.py

@@ -17,16 +17,16 @@ class Governor:
     def __init__(self, config):
         self.nap_time = config['loop_wait']
         self.postgresql = Postgresql(config['postgresql'])
-        self.ha = Ha(self.postgresql, self.get_dcs(config))
+        self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config))
         host, port = config['restapi']['listen'].split(':')
         self.api = RestApiServer(self, config['restapi'])
         self.next_run = time.time()
         self.shutdown_member_ttl = 300
 
     @staticmethod
-    def get_dcs(config):
+    def get_dcs(name, config):
         if 'etcd' in config:
-            return Etcd(config['etcd'])
+            return Etcd(name, config['etcd'])
         raise Exception('Can not find sutable configuration of distributed configuration store')
 
     def touch_member(self, ttl=None):
@@ -36,20 +36,20 @@ class Governor:
                 # Do not update member TTL when it is far from being expired
                 if m.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl:
                     return True
-        return self.ha.dcs.touch_member(self.postgresql.name, connection_string, ttl)
+        return self.ha.dcs.touch_member(connection_string, ttl)
 
     def initialize(self):
         # wait for etcd to be available
         while not self.touch_member():
-            logging.info('waiting on etcd')
+            logging.info('waiting on DCS')
             sleep(5)
 
         # is data directory empty?
         if self.postgresql.data_directory_empty():
             # racing to initialize
-            if self.ha.dcs.race('/initialize', self.postgresql.name):
+            if self.ha.dcs.race('/initialize'):
                 self.postgresql.initialize()
-                self.ha.dcs.take_leader(self.postgresql.name)
+                self.ha.dcs.take_leader()
                 self.postgresql.start()
                 self.postgresql.create_replication_user()
             else:
@@ -70,7 +70,7 @@ class Governor:
         if nap_time <= 0:
             self.next_run = current_time
         else:
-            sleep(nap_time)
+            self.ha.dcs.sleep(nap_time)
 
     def run(self):
         self.api.start()
@@ -110,7 +110,7 @@ def main():
     finally:
         governor.touch_member(governor.shutdown_member_ttl)  # schedule member removal
         governor.postgresql.stop()
-        governor.ha.dcs.delete_leader(governor.postgresql.name)
+        governor.ha.dcs.delete_leader()
 
 
 if __name__ == '__main__':

helpers/dcs.py

@@ -1,7 +1,20 @@
 import abc
+import sys
 
 from collections import namedtuple
-from helpers.utils import calculate_ttl
+from helpers.utils import calculate_ttl, sleep
+
+if sys.hexversion >= 0x03000000:
+    from urllib.parse import urlparse, urlunparse, parse_qsl
+else:
+    from urlparse import urlparse, urlunparse, parse_qsl
+
+
+def parse_connection_string(value):
+    scheme, netloc, path, params, query, fragment = urlparse(value)
+    conn_url = urlunparse((scheme, netloc, path, params, '', fragment))
+    api_url = ([v for n, v in parse_qsl(query) if n == 'application_name'] or [None])[0]
+    return conn_url, api_url
 
 
 class DCSError(Exception):
@@ -10,6 +23,10 @@ class DCSError(Exception):
         self.value = value
 
     def __str__(self):
+        """
+        >>> str(DCSError('foo'))
+        "'foo'"
+        """
         return repr(self.value)
 
 
@@ -29,7 +46,8 @@ class AbstractDCS:
 
     __metaclass__ = abc.ABCMeta
 
-    def __init__(self, config):
+    def __init__(self, name, config):
+        self._name = name
         self._base_path = '/service/' + config['scope']
 
     def client_path(self, path):
@@ -37,15 +55,15 @@ class AbstractDCS:
 
     @abc.abstractmethod
     def get_cluster(self):
-        raise NotImplementedError
+        """get_cluster"""
 
     @abc.abstractmethod
     def update_leader(self, state_handler):
-        raise NotImplementedError
+        """update_leader"""
 
     @abc.abstractmethod
-    def attempt_to_acquire_leader(self, value):
-        raise NotImplementedError
+    def attempt_to_acquire_leader(self):
+        """attempt_to_acquire_leader"""
 
     def current_leader(self):
         try:
@@ -55,21 +73,20 @@ class AbstractDCS:
             return None
 
     @abc.abstractmethod
-    def touch_member(self, member, connection_string, ttl=None):
-        raise NotImplementedError
+    def touch_member(self, connection_string, ttl=None):
+        """touch_member"""
 
     @abc.abstractmethod
-    def take_leader(self, value):
-        raise NotImplementedError
+    def take_leader(self):
+        """take_leader"""
 
     @abc.abstractmethod
-    def race(self, path, value):
-        raise NotImplementedError
+    def race(self, path):
+        """race"""
 
     @abc.abstractmethod
-    def delete_member(self, member):
-        raise NotImplementedError
+    def delete_leader(self):
+        """delete_leader"""
 
-    @abc.abstractmethod
-    def delete_leader(self, value):
-        raise NotImplementedError
+    def sleep(self, timeout):
+        sleep(timeout)

helpers/etcd.py

@@ -2,19 +2,13 @@ import logging
 import random
 import requests
 import socket
-import sys
 
 from dns.exception import DNSException
 from dns import resolver
-from helpers.dcs import AbstractDCS, Cluster, DCSError, Member
+from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string
 from helpers.utils import sleep
 from requests.exceptions import RequestException
 
-if sys.hexversion >= 0x03000000:
-    from urllib.parse import urlparse, urlunparse, parse_qsl
-else:
-    from urlparse import urlparse, urlunparse, parse_qsl
-
 logger = logging.getLogger(__name__)
 
 
@@ -196,8 +190,8 @@ class Client:
 
 class Etcd(AbstractDCS):
 
-    def __init__(self, config):
-        super(Etcd, self).__init__(config)
+    def __init__(self, name, config):
+        super(Etcd, self).__init__(name, config)
         self.ttl = config['ttl']
         self.member_ttl = config.get('member_ttl', 3600)
         self.client = self.get_etcd_client(config)
@@ -243,9 +237,7 @@ class Etcd(AbstractDCS):
 
     @staticmethod
     def member(node):
-        scheme, netloc, path, params, query, fragment = urlparse(node['value'])
-        conn_url = urlunparse((scheme, netloc, path, params, '', fragment))
-        api_url = ([v for n, v in parse_qsl(query) if n == 'application_name'] or [None])[0]
+        conn_url, api_url = parse_connection_string(node['value'])
         expiration = node.get('expiration', None)
         ttl = node.get('ttl', None)
         return Member(node['modifiedIndex'], node['key'].split('/')[-1], conn_url, api_url, expiration, ttl)
@@ -287,21 +279,21 @@ class Etcd(AbstractDCS):
 
         raise EtcdError('Etcd is not responding properly')
 
-    def touch_member(self, member, connection_string, ttl=None):
+    def touch_member(self, connection_string, ttl=None):
         try:
-            return self.put_client_path('/members/' + member, value=connection_string, ttl=ttl or self.member_ttl)
+            return self.put_client_path('/members/' + self._name, value=connection_string, ttl=ttl or self.member_ttl)
         except EtcdError:
             return False
 
-    def take_leader(self, value):
+    def take_leader(self):
         try:
-            return self.put_client_path('/leader', value=value, ttl=self.ttl)
+            return self.put_client_path('/leader', value=self._name, ttl=self.ttl)
         except EtcdError:
             return False
 
-    def attempt_to_acquire_leader(self, value):
+    def attempt_to_acquire_leader(self):
         try:
-            ret = self.put_client_path('/leader', value=value, ttl=self.ttl, prevExist=False)
+            ret = self.put_client_path('/leader', value=self._name, ttl=self.ttl, prevExist=False)
             ret or logger.info('Could not take out TTL lock')
             return ret
         except EtcdError:
@@ -316,14 +308,11 @@ class Etcd(AbstractDCS):
             return True
         return False
 
-    def race(self, path, value):
+    def race(self, path):
         try:
-            return self.put_client_path(path, value=value, prevExist=False)
+            return self.put_client_path(path, value=self._name, prevExist=False)
         except EtcdError:
             return False
 
-    def delete_member(self, member):
-        return self.delete_client_path('/members/' + member)
-
-    def delete_leader(self, value):
-        return self.delete_client_path('/leader?prevValue=' + value)
+    def delete_leader(self):
+        return self.delete_client_path('/leader?prevValue=' + self._name)

helpers/ha.py

@@ -17,7 +17,7 @@ class Ha:
         self.cluster = self.dcs.get_cluster()
 
     def acquire_lock(self):
-        return self.dcs.attempt_to_acquire_leader(self.state_handler.name)
+        return self.dcs.attempt_to_acquire_leader()
 
     def update_lock(self):
         return self.dcs.update_leader(self.state_handler)

tests/test_etcd.py

@@ -8,7 +8,7 @@ import unittest
 
 from dns.exception import DNSException
 from helpers.dcs import Cluster, Member
-from helpers.etcd import Client, CurrentLeaderError, Etcd, EtcdConnectionFailed, EtcdError
+from helpers.etcd import Client, Etcd, EtcdConnectionFailed, EtcdError
 
 
 class MockResponse:
@@ -176,7 +176,7 @@ class TestEtcd(unittest.TestCase):
         requests.put = requests_put
         requests.delete = requests_delete
         time.sleep = time_sleep
-        self.etcd = Etcd({'ttl': 30, 'host': 'localhost:2379', 'scope': 'test'})
+        self.etcd = Etcd('foo', {'ttl': 30, 'host': 'localhost:2379', 'scope': 'test'})
 
     def test_get_etcd_client(self):
         time.sleep = time_sleep_exception
@@ -208,10 +208,10 @@ class TestEtcd(unittest.TestCase):
         self.assertFalse(self.etcd.touch_member('', ''))
 
     def test_take_leader(self):
-        self.assertFalse(self.etcd.take_leader(''))
+        self.assertFalse(self.etcd.take_leader())
 
     def test_attempt_to_acquire_leader(self):
-        self.assertFalse(self.etcd.attempt_to_acquire_leader(''))
+        self.assertFalse(self.etcd.attempt_to_acquire_leader())
 
     def test_update_leader(self):
         url = self.etcd.client._base_uri = self.etcd.client._base_uri.replace('local', 'remote')
@@ -220,7 +220,4 @@ class TestEtcd(unittest.TestCase):
         self.assertFalse(self.etcd.update_leader(MockPostgresql()))
 
     def test_race(self):
-        self.assertFalse(self.etcd.race('', ''))
-
-    def test_delete_member(self):
-        self.assertFalse(self.etcd.delete_member(''))
+        self.assertFalse(self.etcd.race(''))

tests/test_ha.py

@@ -71,7 +71,7 @@ class TestHa(unittest.TestCase):
         requests.put = requests_put
         requests.delete = requests_delete
         self.p = MockPostgresql()
-        self.e = Etcd({'ttl': 30, 'host': 'remotehost:2379', 'scope': 'test'})
+        self.e = Etcd('foo', {'ttl': 30, 'host': 'remotehost:2379', 'scope': 'test'})
         self.ha = Ha(self.p, self.e)
         self.ha.load_cluster_from_dcs()
         self.ha.cluster = Cluster(False, None, None, [])