diff --git a/features/environment.py b/features/environment.py index 19dc7974..deb8466c 100644 --- a/features/environment.py +++ b/features/environment.py @@ -1,11 +1,6 @@ import abc -import consul import datetime -import etcd -import kazoo.client -import kazoo.exceptions import os -import psutil import psycopg2 import json import shutil @@ -249,51 +244,16 @@ class PatroniController(AbstractController): except Exception: return None - def database_is_running(self): - pid = self._get_pid() - if not pid: - return False - try: - os.kill(pid, 0) - except OSError: - return False - return True - def patroni_hang(self, timeout): hang = ProcessHang(self._handle.pid, timeout) self._closables.append(hang) hang.start() - def checkpoint_hang(self, timeout): - pid = self._get_pid() - if not pid: - return False - proc = psutil.Process(pid) - for child in proc.children(): - if 'checkpoint' in child.cmdline()[0]: - checkpointer = child - break - else: - return False - hang = ProcessHang(checkpointer.pid, timeout) - self._closables.append(hang) - hang.start() - return True - def cancel_background(self): for obj in self._closables: obj.close() self._closables = [] - def terminate_backends(self): - pid = self._get_pid() - if not pid: - return False - proc = psutil.Process(pid) - for p in proc.children(): - if 'process' not in p.cmdline()[0]: - p.terminate() - @property def backup_source(self): return 'postgres://{username}:{password}@{host}:{port}/{database}'.format(**self._replication) @@ -375,9 +335,11 @@ class ConsulController(AbstractDcsController): super(ConsulController, self).__init__(context) os.environ['PATRONI_CONSUL_HOST'] = 'localhost:8500' os.environ['PATRONI_CONSUL_REGISTER_SERVICE'] = 'on' - self._client = consul.Consul() self._config_file = None + import consul + self._client = consul.Consul() + def _start(self): self._config_file = self._work_directory + '.json' with open(self._config_file, 'wb') as f: @@ -417,6 +379,8 @@ class EtcdController(AbstractDcsController): def __init__(self, context): super(EtcdController, self).__init__(context) os.environ['PATRONI_ETCD_HOST'] = 'localhost:2379' + + import etcd self._client = etcd.Client(port=2379) def _start(self): @@ -424,12 +388,14 @@ class EtcdController(AbstractDcsController): stdout=self._log, stderr=subprocess.STDOUT) def query(self, key, scope='batman'): + import etcd try: return self._client.get(self.path(key, scope)).value except etcd.EtcdKeyNotFound: return None def cleanup_service_tree(self): + import etcd try: self._client.delete(self.path(scope=''), recursive=True) except (etcd.EtcdKeyNotFound, etcd.EtcdConnectionFailed): @@ -474,12 +440,12 @@ class KubernetesController(AbstractDcsController): def delete_pod(self, name): try: self._api.delete_namespaced_pod(name, self._namespace, body=self._client.V1DeleteOptions()) - except: + except Exception: pass while True: try: self._api.read_namespaced_pod(name, self._namespace) - except: + except Exception: break def query(self, key, scope='batman'): @@ -493,17 +459,17 @@ class KubernetesController(AbstractDcsController): return e.metadata.annotations[key] else: return json.dumps(e.metadata.annotations) - except: + except Exception: return None def cleanup_service_tree(self): try: self._api.delete_collection_namespaced_pod(self._namespace, label_selector=self._label_selector) - except: + except Exception: pass try: self._api.delete_collection_namespaced_endpoints(self._namespace, label_selector=self._label_selector) - except: + except Exception: pass while True: @@ -523,18 +489,22 @@ class ZooKeeperController(AbstractDcsController): super(ZooKeeperController, self).__init__(context, False) if export_env: os.environ['PATRONI_ZOOKEEPER_HOSTS'] = "'localhost:2181'" + + import kazoo.client self._client = kazoo.client.KazooClient() def _start(self): pass # TODO: implement later def query(self, key, scope='batman'): + import kazoo.exceptions try: return self._client.get(self.path(key, scope))[0].decode('utf-8') except kazoo.exceptions.NoNodeError: return None def cleanup_service_tree(self): + import kazoo.exceptions try: self._client.delete(self.path(scope=''), recursive=True) except (kazoo.exceptions.NoNodeError): @@ -594,9 +564,8 @@ class PatroniPoolController(object): self._processes[name].start(max_wait_limit) def __getattr__(self, func): - if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to', 'add_tag_to_config', - 'get_watchdog', 'database_is_running', 'checkpoint_hang', 'patroni_hang', - 'terminate_backends', 'backup']: + if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to', + 'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup']: raise AttributeError("PatroniPoolController instance has no attribute '{0}'".format(func)) def wrapper(name, *args, **kwargs): @@ -610,7 +579,7 @@ class PatroniPoolController(object): self._processes.clear() def create_and_set_output_directory(self, feature_name): - feature_dir = os.path.join(self.patroni_path, 'features/output', feature_name.replace(' ', '_')) + feature_dir = os.path.join(self.patroni_path, 'features', 'output', feature_name.replace(' ', '_')) if os.path.exists(feature_dir): shutil.rmtree(feature_dir) os.makedirs(feature_dir) @@ -637,7 +606,7 @@ class PatroniPoolController(object): 'parameters': { 'archive_mode': 'on', 'archive_command': 'mkdir -p {0} && test ! -f {0}/%f && cp %p {0}/%f'.format( - os.path.join(self.patroni_path, 'data/wal_archive')) + os.path.join(self.patroni_path, 'data', 'wal_archive')) }, 'authentication': { 'superuser': {'password': 'zalando1'}, @@ -654,7 +623,7 @@ class PatroniPoolController(object): 'method': 'backup_restore', 'backup_restore': { 'command': 'features/backup_restore.sh --sourcedir=' + os.path.join(self.patroni_path, - 'data/basebackup'), + 'data', 'basebackup'), 'recovery_conf': { 'recovery_target_action': 'promote', 'recovery_target_timeline': 'latest', diff --git a/features/steps/standby_cluster.py b/features/steps/standby_cluster.py index ce2866a9..03c62e5f 100644 --- a/features/steps/standby_cluster.py +++ b/features/steps/standby_cluster.py @@ -31,7 +31,7 @@ def start_patroni(context, name, cluster_name): "callbacks": {c: callback + name for c in ('on_start', 'on_stop', 'on_restart', 'on_role_change')}, "backup_restore": { "command": "features/backup_restore.sh --sourcedir=" + os.path.join(context.pctl.patroni_path, - "data/basebackup")} + 'data', 'basebackup')} } }) diff --git a/features/steps/watchdog.py b/features/steps/watchdog.py index 58f50e5e..9f2b9cf0 100644 --- a/features/steps/watchdog.py +++ b/features/steps/watchdog.py @@ -44,31 +44,6 @@ def watchdog_was_triggered(context, name, timeout): assert False -@then('{name:w} watchdog was not triggered') -def watchdog_was_not_triggered(context, name): - assert not context.pctl.get_watchdog(name).was_triggered - - -@step('{name:w} checkpoint takes {timeout:d} seconds') -def checkpoint_hang(context, name, timeout): - assert context.pctl.checkpoint_hang(name, timeout) - - @step('{name:w} hangs for {timeout:d} seconds') def patroni_hang(context, name, timeout): return context.pctl.patroni_hang(name, timeout) - - -@step('I terminate {name:w} user processes') -def terminate_backends(context, name): - return context.pctl.terminate_backends(name) - - -@step('Sleep for {timeout:d} seconds') -def dcs_connection_lost(context, timeout): - time.sleep(timeout) - - -@then('{name:w} database is running') -def database_is_running(context, name): - assert context.pctl.database_is_running(name) diff --git a/patroni/__init__.py b/patroni/__init__.py index 0b5f9625..9ad27728 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -162,6 +162,11 @@ class Patroni(object): def patroni_main(): + if sys.version_info >= (3, 4): + # The default, forking, method is not a good idea in a multithreaded process: https://bugs.python.org/issue6721 + import multiprocessing + multiprocessing.set_start_method('spawn') + patroni = Patroni() try: patroni.run() @@ -176,13 +181,6 @@ def fatal(string, *args): sys.exit(1) -def use_spawn_start_method(): - if sys.version_info >= (3, 4): - # The default, forking, method is not a good idea in a multithreaded process: https://bugs.python.org/issue6721 - import multiprocessing - multiprocessing.set_start_method('spawn') - - def check_psycopg2(): min_psycopg2 = (2, 5, 4) min_psycopg2_str = '.'.join(map(str, min_psycopg2)) @@ -205,7 +203,6 @@ def check_psycopg2(): def main(): - use_spawn_start_method() check_psycopg2() if os.getpid() != 1: return patroni_main() diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 25286e7c..4257e834 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -481,7 +481,8 @@ class Kubernetes(AbstractDCS): """Unused""" def manual_failover(self, leader, candidate, scheduled_at=None, index=None): - annotations = {'leader': leader or None, 'member': candidate or None, 'scheduled_at': scheduled_at} + annotations = {'leader': leader or None, 'member': candidate or None, + 'scheduled_at': scheduled_at and scheduled_at.isoformat()} patch = bool(self.cluster and isinstance(self.cluster.failover, Failover) and self.cluster.failover.index) return self.patch_or_create(self.failover_path, annotations, index, bool(index or patch), False) diff --git a/patroni/watchdog/base.py b/patroni/watchdog/base.py index 0007dee1..5c155101 100644 --- a/patroni/watchdog/base.py +++ b/patroni/watchdog/base.py @@ -57,7 +57,7 @@ class WatchdogConfig(object): return not self == other def get_impl(self): - if self.driver == 'testing': + if self.driver == 'testing': # pragma: no cover from patroni.watchdog.linux import TestingWatchdogDevice return TestingWatchdogDevice.from_config(self.driver_config) elif platform.system() == 'Linux' and self.driver == 'default': diff --git a/patroni/watchdog/linux.py b/patroni/watchdog/linux.py index fcb363b2..ce80b5c1 100644 --- a/patroni/watchdog/linux.py +++ b/patroni/watchdog/linux.py @@ -16,11 +16,11 @@ IOC_DIRBITS = 2 # Non-generic platform special cases machine = platform.machine() -if machine in ['mips', 'sparc', 'powerpc', 'ppc64']: +if machine in ['mips', 'sparc', 'powerpc', 'ppc64']: # pragma: no cover IOC_SIZEBITS = 13 IOC_DIRBITS = 3 IOC_NONE, IOC_WRITE, IOC_READ = 1, 2, 4 -elif machine == 'parisc': +elif machine == 'parisc': # pragma: no cover IOC_WRITE, IOC_READ = 2, 1 IOC_NRSHIFT = 0 @@ -218,7 +218,7 @@ class LinuxWatchdogDevice(WatchdogBase): return timeout.value -class TestingWatchdogDevice(LinuxWatchdogDevice): +class TestingWatchdogDevice(LinuxWatchdogDevice): # pragma: no cover """Converts timeout ioctls to regular writes that can be intercepted from a named pipe.""" timeout = 60 diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 0623681b..39f4eab1 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -157,17 +157,18 @@ class TestCacheBuilder(unittest.TestCase): @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map) @patch('patroni.dcs.kubernetes.ObjectCache._watch') def test__build_cache(self, mock_response): - mock_response.return_value.read_chunked.return_value = json.dumps( + mock_response.return_value.read_chunked.return_value = [json.dumps( {'type': 'MODIFIED', 'object': {'metadata': { 'name': self.k.config_path, 'resourceVersion': '2', 'annotations': {self.k._CONFIG: 'foo'}}}} - ) + '\n' + json.dumps( + ).encode('utf-8'), ('\n' + json.dumps( {'type': 'DELETED', 'object': {'metadata': { 'name': self.k.config_path, 'resourceVersion': '3'}}} ) + '\n' + json.dumps( {'type': 'MDIFIED', 'object': {'metadata': {'name': self.k.config_path}}} - ) + '\n' + json.dumps({'object': {'code': 410}}) + '\n' + ) + '\n' + json.dumps({'object': {'code': 410}}) + '\n').encode('utf-8')] self.k._kinds._build_cache() @patch('patroni.dcs.kubernetes.logger.error', Mock(side_effect=SleepException)) + @patch('patroni.dcs.kubernetes.ObjectCache._build_cache', Mock(side_effect=Exception)) def test_run(self): self.assertRaises(SleepException, self.k._pods.run) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 5f6a51f5..99758d95 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -2,7 +2,6 @@ import etcd import logging import os import signal -import sys import time import unittest @@ -41,21 +40,20 @@ class MockFrozenImporter(object): @patch.object(etcd.Client, 'read', etcd_read) class TestPatroni(unittest.TestCase): + @patch('sys.argv', ['patroni.py', 'postgres0.yml']) @patch('pkgutil.get_importer', Mock(return_value=MockFrozenImporter())) @patch('sys.frozen', Mock(return_value=True), create=True) @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock()) @patch.object(etcd.Client, 'read', etcd_read) @patch.object(Thread, 'start', Mock()) + @patch.object(Client, 'machines', PropertyMock(return_value=['http://remotehost:2379'])) def setUp(self): self._handlers = logging.getLogger().handlers[:] RestApiServer._BaseServer__is_shut_down = Mock() RestApiServer._BaseServer__shutdown_request = True RestApiServer.socket = 0 - with patch.object(Client, 'machines') as mock_machines: - mock_machines.__get__ = Mock(return_value=['http://remotehost:2379']) - sys.argv = ['patroni.py', 'postgres0.yml'] - os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0' - self.p = Patroni() + os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0' + self.p = Patroni() def tearDown(self): logging.getLogger().handlers[:] = self._handlers @@ -66,24 +64,26 @@ class TestPatroni(unittest.TestCase): self.p.load_dynamic_configuration() self.p.load_dynamic_configuration() + @patch('sys.argv', ['patroni.py', 'postgres0.yml']) @patch('time.sleep', Mock(side_effect=SleepException)) + @patch('multiprocessing.set_start_method', Mock(), create=True) @patch.object(etcd.Client, 'delete', Mock()) - @patch.object(Client, 'machines') + @patch.object(Client, 'machines', PropertyMock(return_value=['http://remotehost:2379'])) @patch.object(Thread, 'join', Mock()) - def test_patroni_patroni_main(self, mock_machines): + def test_patroni_patroni_main(self): with patch('subprocess.call', Mock(return_value=1)): - sys.argv = ['patroni.py', 'postgres0.yml'] - - mock_machines.__get__ = Mock(return_value=['http://remotehost:2379']) with patch.object(Patroni, 'run', Mock(side_effect=SleepException)): + os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0' self.assertRaises(SleepException, patroni_main) with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())): with patch('patroni.ha.Ha.is_paused', Mock(return_value=True)): + os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0' patroni_main() + with patch('patroni.Patroni', Mock(side_effect=Exception)), patch('sys.version_info', (3, 6)): + self.assertRaises(Exception, patroni_main) @patch('os.getpid') @patch('multiprocessing.Process') - @patch('patroni.use_spawn_start_method', Mock()) @patch('patroni.patroni_main', Mock()) def test_patroni_main(self, mock_process, mock_getpid): mock_getpid.return_value = 2