Merge branch 'master' of github.com:CyberDem0n/governor

.travis.yml

@@ -0,0 +1,12 @@
+language: python
+python:
+  - "2.7"
+  - "3.3"
+  - "3.4"
+install:
+  - pip install -r requirements.txt
+  - pip install coveralls
+script:
+  - python setup.py test
+after_success:
+  - coveralls

helpers/postgresql.py

@@ -87,12 +87,9 @@ class Postgresql:
         return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == []
 
     def initialize(self):
-        if os.system(self._pg_ctl + ' initdb -o --encoding=UTF8') == 0:
-            self.write_pg_hba()
-
-            return True
-
-        return False
+        ret = os.system(self._pg_ctl + ' initdb -o --encoding=UTF8') == 0
+        ret and self.write_pg_hba()
+        return ret
 
     def sync_from_leader(self, leader):
         r = parseurl(leader.address)

setup.py

@@ -0,0 +1,161 @@
+#!/usr/bin/env python
+
+"""
+    Setup file for governor
+"""
+
+import sys
+import os
+import inspect
+
+import setuptools
+from setuptools.command.test import test as TestCommand
+from setuptools import setup
+
+if sys.version_info < (2, 6, 0):
+    sys.stderr.write('FATAL: governor needs to be run with Python 2.6+\n')
+    sys.exit(1)
+
+__location__ = os.path.join(os.getcwd(), os.path.dirname(inspect.getfile(inspect.currentframe())))
+
+
+NAME = 'governor'
+MAIN_PACKAGE = 'governor.py'
+HELPERS = 'helpers'
+VERSION = '0.1'
+DESCRIPTION = 'A Template for PostgreSQL HA with etcd'
+LICENSE = 'The MIT License'
+URL = 'https://github.com/zalando/governor'
+AUTHOR = 'Alexander Kukushkin'
+EMAIL = '[email protected]'
+
+COVERAGE_XML = True
+COVERAGE_HTML = False
+JUNIT_XML = True
+
+# Add here all kinds of additional classifiers as defined under
+# https://pypi.python.org/pypi?%3Aaction=list_classifiers
+CLASSIFIERS = [
+    'Development Status :: 4 - Beta',
+    'Environment :: Console',
+    'Intended Audience :: Developers',
+    'Intended Audience :: System Administrators',
+    'License :: OSI Approved :: The MIT License',
+    'Operating System :: POSIX :: Linux',
+    'Programming Language :: Python',
+    'Programming Language :: Python :: 2',
+    'Programming Language :: Python :: 2.6',
+    'Programming Language :: Python :: 2.7',
+    'Programming Language :: Python :: 3',
+    'Programming Language :: Python :: 3.3',
+    'Programming Language :: Python :: 3.4',
+    'Programming Language :: Python :: Implementation :: CPython',
+]
+
+CONSOLE_SCRIPTS = ['governor = governor:main']
+
+
+class PyTest(TestCommand):
+
+    user_options = [('cov=', None, 'Run coverage'), ('cov-xml=', None, 'Generate junit xml report'), ('cov-html=',
+                    None, 'Generate junit html report'), ('junitxml=', None, 'Generate xml of test results')]
+
+    def initialize_options(self):
+        TestCommand.initialize_options(self)
+        self.cov_xml = False
+        self.cov_html = False
+        self.junitxml = None
+
+    def finalize_options(self):
+        TestCommand.finalize_options(self)
+        if self.cov_xml or self.cov_html:
+            self.cov = ['--cov', MAIN_PACKAGE, '--cov', HELPERS, '--cov-report', 'term-missing']
+            if self.cov_xml:
+                self.cov.extend(['--cov-report', 'xml'])
+            if self.cov_html:
+                self.cov.extend(['--cov-report', 'html'])
+        if self.junitxml is not None:
+            self.junitxml = ['--junitxml', self.junitxml]
+
+    def run_tests(self):
+        try:
+            import pytest
+        except:
+            raise RuntimeError('py.test is not installed, run: pip install pytest')
+        params = {'args': self.test_args}
+        if self.cov:
+            params['args'] += self.cov
+            params['plugins'] = ['cov']
+        if self.junitxml:
+            params['args'] += self.junitxml
+        params['args'] += ['--doctest-modules', HELPERS, '-s']
+        errno = pytest.main(**params)
+        sys.exit(errno)
+
+
+def get_install_requirements(path):
+    content = open(os.path.join(__location__, path)).read()
+    return [req for req in content.split('\n') if req != '']
+
+
+def read(fname):
+    return open(os.path.join(__location__, fname)).read()
+
+
+def check_deps(deps):
+    '''check dependency licenses'''
+    from pkg_resources import Requirement
+    import requests
+    for dep in deps:
+        dep = Requirement.parse(dep)
+        url = 'https://pypi.python.org/pypi/{}/json'.format(dep.project_name)
+        r = requests.get(url)
+        data = r.json()
+        print(data['info'].get('name'), data['info'].get('license'))
+
+
+def setup_package():
+    # Assemble additional setup commands
+    cmdclass = {}
+    cmdclass['test'] = PyTest
+
+    # Some helper variables
+    version = os.getenv('GO_PIPELINE_LABEL', VERSION)
+
+    install_reqs = get_install_requirements('requirements.txt')
+
+    # check_deps(install_reqs)
+
+    command_options = {'test': {'test_suite': ('setup.py', 'tests')}}
+    if JUNIT_XML:
+        command_options['test']['junitxml'] = 'setup.py', 'junit.xml'
+    if COVERAGE_XML:
+        command_options['test']['cov_xml'] = 'setup.py', True
+    if COVERAGE_HTML:
+        command_options['test']['cov_html'] = 'setup.py', True
+
+    setup(
+        name=NAME,
+        version=version,
+        url=URL,
+        description=DESCRIPTION,
+        author=AUTHOR,
+        author_email=EMAIL,
+        license=LICENSE,
+        keywords='aws docker ec2 elb lb boto deployment route53 stack traffic',
+        long_description=read('README.md'),
+        classifiers=CLASSIFIERS,
+        test_suite='tests',
+        packages=setuptools.find_packages(exclude=['tests', 'tests.*']),
+        package_data={MAIN_PACKAGE: ["*.json"]},
+        install_requires=install_reqs,
+        setup_requires=['six', 'flake8'],
+        cmdclass=cmdclass,
+        tests_require=['pytest-cov', 'pytest'],
+        command_options=command_options,
+        entry_points={'console_scripts': CONSOLE_SCRIPTS},
+    )
+
+
+if __name__ == '__main__':
+    setup_package()

tests/test_etcd.py

@@ -0,0 +1,82 @@
+import unittest
+import requests
+import time
+import json
+
+from helpers.etcd import Cluster, Etcd
+from helpers.errors import EtcdError, CurrentLeaderError
+
+
+class MockResponse:
+
+    def __init__(self):
+        self.status_code = 200
+        self.content = '{}'
+
+    def json(self):
+        return json.loads(self.content)
+
+
+def requests_get(url, **kwargs):
+    if url.startswith('http://local'):
+        raise Exception()
+    response = MockResponse()
+    if url.startswith('http://remote'):
+        response.content = '{"action":"get","node":{"key":"/service/batman5","dir":true,"nodes":[{"key":"/service/batman5/initialize","value":"postgresql0","modifiedIndex":1582,"createdIndex":1582},{"key":"/service/batman5/leader","value":"postgresql1","expiration":"2015-05-15T09:11:00.037397538Z","ttl":21,"modifiedIndex":20728,"createdIndex":20434},{"key":"/service/batman5/optime","dir":true,"nodes":[{"key":"/service/batman5/optime/leader","value":"2164261704","modifiedIndex":20729,"createdIndex":20729}],"modifiedIndex":20437,"createdIndex":20437},{"key":"/service/batman5/members","dir":true,"nodes":[{"key":"/service/batman5/members/postgresql1","value":"postgres://replicator:[email protected]:5434/postgres","expiration":"2015-05-15T09:10:59.949384522Z","ttl":21,"modifiedIndex":20727,"createdIndex":20727},{"key":"/service/batman5/members/postgresql0","value":"postgres://replicator:[email protected]:5433/postgres","expiration":"2015-05-15T09:11:09.611860899Z","ttl":30,"modifiedIndex":20730,"createdIndex":20730}],"modifiedIndex":1581,"createdIndex":1581}],"modifiedIndex":1581,"createdIndex":1581}}'
+    elif url.startswith('http://other'):
+        response.status_code = 404
+    return response
+
+
+def requests_put(url, **kwargs):
+    if url.startswith('http://local'):
+        raise Exception()
+    response = MockResponse()
+    response.status_code = 201
+    return response
+
+
+def requests_delete(url):
+    if url.startswith('http://local'):
+        raise Exception()
+    response = MockResponse()
+    response.status_code = 204
+    return response
+
+
+def time_sleep(_):
+    pass
+
+
+class TestEtcd(unittest.TestCase):
+
+    def __init__(self, method_name='runTest'):
+        self.setUp = self.set_up
+        super(TestEtcd, self).__init__(method_name)
+
+    def set_up(self):
+        requests.get = requests_get
+        requests.put = requests_put
+        requests.delete = requests_delete
+        time.sleep = time_sleep
+        self.etcd = Etcd({'ttl': 30, 'host': 'localhost', 'scope': 'test'})
+
+    def test_get_client_path(self):
+        self.assertRaises(Exception, self.etcd.get_client_path, '', 2)
+
+    def test_put_client_path(self):
+        self.assertFalse(self.etcd.put_client_path(''))
+
+    def test_delete_client_path(self):
+        self.assertFalse(self.etcd.delete_client_path(''))
+
+    def test_get_cluster(self):
+        self.assertRaises(EtcdError, self.etcd.get_cluster)
+        self.etcd.base_client_url = self.etcd.base_client_url.replace('local', 'remote')
+        cluster = self.etcd.get_cluster()
+        self.assertIsInstance(cluster, Cluster)
+        self.etcd.base_client_url = self.etcd.base_client_url.replace('remote', 'other')
+        self.etcd.get_cluster()
+
+    def test_current_leader(self):
+        self.assertRaises(CurrentLeaderError, self.etcd.current_leader)

tests/test_governor.py

@@ -0,0 +1,68 @@
+import os
+import psycopg2
+import unittest
+import requests
+import sys
+import time
+import yaml
+
+from governor import Governor, main, sigchld_handler
+from test_ha import true, false
+from test_postgresql import Postgresql, os_system, psycopg2_connect
+from test_etcd import requests_get, requests_put, requests_delete
+
+
+def nop(*args, **kwargs):
+    pass
+
+
+def os_waitpid(a, b):
+    return (0, 0)
+
+
+class TestGovernor(unittest.TestCase):
+
+    def __init__(self, method_name='runTest'):
+        self.setUp = self.set_up
+        self.tearDown = self.tear_down
+        super(TestGovernor, self).__init__(method_name)
+
+    def set_up(self):
+        os.system = os_system
+        psycopg2.connect = psycopg2_connect
+        requests.get = requests_get
+        requests.put = requests_put
+        requests.delete = requests_delete
+        time.sleep = nop
+        Governor.run = nop
+        self.write_pg_hba = Postgresql.write_pg_hba
+        self.write_recovery_conf = Postgresql.write_recovery_conf
+        Postgresql.write_pg_hba = nop
+        Postgresql.write_recovery_conf = nop
+
+    def tear_down(self):
+        Postgresql.write_pg_hba = self.write_pg_hba
+        Postgresql.write_recovery_conf = self.write_recovery_conf
+
+    def test_governor_main(self):
+        sys.argv = ['governor.py', 'postgres0.yml']
+        main()
+
+    def test_governor_initialize(self):
+        with open('postgres0.yml', 'r') as f:
+            config = yaml.load(f)
+            g = Governor(config)
+            g.etcd.base_client_url = 'http://remote'
+            g.etcd.client_url
+            g.postgresql.data_directory_empty = true
+            g.etcd.race = true
+            g.initialize()
+            g.etcd.race = false
+            g.initialize()
+            g.postgresql.data_directory_empty = false
+            g.initialize()
+
+    def test_sigchld_handler(self):
+        sigchld_handler(None, None)
+        os.waitpid = os_waitpid
+        sigchld_handler(None, None)

tests/test_ha.py

@@ -0,0 +1,121 @@
+import unittest
+import requests
+
+from helpers.etcd import Etcd
+from helpers.ha import Ha
+from test_etcd import requests_get, requests_put, requests_delete
+
+
+def true(*args, **kwargs):
+    return True
+
+
+def false(*args, **kwargs):
+    return False
+
+
+class MockPostgresql:
+
+    def __init__(self):
+        self.name = 'postgresql0'
+
+    def is_healthy(self):
+        return True
+
+    def write_recovery_conf(self, _):
+        return True
+
+    def start(self):
+        return True
+
+    def is_healthiest_node(self, members):
+        return True
+
+    def is_leader(self):
+        return True
+
+    def promote(self):
+        return True
+
+    def demote(self, _):
+        return True
+
+    def follow_the_leader(self, _):
+        return True
+
+    def create_replication_slots(self, _):
+        return True
+
+    def last_operation(self):
+        return 0
+
+
+class TestHa(unittest.TestCase):
+
+    def __init__(self, method_name='runTest'):
+        self.setUp = self.set_up
+        super(TestHa, self).__init__(method_name)
+
+    def set_up(self):
+        requests.get = requests_get
+        requests.put = requests_put
+        requests.delete = requests_delete
+        self.p = MockPostgresql()
+        self.e = Etcd({'ttl': 30, 'host': 'remotehost', 'scope': 'test'})
+        self.ha = Ha(self.p, self.e)
+
+    def test_start_as_slave(self):
+        self.p.is_healthy = false
+        self.assertEquals(self.ha.run_cycle(), 'started as a secondary')
+
+    def test_start_as_readonly(self):
+        self.p.is_leader = self.p.is_healthy = false
+        self.ha.has_lock = true
+        self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock')
+
+    def test_acquire_lock_as_master(self):
+        self.ha.is_unlocked = true
+        self.assertEquals(self.ha.run_cycle(), 'acquired session lock as a leader')
+
+    def test_promoted_by_acquiring_lock(self):
+        self.ha.is_unlocked = true
+        self.p.is_leader = false
+        self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
+
+    def test_demote_after_failing_to_obtain_lock(self):
+        self.ha.is_unlocked = true
+        self.ha.acquire_lock = false
+        self.assertEquals(self.ha.run_cycle(), 'demoted self due after trying and failing to obtain lock')
+
+    def test_follow_new_leader_after_failing_to_obtain_lock(self):
+        self.ha.is_unlocked = true
+        self.ha.acquire_lock = false
+        self.p.is_leader = false
+        self.assertEquals(self.ha.run_cycle(), 'following new leader after trying and failing to obtain lock')
+
+    def test_demote_because_not_healthiest(self):
+        self.ha.is_unlocked = true
+        self.p.is_healthiest_node = false
+        self.assertEquals(self.ha.run_cycle(), 'demoting self because i am not the healthiest node')
+
+    def test_follow_new_leader_because_not_healthiest(self):
+        self.ha.is_unlocked = true
+        self.p.is_healthiest_node = false
+        self.p.is_leader = false
+        self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
+
+    def test_promote_because_have_lock(self):
+        self.ha.has_lock = true
+        self.p.is_leader = false
+        self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock')
+
+    def test_leader_with_lock(self):
+        self.ha.has_lock = true
+        self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+
+    def test_demote_because_not_having_lock(self):
+        self.assertEquals(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader')
+
+    def test_follow_the_leader(self):
+        self.p.is_leader = false
+        self.assertEquals(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')

tests/test_postgresql.py

@@ -0,0 +1,146 @@
+import os
+import psycopg2
+import unittest
+import shutil
+
+from helpers.etcd import Cluster, Member
+from helpers.postgresql import Postgresql
+
+
+def os_system(cmd):
+    return 0
+
+
+class MockCursor:
+
+    def __init__(self):
+        self.current = 0
+        self.results = []
+
+    def execute(self, sql, *params):
+        if sql.startswith('blabla'):
+            raise psycopg2.OperationalError()
+        elif sql.startswith('SELECT slot_name'):
+            self.results = [('blabla'), ('foobar')]
+        elif sql.startswith('SELECT pg_current_xlog_location()'):
+            self.results = [(0,)]
+        elif sql.startswith('SELECT %s - (pg_last_xlog_replay_location()'):
+            self.results = [(0,)]
+        elif sql.startswith('SELECT pg_last_xlog_replay_location()'):
+            self.results = [(0,)]
+        elif sql.startswith('SELECT pg_is_in_recovery()'):
+            self.results = [(False, )]
+        else:
+            self.results = []
+
+    def fetchone(self):
+        return self.results[0]
+
+    def close(self):
+        pass
+
+    def __iter__(self):
+        for i in self.results:
+            yield i
+
+
+class MockConnect:
+
+    def __init__(self):
+        self.autocommit = False
+
+    def cursor(self):
+        return MockCursor()
+
+    def close(self):
+        if not self.autocommit:
+            raise psycopg2.OperationalError()
+
+
+def psycopg2_connect(*args, **kwargs):
+
+    return MockConnect()
+
+
+def is_running():
+    return False
+
+
+class TestPostgresql(unittest.TestCase):
+
+    def __init__(self, method_name='runTest'):
+        self.setUp = self.set_up
+        self.tearDown = self.tear_down
+        super(TestPostgresql, self).__init__(method_name)
+
+    def set_up(self):
+        os.system = os_system
+        self.p = Postgresql({'name': 'test0', 'data_dir': 'data/test0', 'listen': '127.0.0.1, 127.0.0.2:5432', 'connect_address': '127.0.0.2:5432', 'replication': {
+                            'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, 'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}})
+        psycopg2.connect = psycopg2_connect
+        if not os.path.exists(self.p.data_dir):
+            os.makedirs(self.p.data_dir)
+        self.leader = Member('leader', 'postgres://replicator:[email protected]:5434/postgres', 28)
+
+    def tear_down(self):
+        shutil.rmtree('data')
+
+    def test_data_directory_empty(self):
+        self.assertTrue(self.p.data_directory_empty())
+
+    def test_initialize(self):
+        self.assertTrue(self.p.initialize())
+        self.assertTrue(os.path.exists(os.path.join(self.p.data_dir, 'pg_hba.conf')))
+
+    def test_start(self):
+        self.assertFalse(self.p.start())
+        self.p.is_running = is_running
+        with open(os.path.join(self.p.data_dir, 'postmaster.pid'), 'w'):
+            pass
+        self.assertTrue(self.p.start())
+
+    def test_sync_from_leader(self):
+        self.assertTrue(self.p.sync_from_leader(self.leader))
+
+    def test_follow_the_leader(self):
+        self.p.demote(self.leader)
+        self.p.follow_the_leader(None)
+        self.p.demote(self.leader)
+        self.p.follow_the_leader(self.leader)
+        self.p.follow_the_leader(Member('leader', 'postgres://replicator:[email protected]:5435/postgres', 28))
+
+    def test_create_replication_slots(self):
+        self.p.start()
+        self.p.create_replication_slots('qaz')
+
+    def test_query(self):
+        self.p.query('select 1')
+        self.p.conn.autocommit = False
+        self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla')
+        self.p.query('select %s', 1)
+
+    def test_is_healthiest_node(self):
+        leader = Member('leader', 'postgres://replicator:[email protected]:5435/postgres', 28)
+        me = Member('test0', 'postgres://replicator:[email protected]:5434/postgres', 28)
+        other = Member('test1', 'postgres://replicator:[email protected]:5433/postgres', 28)
+        cluster = Cluster(leader, 0, [leader, me, other])
+        self.assertTrue(self.p.is_healthiest_node(cluster))
+        self.p.config['maximum_lag_on_failover'] = -1
+        self.assertFalse(self.p.is_healthiest_node(cluster))
+
+    def test_is_leader(self):
+        self.assertTrue(self.p.is_leader())
+
+    def test_reload(self):
+        self.assertTrue(self.p.reload())
+
+    def test_is_healthy(self):
+        self.assertTrue(self.p.is_healthy())
+        self.p.is_running = is_running
+        self.assertFalse(self.p.is_healthy())
+
+    def test_promote(self):
+        self.assertTrue(self.p.promote())
+
+    def test_last_operation(self):
+        self.assertEquals(self.p.last_operation(), 0)

tox.ini

@@ -0,0 +1,2 @@
+[flake8]
+max-line-length=120