Merge branch 'etcd-cluster-support' of github.com:CyberDem0n/governor into etcd-cluster-support

Conflicts:
	requirements-py2.txt
	tests/test_governor.py

helpers/errors.py

@@ -9,3 +9,7 @@ class EtcdError(Exception):
 
 class CurrentLeaderError(EtcdError):
     pass
+
+
+class EtcdConnectionFailed(EtcdError):
+    pass

helpers/etcd.py

@@ -1,11 +1,14 @@
 import logging
 import requests
+import socket
 import sys
 
-from requests.exceptions import RequestException
 from collections import namedtuple
-from helpers.errors import CurrentLeaderError, EtcdError
+from dns.exception import DNSException
+from dns import resolver
+from helpers.errors import CurrentLeaderError, EtcdError, EtcdConnectionFailed
 from helpers.utils import sleep
+from requests.exceptions import RequestException
 
 if sys.hexversion >= 0x03000000:
     from urllib.parse import urlparse, urlunparse, parse_qsl
@@ -35,58 +38,186 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem
         return not (self.leader and self.leader.hostname)
 
 
+class Client:
+
+    API_VERSION = 'v2'
+
+    def __init__(self, config):
+        self._config = config
+        self.timeout = 5
+        self._base_uri = None
+        self._members_cache = []
+        self.load_members()
+
+    def client_url(self, path):
+        return self._base_uri + path
+
+    def _next_server(self):
+        self._base_uri = None
+        try:
+            self._base_uri = self._members_cache.pop()
+        except IndexError:
+            logger.error('Members cache is empty, can not retry.')
+            raise EtcdConnectionFailed('No more members in the cluster')
+        else:
+            logger.info('Selected new etcd server %s', self._base_uri)
+
+    def _get(self, path):
+        response = None
+        while response is None:
+            uri = self.client_url(path)
+            try:
+                logger.info('GET %s', uri)
+                response = requests.get(uri, timeout=self.timeout)
+            except RequestException:
+                self._next_server()
+
+        logger.debug([response.status_code, response.content])
+        try:
+            return response.json(), response.status_code
+        except (TypeError, ValueError):
+            raise EtcdError('Bad response from %s: %s' % (uri, response.content))
+
+    @staticmethod
+    def get_srv_record(host):
+        try:
+            return [(r.target.rstrip('.'), r.port) for r in resolver.query('_etcd-server._tcp.' + host, 'SRV')]
+        except DNSException:
+            logger.exception('Can not resolve SRV for %s', host)
+        return []
+
+    @staticmethod
+    def get_peers_urls_from_dns(host):
+        return ['http://{}:{}'.format(h, p) for h, p in Client.get_srv_record(host)]
+
+    @staticmethod
+    def get_client_urls_from_dns(addr):
+        host, port = addr.split(':')
+        ret = []
+        try:
+            for r in set(socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)):
+                ret.append('http://{}:{}/{}'.format(r[4][0], r[4][1], Client.API_VERSION))
+        except socket.error:
+            logger.exception('Can not resolve %s', host)
+        return list(set(ret)) if ret else ['http://{}:{}/{}'.format(host, port, Client.API_VERSION)]
+
+    def load_members(self):
+        load_from_srv = False
+        if not self._base_uri:
+            if 'discovery_srv' in self._config:
+                load_from_srv = True
+                self._members_cache = self.get_peers_urls_from_dns(self._config['discovery_srv'])
+            elif 'host' in self._config:
+                self._members_cache = self.get_client_urls_from_dns(self._config['host'])
+            else:
+                raise Exception('Neither discovery_srv nor host are defined in etcd section of config')
+            self._next_server()
+
+        response, status_code = self._get('/members')
+        if status_code != 200:
+            raise EtcdError('Got response with code=%s from %s' % (status_code, self._base_uri))
+
+        members_cache = []
+        for member in response if load_from_srv else response['members']:
+            members_cache.extend([m + '/' + self.API_VERSION for m in member['clientURLs']])
+        self._members_cache = list(set(members_cache))  # TODO: randomize
+        if load_from_srv:
+            self._next_server()
+        else:
+            try:
+                self._members_cache.remove(self._base_uri)
+            except ValueError:
+                pass
+
+    def get(self, path):
+        if not self._base_uri:
+            self.load_members()
+        old_base_uri = self._base_uri
+        try:
+            return self._get(path)
+        finally:
+            if self._base_uri != old_base_uri:
+                try:
+                    self.load_members()
+                except EtcdError:
+                    logger.exception('load_members')
+
+    def put(self, path, **data):
+        if not self._base_uri:
+            self.load_members()
+        old_base_uri = self._base_uri
+        response = None
+        while response is None:
+            uri = self.client_url(path)
+            try:
+                logger.info('PUT %s', uri)
+                response = requests.put(uri, timeout=self.timeout, data=data)
+            except RequestException:
+                logger.exception('PUT %s data=%s', uri, data)
+                self._next_server()
+
+        if self._base_uri != old_base_uri:
+            try:
+                self.load_members()
+            except EtcdError:
+                logger.exception('load_members')
+        return response.status_code in [200, 201, 202, 204]
+
+    def delete(self, path):
+        if not self._base_uri:
+            self.load_members()
+        old_base_uri = self._base_uri
+        response = None
+        while response is None:
+            uri = self.client_url(path)
+            try:
+                logger.info('DELETE %s', uri)
+                response = requests.delete(uri, timeout=self.timeout)
+            except RequestException:
+                logger.exception('DELETE %s', uri)
+                self._next_server()
+
+        if self._base_uri != old_base_uri:
+            try:
+                self.load_members()
+            except EtcdError:
+                logger.exception('load_members')
+        return response.status_code in [200, 202, 204]
+
+
 class Etcd:
 
     def __init__(self, config):
         self.ttl = config['ttl']
         self.member_ttl = config.get('member_ttl', 3600)
-        self.base_client_url = 'http://{host}/v2/keys/service/{scope}'.format(**config)
-        self.postgres_cluster = None
+        self._base_path = '/keys/service/' + config['scope']
+        self.client = self.get_etcd_client(config)
 
-    def get_client_path(self, path, max_attempts=1):
-        attempts = 0
-        response = None
-
-        while True:
-            ex = None
+    def get_etcd_client(self, config):
+        client = None
+        while not client:
             try:
-                response = requests.get(self.client_url(path))
-                if response.status_code == 200:
-                    break
-            except RequestException as e:
-                logger.exception('get_client_path')
-                ex = e
+                client = Client(config)
+            except EtcdError:
+                logger.info('waiting on etcd')
+                sleep(5)
+        return client
 
-            attempts += 1
-            if attempts < max_attempts:
-                logger.info('Failed to return %s, trying again. (%s of %s)', path, attempts, max_attempts)
-                sleep(3)
-            elif ex:
-                raise ex
-            else:
-                break
+    def client_path(self, path):
+        return self._base_path + path
 
-        return response.json(), response.status_code
+    def get_client_path(self, path):
+        return self.client.get(self.client_path(path))
 
     def put_client_path(self, path, **data):
-        try:
-            response = requests.put(self.client_url(path), data=data)
-            return response.status_code in [200, 201, 202, 204]
-        except RequestException:
-            logger.exception('PUT %s data=%s', path, data)
-        raise EtcdError('Etcd is not responding properly')
+        return self.client.put(self.client_path(path), **data)
 
     def delete_client_path(self, path):
         try:
-            response = requests.delete(self.client_url(path))
-            return response.status_code in [200, 202, 204]
-        except RequestException:
-            logger.exception('DELETE %s', path)
+            return self.client.delete(self.client_path(path))
+        except EtcdConnectionFailed:
             return False
 
-    def client_url(self, path):
-        return self.base_client_url + path
-
     @staticmethod
     def find_node(node, key):
         """

helpers/ha.py

@@ -42,7 +42,7 @@ class Ha:
                 self.state_handler.start()
                 if not has_lock:
                     return 'started as a secondary'
-                logging.info('started as readonly because i had the session lock')
+                logger.info('started as readonly because i had the session lock')
                 self.load_cluster_from_etcd()
 
             if self.cluster.is_unlocked():