Merge branches watch-leader-key and package-refactoring

patroni.py

@@ -1,125 +1,5 @@
 #!/usr/bin/env python
-import logging
+from patroni import main
-import os
-import sys
-import time
-import yaml
-
-from helpers.api import RestApiServer
-from helpers.etcd import Etcd
-from helpers.ha import Ha
-from helpers.postgresql import Postgresql
-from helpers.utils import setup_signal_handlers, sleep, reap_children
-from helpers.zookeeper import ZooKeeper
-
-logger = logging.getLogger(__name__)
-
-
-class Patroni:
-
-    def __init__(self, config):
-        self.nap_time = config['loop_wait']
-        self.postgresql = Postgresql(config['postgresql'])
-        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(name, config):
-        if 'etcd' in config:
-            return Etcd(name, config['etcd'])
-        if 'zookeeper' in config:
-            return ZooKeeper(name, config['zookeeper'])
-        raise Exception('Can not find sutable configuration of distributed configuration store')
-
-    def touch_member(self, ttl=None):
-        connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string
-        if self.ha.cluster:
-            for m in self.ha.cluster.members:
-                # 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(connection_string, ttl)
-
-    def initialize(self):
-        # wait for etcd to be available
-        while not self.touch_member():
-            logger.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.initialize()
-                self.ha.dcs.take_leader()
-                self.postgresql.start()
-                self.postgresql.create_replication_user()
-                self.postgresql.create_connection_users()
-            else:
-                while True:
-                    leader = self.ha.dcs.current_leader()
-                    if leader and self.postgresql.sync_from_leader(leader):
-                        self.postgresql.write_recovery_conf(leader)
-                        self.postgresql.start()
-                        break
-                    sleep(5)
-        elif self.postgresql.is_running():
-            self.postgresql.load_replication_slots()
-
-    def schedule_next_run(self):
-        if self.postgresql.is_promoted:
-            self.next_run = time.time()
-        self.next_run += self.nap_time
-        current_time = time.time()
-        nap_time = self.next_run - current_time
-        if nap_time <= 0:
-            self.next_run = current_time
-        else:
-            self.ha.dcs.watch(nap_time)
-
-    def run(self):
-        self.api.start()
-        self.next_run = time.time()
-
-        while True:
-            self.touch_member()
-            logger.info(self.ha.run_cycle())
-            try:
-                if self.ha.state_handler.is_leader():
-                    self.ha.cluster and self.ha.state_handler.create_replication_slots(self.ha.cluster)
-                else:
-                    self.ha.state_handler.drop_replication_slots()
-            except:
-                logger.exception('Exception when changing replication slots')
-            reap_children()
-            self.schedule_next_run()
-
-
-def main():
-    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
-    logging.getLogger('requests').setLevel(logging.WARNING)
-    setup_signal_handlers()
-
-    if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
-        print('Usage: {} config.yml'.format(sys.argv[0]))
-        return
-
-    with open(sys.argv[1], 'r') as f:
-        config = yaml.load(f)
-
-    patroni = Patroni(config)
-    try:
-        patroni.initialize()
-        patroni.run()
-    except KeyboardInterrupt:
-        pass
-    finally:
-        patroni.touch_member(patroni.shutdown_member_ttl)  # schedule member removal
-        patroni.postgresql.stop()
-        patroni.ha.dcs.delete_leader()
 
 
 if __name__ == '__main__':

patroni/__init__.py

@@ -0,0 +1,121 @@
+import logging
+import os
+import sys
+import time
+import yaml
+
+from patroni.api import RestApiServer
+from patroni.etcd import Etcd
+from patroni.ha import Ha
+from patroni.postgresql import Postgresql
+from patroni.utils import setup_signal_handlers, sleep, reap_children
+from patroni.zookeeper import ZooKeeper
+
+logger = logging.getLogger(__name__)
+
+
+class Patroni:
+
+    def __init__(self, config):
+        self.nap_time = config['loop_wait']
+        self.postgresql = Postgresql(config['postgresql'])
+        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(name, config):
+        if 'etcd' in config:
+            return Etcd(name, config['etcd'])
+        if 'zookeeper' in config:
+            return ZooKeeper(name, config['zookeeper'])
+        raise Exception('Can not find sutable configuration of distributed configuration store')
+
+    def touch_member(self, ttl=None):
+        connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string
+        if self.ha.cluster:
+            for m in self.ha.cluster.members:
+                # 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(connection_string, ttl)
+
+    def initialize(self):
+        # wait for etcd to be available
+        while not self.touch_member():
+            logger.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.initialize()
+                self.ha.dcs.take_leader()
+                self.postgresql.start()
+                self.postgresql.create_replication_user()
+                self.postgresql.create_connection_users()
+            else:
+                while True:
+                    leader = self.ha.dcs.current_leader()
+                    if leader and self.postgresql.sync_from_leader(leader):
+                        self.postgresql.write_recovery_conf(leader)
+                        self.postgresql.start()
+                        break
+                    sleep(5)
+        elif self.postgresql.is_running():
+            self.postgresql.load_replication_slots()
+
+    def schedule_next_run(self):
+        if self.postgresql.is_promoted:
+            self.next_run = time.time()
+        self.next_run += self.nap_time
+        current_time = time.time()
+        nap_time = self.next_run - current_time
+        if nap_time <= 0:
+            self.next_run = current_time
+        else:
+            self.ha.dcs.watch(nap_time)
+
+    def run(self):
+        self.api.start()
+        self.next_run = time.time()
+
+        while True:
+            self.touch_member()
+            logger.info(self.ha.run_cycle())
+            try:
+                if self.ha.state_handler.is_leader():
+                    self.ha.cluster and self.ha.state_handler.create_replication_slots(self.ha.cluster)
+                else:
+                    self.ha.state_handler.drop_replication_slots()
+            except:
+                logger.exception('Exception when changing replication slots')
+            reap_children()
+            self.schedule_next_run()
+
+
+def main():
+    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
+    logging.getLogger('requests').setLevel(logging.WARNING)
+    setup_signal_handlers()
+
+    if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
+        print('Usage: {} config.yml'.format(sys.argv[0]))
+        return
+
+    with open(sys.argv[1], 'r') as f:
+        config = yaml.load(f)
+
+    patroni = Patroni(config)
+    try:
+        patroni.initialize()
+        patroni.run()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        patroni.touch_member(patroni.shutdown_member_ttl)  # schedule member removal
+        patroni.postgresql.stop()
+        patroni.ha.dcs.delete_leader()

patroni/__main__.py

@@ -0,0 +1,5 @@
+from patroni import main
+
+
+if __name__ == '__main__':
+    main()

patroni/dcs.py

@@ -1,8 +1,8 @@
 import abc
 
 from collections import namedtuple
-from helpers import DCSError
+from patroni.exceptions import DCSError
-from helpers.utils import calculate_ttl, sleep
+from patroni.utils import calculate_ttl, sleep
 from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl
 
 
@@ -24,6 +24,7 @@ def parse_connection_string(value):
 
 
 class Member(namedtuple('Member', 'index,name,conn_url,api_url,expiration,ttl')):
+
     """Immutable object (namedtuple) which represents single member of PostgreSQL cluster.
     Consists of the following fields:
     :param index: modification index of a given member key in a Configuration Store
@@ -38,6 +39,7 @@ class Member(namedtuple('Member', 'index,name,conn_url,api_url,expiration,ttl'))
 
 
 class Leader(namedtuple('Leader', 'index,expiration,ttl,member')):
+
     """Immutable object (namedtuple) which represents leader key.
     Consists of the following fields:
     :param index: modification index of a leader key in a Configuration Store
@@ -55,6 +57,7 @@ class Leader(namedtuple('Leader', 'index,expiration,ttl,member')):
 
 
 class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')):
+
     """Immutable object (namedtuple) which represents PostgreSQL cluster.
     Consists of the following fields:
     :param initialize: boolean, shows whether this cluster has initialization key stored in DC or not.

patroni/etcd.py

@@ -10,8 +10,8 @@ import urllib3
 
 from dns.exception import DNSException
 from dns import resolver
-from helpers.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string
+from patroni.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string
-from helpers.utils import Retry, RetryFailedError, sleep
+from patroni.utils import Retry, RetryFailedError, sleep
 from requests.exceptions import RequestException
 
 logger = logging.getLogger(__name__)

patroni/exceptions.py

@@ -3,6 +3,7 @@ class PatroniException(Exception):
 
 
 class DCSError(PatroniException):
+
     """Parent class for all kind of exceptions related to selected distributed configuration store"""
 
     def __init__(self, value):

patroni/ha.py

@@ -1,6 +1,6 @@
 import logging
 
-from helpers.dcs import DCSError
+from patroni.dcs import DCSError
 from psycopg2 import InterfaceError, OperationalError
 
 logger = logging.getLogger(__name__)

patroni/postgresql.py

@@ -5,7 +5,7 @@ import shlex
 import shutil
 import subprocess
 
-from helpers.utils import sleep
+from patroni.utils import sleep
 from six.moves.urllib_parse import urlparse
 
 logger = logging.getLogger(__name__)

patroni/utils.py

@@ -6,7 +6,7 @@ import signal
 import sys
 import time
 
-from helpers import DCSError
+from patroni.exceptions import DCSError
 
 interrupted_sleep = False
 reap_children = False
@@ -87,10 +87,12 @@ def reap_children():
 
 
 class RetryFailedError(DCSError):
+
     """Raised when retrying an operation ultimately failed, after retrying the maximum number of attempts."""
 
 
 class Retry:
+
     """Helper for retrying a method in the face of retry-able exceptions"""
 
     def __init__(self, max_tries=1, delay=0.1, backoff=2, max_jitter=0.8, max_delay=3600,

patroni/version.py

@@ -0,0 +1 @@
+__version__ = '0.1'

patroni/zookeeper.py

@@ -3,10 +3,10 @@ import random
 import requests
 import time
 
-from helpers.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string
-from helpers.utils import sleep
 from kazoo.client import KazooClient, KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
+from patroni.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string
+from patroni.utils import sleep
 from requests.exceptions import RequestException
 
 logger = logging.getLogger(__name__)

setup.py

@@ -19,13 +19,20 @@ if sys.version_info < (2, 7, 0):
 __location__ = os.path.join(os.getcwd(), os.path.dirname(inspect.getfile(inspect.currentframe())))
 
 
+def read_version(package):
+    data = {}
+    with open(os.path.join(package, 'version.py'), 'r') as fd:
+        exec(fd.read(), data)
+    return data['__version__']
+
+
 NAME = 'patroni'
-MAIN_PACKAGE = 'patroni.py'
+MAIN_PACKAGE = NAME
-HELPERS = 'helpers'
 SCRIPTS = 'scripts'
+VERSION = read_version(MAIN_PACKAGE)
 VERSION = '0.1'
 DESCRIPTION = 'A Template for PostgreSQL HA with etcd'
-LICENSE = 'The MIT License'
+LICENSE = 'MIT License'
 
 COVERAGE_XML = True
 COVERAGE_HTML = False
@@ -62,8 +69,7 @@ class PyTest(TestCommand):
     def finalize_options(self):
         TestCommand.finalize_options(self)
         if self.cov_xml or self.cov_html:
-            self.cov = ['--cov', MAIN_PACKAGE, '--cov', HELPERS, '--cov', SCRIPTS, '--cov-report',
+            self.cov = ['--cov', MAIN_PACKAGE, '--cov', MAIN_PACKAGE, '--cov-report', 'term-missing']
-                        'term-missing']
             if self.cov_xml:
                 self.cov.extend(['--cov-report', 'xml'])
             if self.cov_html:
@@ -82,7 +88,7 @@ class PyTest(TestCommand):
             params['plugins'] = ['cov']
         if self.junitxml:
             params['args'] += self.junitxml
-        params['args'] += ['--doctest-modules', HELPERS, '--doctest-modules', SCRIPTS, '-s']
+        params['args'] += ['--doctest-modules', MAIN_PACKAGE, '-s', '-vv']
         errno = pytest.main(**params)
         sys.exit(errno)
 

tests/test_api.py

@@ -1,7 +1,7 @@
 import psycopg2
 import unittest
 
-from helpers.api import RestApiHandler, RestApiServer
+from patroni.api import RestApiHandler, RestApiServer
 from six import BytesIO as IO
 from test_postgresql import psycopg2_connect
 

tests/test_aws.py

@@ -2,7 +2,7 @@ import unittest
 import requests
 import boto.ec2
 from collections import namedtuple
-from scripts.aws import AWSConnection
+from patroni.scripts.aws import AWSConnection
 from requests.exceptions import RequestException
 
 

tests/test_etcd.py

@@ -9,9 +9,9 @@ import time
 import unittest
 
 from dns.exception import DNSException
-from helpers.dcs import Cluster, DCSError, Leader, Member
-from helpers.etcd import Client, Etcd
 from mock import Mock, patch
+from patroni.dcs import Cluster, DCSError, Leader, Member
+from patroni.etcd import Client, Etcd
 
 
 class MockResponse:

tests/test_ha.py

@@ -1,9 +1,9 @@
 import unittest
 
-from helpers.dcs import Cluster, DCSError
-from helpers.etcd import Client, Etcd
-from helpers.ha import Ha
 from mock import Mock, patch
+from patroni.dcs import Cluster, DCSError
+from patroni.etcd import Client, Etcd
+from patroni.ha import Ha
 from test_etcd import etcd_read, etcd_write
 
 

tests/test_patroni.py

@@ -1,5 +1,5 @@
 import datetime
-import helpers.zookeeper
+import patroni.zookeeper
 import psycopg2
 import subprocess
 import sys
@@ -7,14 +7,14 @@ import time
 import unittest
 import yaml
 
-from helpers.api import RestApiServer
-from helpers.dcs import Cluster, Member
-from helpers.etcd import Etcd
-from helpers.zookeeper import ZooKeeper
 from mock import Mock, patch
+from patroni.api import RestApiServer
+from patroni.dcs import Cluster, Member
+from patroni.etcd import Etcd
 from patroni import Patroni, main
+from patroni.zookeeper import ZooKeeper
 from six.moves import BaseHTTPServer
-from test_etcd import Client, etcd_read, etcd_write, etcd_watch
+from test_etcd import Client, etcd_read, etcd_write
 from test_ha import true, false
 from test_postgresql import Postgresql, subprocess_call, psycopg2_connect
 from test_zookeeper import MockKazooClient
@@ -74,7 +74,7 @@ class TestPatroni(unittest.TestCase):
         Postgresql.write_recovery_conf = self.write_recovery_conf
 
     def test_get_dcs(self):
-        helpers.zookeeper.KazooClient = MockKazooClient
+        patroni.zookeeper.KazooClient = MockKazooClient
         self.assertIsInstance(self.p.get_dcs('', {'zookeeper': {'scope': '', 'hosts': ''}}), ZooKeeper)
         self.assertRaises(Exception, self.p.get_dcs, '', {})
 

tests/test_postgresql.py

@@ -4,8 +4,8 @@ import shutil
 import subprocess
 import unittest
 
-from helpers.dcs import Cluster, Leader, Member
+from patroni.dcs import Cluster, Leader, Member
-from helpers.postgresql import Postgresql
+from patroni.postgresql import Postgresql
 
 
 def nop(*args, **kwargs):

tests/test_restore.py

@@ -1,7 +1,7 @@
 import unittest
 from mock import MagicMock, patch
 import os
-from scripts.restore import Restore, WALERestore
+from patroni.scripts.restore import Restore, WALERestore
 
 
 def fake_cursor_fetchone(*args, **kwargs):

tests/test_utils.py

@@ -2,8 +2,8 @@ import os
 import time
 import unittest
 
-from helpers import DCSError
+from patroni.exceptions import DCSError
-from helpers.utils import Retry, RetryFailedError, reap_children, sigchld_handler, sigterm_handler, sleep
+from patroni.utils import Retry, RetryFailedError, reap_children, sigchld_handler, sigterm_handler, sleep
 
 
 def nop(*args, **kwargs):
@@ -92,7 +92,7 @@ class TestRetrySleeper(unittest.TestCase):
             pass
 
         retry = self._makeOne(deadline=0.0001, sleep_func=sleep_func)
-        self.assertRaises(RetryFailedError, retry, self._fail(times=10))
+        self.assertRaises(RetryFailedError, retry, self._fail(times=100))
 
     def test_copy(self):
         _sleep = lambda t: None

tests/test_zookeeper.py

@@ -1,9 +1,9 @@
-import helpers.zookeeper
+import patroni.zookeeper
 import requests
 import unittest
 
-from helpers.dcs import Leader
+from patroni.dcs import Leader
-from helpers.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError
+from patroni.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError
 from kazoo.client import KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
 from kazoo.protocol.states import ZnodeStat
@@ -110,7 +110,7 @@ class TestExhibitorEnsembleProvider(unittest.TestCase):
 
     def set_up(self):
         requests.get = requests_get
-        helpers.zookeeper.sleep = exhibitor_sleep
+        patroni.zookeeper.sleep = exhibitor_sleep
 
     def test_init(self):
         self.assertRaises(SleepException, ExhibitorEnsembleProvider, ['localhost'], 8181)
@@ -124,7 +124,7 @@ class TestZooKeeper(unittest.TestCase):
 
     def set_up(self):
         requests.get = requests_get
-        helpers.zookeeper.KazooClient = MockKazooClient
+        patroni.zookeeper.KazooClient = MockKazooClient
         self.zk = ZooKeeper('foo', {'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, 'scope': 'test'})
 
     def test_session_listener(self):