From 74455905466377d79de5d6262a410edb8a6deea1 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 7 Jul 2015 12:26:52 +0200 Subject: [PATCH] Keep node name in AbstractDCS class It eliminates need to pass this name into most of the methods --- governor.py | 18 ++++++++-------- helpers/dcs.py | 51 ++++++++++++++++++++++++++++++---------------- helpers/etcd.py | 39 +++++++++++++---------------------- helpers/ha.py | 2 +- tests/test_etcd.py | 13 +++++------- tests/test_ha.py | 2 +- 6 files changed, 64 insertions(+), 61 deletions(-) diff --git a/governor.py b/governor.py index cfe1c546..ec4b39ad 100755 --- a/governor.py +++ b/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__': diff --git a/helpers/dcs.py b/helpers/dcs.py index 0b6c9332..a5e43488 100644 --- a/helpers/dcs.py +++ b/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) diff --git a/helpers/etcd.py b/helpers/etcd.py index 72c18e1c..1e971f30 100644 --- a/helpers/etcd.py +++ b/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) diff --git a/helpers/ha.py b/helpers/ha.py index 1ffa06ae..3d48505c 100644 --- a/helpers/ha.py +++ b/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) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index a5416c45..c0361a6f 100644 --- a/tests/test_etcd.py +++ b/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('')) diff --git a/tests/test_ha.py b/tests/test_ha.py index 299bc4b6..fcb010c4 100644 --- a/tests/test_ha.py +++ b/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, [])