diff --git a/features/environment.py b/features/environment.py index 81dfb9f2..8377ee79 100644 --- a/features/environment.py +++ b/features/environment.py @@ -81,6 +81,7 @@ class AbstractController(object): def cancel_background(self): pass + class PatroniController(AbstractController): __PORT = 5440 PATRONI_CONFIG = '{}.yml' @@ -275,6 +276,7 @@ class PatroniController(AbstractController): if 'process' not in p.cmdline()[0]: p.terminate() + class ProcessHang(object): """A background thread implementing a cancelable process hang via SIGSTOP.""" @@ -499,7 +501,8 @@ class PatroniPoolController(object): def start(self, name, max_wait_limit=20, tags=None, with_watchdog=False): if name not in self._processes: - self._processes[name] = PatroniController(self._context, name, self.patroni_path, self._output_dir, tags, with_watchdog=with_watchdog) + self._processes[name] = PatroniController(self._context, name, self.patroni_path, + self._output_dir, tags, with_watchdog) self._processes[name].start(max_wait_limit) def __getattr__(self, func): @@ -540,7 +543,7 @@ class WatchdogMonitor(object): def __init__(self, name, work_directory, output_dir): self.fifo_path = os.path.join(work_directory, 'data', 'watchdog.{0}.fifo'.format(name)) self.fifo_file = None - self._stop_requested = False # Relying on bool setting being atomic + self._stop_requested = False # Relying on bool setting being atomic self._thread = None self.last_ping = None self.was_pinged = False @@ -637,7 +640,6 @@ class WatchdogMonitor(object): self._thread.join() self._thread = None - def reset(self): self._log("reset") self.was_pinged = self.was_closed = self._was_triggered = False diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py index ca554754..e6734f36 100644 --- a/features/steps/patroni_api.py +++ b/features/steps/patroni_api.py @@ -111,7 +111,8 @@ def check_response(context, component, data): assert context.status_code == int(data),\ "status code {0} != {1}, response: {2}".format(context.status_code, data, context.response) elif component == 'returncode': - assert context.status_code == int(data), "return code {0} != {1}, {2}".format(context.status_code, data, context.response) + assert context.status_code == int(data), "return code {0} != {1}, {2}".format(context.status_code, + data, context.response) elif component == 'text': assert context.response == data.strip('"'), "response {0} does not contain {1}".format(context.response, data) elif component == 'output': diff --git a/features/steps/watchdog.py b/features/steps/watchdog.py index 255a7391..e37ace3d 100644 --- a/features/steps/watchdog.py +++ b/features/steps/watchdog.py @@ -1,6 +1,7 @@ from behave import step, then import time + def polling_loop(timeout, interval=1): """Returns an iterator that returns values until timeout has passed. Timeout is measured from start of iteration.""" start_time = time.time() @@ -12,7 +13,6 @@ def polling_loop(timeout, interval=1): time.sleep(interval) - @step('I start {name:w} with watchdog') def start_patroni_with_watchdog(context, name): return context.pctl.start(name, with_watchdog=True) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 9d31f9d1..6fce3018 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -954,13 +954,9 @@ class Postgresql(object): def _wait_for_connection_close(self, pid): try: with self.connection().cursor() as cur: - while True: # Need a timeout here? - if pid == self.get_pid() and self.is_pid_running(pid): - cur.execute("SELECT 1") - time.sleep(STOP_POLLING_INTERVAL) - continue - else: - break + while pid == self.get_pid() and self.is_pid_running(pid): # Need a timeout here? + cur.execute("SELECT 1") + time.sleep(STOP_POLLING_INTERVAL) except psycopg2.Error: pass @@ -1319,50 +1315,6 @@ class Postgresql(object): self.call_nowait(ACTION_ON_ROLE_CHANGE) return True - def _do_rewind(self, leader): - logger.info("rewind flag is set") - - if self.is_running() and not self.stop(checkpoint=False): - logger.warning('Can not run pg_rewind because postgres is still running') - return False - - # prepare pg_rewind connection - r = leader.conn_kwargs(self._superuser) - - # first make sure that we are really trying to rewind - # from the master and run a checkpoint on a t in order to - # make it store the new timeline (5540277D.8020309@iki.fi) - leader_status = self.checkpoint(r) - if leader_status: - logger.warning('Can not use %s for rewind: %s', leader.name, leader_status) - return False - - # at present, pg_rewind only runs when the cluster is shut down cleanly - # and not shutdown in recovery. We have to remove the recovery.conf if present - # and start/shutdown in a single user mode to emulate this. - # XXX: if recovery.conf is linked, it will be written anew as a normal file. - if os.path.isfile(self._recovery_conf) or os.path.islink(self._recovery_conf): - os.unlink(self._recovery_conf) - - # Archived segments might be useful to pg_rewind, - # clean the flags that tell we should remove them. - self.cleanup_archive_status() - - # Start in a single user mode and stop to produce a clean shutdown - opts = self.read_postmaster_opts() - opts.update({'archive_mode': 'on', 'archive_command': 'false'}) - self.single_user_mode(options=opts) - - try: - if not self.rewind(r): - logger.error('unable to rewind the former master') - if self.config.get('remove_data_directory_on_rewind_failure', False): - self.remove_data_directory() - return False - return True - finally: - self._need_rewind = False - def save_configuration_files(self): """ copy postgresql.conf to postgresql.conf.backup to be able to retrive configuration files diff --git a/patroni/watchdog/base.py b/patroni/watchdog/base.py index 6154d583..da65155d 100644 --- a/patroni/watchdog/base.py +++ b/patroni/watchdog/base.py @@ -94,7 +94,7 @@ class Watchdog(object): logger.info("{0} activated with {1} second timeout, timing slack {2} seconds" .format(self.impl.describe(), actual_timeout, slack)) else: - if self.mode == MODE_REQUIRED: + if self.mode == MODE_REQUIRED: # XXX: can we really get here? logger.error("Configuration requires watchdog, but watchdog could not be activated") sys.exit(1) @@ -116,7 +116,7 @@ class Watchdog(object): logger.error("Error while sending keepalive: %s", e) def _get_impl(self): - if self.mode not in [MODE_AUTOMATIC, MODE_REQUIRED]: + if self.mode not in [MODE_AUTOMATIC, MODE_REQUIRED]: # XXX: can't be reached return NullWatchdog() if self.driver == 'testing': diff --git a/patroni/watchdog/linux.py b/patroni/watchdog/linux.py index e9cc3155..92f0db0e 100644 --- a/patroni/watchdog/linux.py +++ b/patroni/watchdog/linux.py @@ -145,7 +145,7 @@ class LinuxWatchdogDevice(WatchdogBase): os.close(self._fd) self._fd = None except OSError as e: - return WatchdogError("Error while closing {0}: {1}".format(self.describe(), e)) + raise WatchdogError("Error while closing {0}: {1}".format(self.describe(), e)) @property def can_be_disabled(self): @@ -176,7 +176,7 @@ class LinuxWatchdogDevice(WatchdogBase): try: _, version, identity = self.get_support() ver_str = " (firmware {0})".format(version) if version else "" - except WatchdogError: + except WatchdogError: # XXX: Can it really be raise when self._fd is not None? pass return identity + ver_str + dev_str diff --git a/tests/test_async_executor.py b/tests/test_async_executor.py index 6f867428..2c726c0e 100644 --- a/tests/test_async_executor.py +++ b/tests/test_async_executor.py @@ -1,7 +1,7 @@ import unittest from mock import Mock, patch -from patroni.async_executor import AsyncExecutor +from patroni.async_executor import AsyncExecutor, CriticalTask from threading import Thread @@ -16,3 +16,11 @@ class TestAsyncExecutor(unittest.TestCase): def test_run(self): self.a.run(Mock(side_effect=Exception())) + + +class TestCriticalTask(unittest.TestCase): + + def test_completed_task(self): + ct = CriticalTask() + ct.complete(1) + self.assertFalse(ct.cancel()) diff --git a/tests/test_ha.py b/tests/test_ha.py index 0a899823..bce463f2 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1,6 +1,7 @@ import datetime import etcd import os +import time import unittest from mock import Mock, MagicMock, PropertyMock, patch @@ -8,12 +9,13 @@ from patroni.config import Config from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState from patroni.dcs.etcd import Client from patroni.exceptions import DCSError, PostgresException -from patroni.ha import Ha, _MemberStatus +from patroni.ha import Ha, _MemberStatus, BackgroundKeepaliveSender from patroni.postgresql import Postgresql from patroni.watchdog import Watchdog from patroni.utils import tzutc from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get from test_postgresql import psycopg2_connect +from threading import Event def true(*args, **kwargs): @@ -246,7 +248,8 @@ class TestHa(unittest.TestCase): def test_demote_because_not_having_lock(self): self.ha.cluster.is_unlocked = false - self.assertEquals(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader') + with patch.object(Watchdog, 'is_running', PropertyMock(return_value=True)): + self.assertEquals(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader') def test_demote_because_update_lock_failed(self): self.ha.cluster.is_unlocked = false @@ -335,6 +338,7 @@ class TestHa(unittest.TestCase): with patch.object(self.ha, "restart_matches", return_value=False): self.assertEquals(self.ha.restart({'foo': 'bar'}), (False, "restart conditions are not satisfied")) + @patch('os.kill', Mock()) def test_restart_in_progress(self): with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)): self.ha.restart({}, run_async=True) @@ -349,9 +353,10 @@ class TestHa(unittest.TestCase): self.ha.update_lock = false self.p.set_role('master') - with patch('patroni.postgresql.Postgresql.stop') as stop_mock: - self.assertEquals(self.ha.run_cycle(), 'lost leader lock during restart') - stop_mock.assert_called() + with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)): + with patch('patroni.postgresql.Postgresql.stop') as stop_mock: + self.assertEquals(self.ha.run_cycle(), 'lost leader lock during restart') + stop_mock.assert_called() @patch('requests.get', requests_get) def test_manual_failover_from_leader(self): @@ -797,6 +802,10 @@ class TestHa(unittest.TestCase): def test_wakup(self): self.ha.wakeup() + def test_shutdown(self): + self.p.is_running = false + self.ha.shutdown() + @patch('time.sleep', Mock()) def test_leader_with_empty_directory(self): self.ha.cluster = get_cluster_initialized_with_leader() @@ -808,3 +817,16 @@ class TestHa(unittest.TestCase): self.ha.has_lock = false # will not say bootstrap from leader as replica can't self elect self.assertEquals(self.ha.run_cycle(), "trying to bootstrap from replica 'other'") + + +class TestBackgroundKeepaliveSender(unittest.TestCase): + + def test_run(self): + safe_event = Event() + ha = Mock() + ha.dcs.loop_wait = 0.1 + with BackgroundKeepaliveSender(ha, safe_event): + time.sleep(1) + safe_event.set() + time.sleep(1) + self.assertTrue(ha.keepalive.call_count > 2) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 81988bf4..e8e024fa 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -1,3 +1,4 @@ +import errno import mock # for the mock.call method, importing it without a namespace breaks python3 import os import psycopg2 @@ -6,6 +7,7 @@ import subprocess import unittest from mock import Mock, MagicMock, PropertyMock, patch, mock_open +from patroni.async_executor import CriticalTask from patroni.dcs import Cluster, Leader, Member, SyncState from patroni.exceptions import PostgresException, PostgresConnectionException from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE @@ -216,13 +218,13 @@ class TestPostgresql(unittest.TestCase): mock_is_running.return_value = True mock_wait_for_port_open.return_value = True mock_wait_for_startup.return_value = False - mock_popen.stdout.readline.return_value = '123' + mock_popen.return_value.stdout.readline.return_value = '123' self.assertTrue(self.p.start()) mock_is_running.return_value = False open(os.path.join(self.data_dir, 'postmaster.pid'), 'w').close() pg_conf = os.path.join(self.data_dir, 'postgresql.conf') open(pg_conf, 'w').close() - self.assertFalse(self.p.start()) + self.assertFalse(self.p.start(task=CriticalTask())) with open(pg_conf) as f: lines = f.readlines() self.assertTrue("f.oo = 'bar'\n" in lines) @@ -233,6 +235,9 @@ class TestPostgresql(unittest.TestCase): mock_wait_for_port_open.return_value = False self.assertFalse(self.p.start()) + task = CriticalTask() + task.cancel() + self.assertFalse(self.p.start(task=task)) @patch.object(Postgresql, 'pg_isready') @patch.object(Postgresql, 'read_pid_file') @@ -266,13 +271,24 @@ class TestPostgresql(unittest.TestCase): mock_pg_isready.return_value = 'garbage' self.assertTrue(self.p.wait_for_port_open(42, 100., 1)) + @patch('time.sleep', Mock()) @patch.object(Postgresql, 'is_running') - def test_stop(self, mock_is_running): + @patch.object(Postgresql, 'get_pid') + def test_stop(self, mock_get_pid, mock_is_running): mock_is_running.return_value = True + mock_get_pid.return_value = 0 self.assertTrue(self.p.stop()) - with patch('subprocess.call', Mock(return_value=1)): - mock_is_running.return_value = False + mock_get_pid.return_value = -1 + self.assertFalse(self.p.stop()) + mock_get_pid.return_value = 123 + with patch('os.kill', Mock(side_effect=[OSError(errno.ESRCH, ''), OSError, None])): self.assertTrue(self.p.stop()) + self.assertFalse(self.p.stop()) + self.p.stop_safepoint_reached.clear() + self.assertTrue(self.p.stop()) + with patch.object(Postgresql, '_signal_postmaster_stop', Mock(return_value=(123, None))): + with patch.object(Postgresql, 'is_pid_running', Mock(side_effect=[True, False, False])): + self.assertTrue(self.p.stop()) def test_restart(self): self.p.start = Mock(return_value=False) @@ -778,3 +794,41 @@ class TestPostgresql(unittest.TestCase): self.p.get_server_parameters(config) self.p.set_synchronous_standby('foo') self.p.get_server_parameters(config) + + @patch.object(Postgresql, 'read_pid_file', Mock(return_value={'pid': 'z'})) + def test_get_pid(self): + self.p.get_pid() + + @patch.object(Postgresql, 'is_running', Mock(return_value=True)) + @patch.object(Postgresql, '_signal_postmaster_stop', Mock(return_value=(123, None))) + @patch.object(Postgresql, 'get_pid', Mock(return_value=123)) + @patch('time.sleep', Mock()) + @patch.object(Postgresql, 'is_pid_running') + def test__wait_for_connection_close(self, mock_is_pid_running): + mock_is_pid_running.side_effect = [True, False, False] + self.p.stop_safepoint_reached.clear() + self.p.stop() + + mock_is_pid_running.side_effect = [True, False, False] + self.p.stop_safepoint_reached.clear() + with patch.object(MockCursor, "execute", Mock(side_effect=psycopg2.Error)): + self.p.stop() + + @patch.object(Postgresql, 'is_running', Mock(return_value=True)) + @patch.object(Postgresql, '_signal_postmaster_stop', Mock(return_value=(123, None))) + @patch.object(Postgresql, 'get_pid', Mock(return_value=123)) + @patch.object(Postgresql, 'is_pid_running', Mock(return_value=False)) + @patch('psutil.Process') + def test__wait_for_user_backends_to_close(self, mock_psutil): + child = Mock() + child.cmdline.return_value = ['foo'] + mock_psutil.return_value.children.return_value = [child] + self.p.stop_safepoint_reached.clear() + self.p.stop() + + @patch('os.kill', Mock(side_effect=[OSError(errno.ESRCH, ''), OSError])) + @patch('time.sleep', Mock()) + @patch.object(Postgresql, 'is_pid_running', Mock(side_effect=[True, False])) + def test_terminate_starting_postmaster(self): + self.p.terminate_starting_postmaster(123) + self.p.terminate_starting_postmaster(123) diff --git a/tests/test_watchdog.py b/tests/test_watchdog.py index cfa4d2ca..1ffae833 100644 --- a/tests/test_watchdog.py +++ b/tests/test_watchdog.py @@ -1,12 +1,13 @@ -import unittest -from mock import patch -import platform import ctypes - -from patroni.watchdog import Watchdog import patroni.watchdog.linux as linuxwd - import sys +import unittest + +from mock import patch, Mock, PropertyMock +from patroni.watchdog import Watchdog, WatchdogError +from patroni.watchdog.base import NullWatchdog +from patroni.watchdog.linux import LinuxWatchdogDevice + class MockDevice(object): def __init__(self, fd, filename, flag): @@ -20,41 +21,46 @@ class MockDevice(object): mock_devices = [None] + def mock_open(filename, flag): fd = len(mock_devices) mock_devices.append(MockDevice(fd, filename, flag)) return fd + def mock_ioctl(fd, op, arg=None, mutate_flag=False): assert 0 < fd < len(mock_devices) dev = mock_devices[fd] - sys.stderr.write("Ioctl %d %d %r\n" %( fd, op, arg)) + sys.stderr.write("Ioctl %d %d %r\n" % (fd, op, arg)) if op == linuxwd.WDIOC_GETSUPPORT: sys.stderr.write("Get support\n") - assert(mutate_flag == True) - arg.options = sum(map(linuxwd.WDIOF.get, ['SETTIMEOUT', 'KEEPALIVEPING', 'MAGICCLOSE'])) + assert(mutate_flag is True) + arg.options = sum(map(linuxwd.WDIOF.get, ['SETTIMEOUT', 'KEEPALIVEPING'])) arg.identity = (ctypes.c_ubyte*32)(*map(ord, 'Mock Watchdog')) elif op == linuxwd.WDIOC_GETTIMEOUT: arg.value = dev.timeout elif op == linuxwd.WDIOC_SETTIMEOUT: sys.stderr.write("Set timeout called with %s\n" % arg.value) assert 0 < arg.value < 65535 - dev.timeout = arg.value + dev.timeout = arg.value - 1 else: raise Exception("Unknown op %d", op) return 0 + def mock_write(fd, string): assert 0 < fd < len(mock_devices) assert len(string) == 1 assert mock_devices[fd].open mock_devices[fd].writes.append(string) + def mock_close(fd): assert 0 < fd < len(mock_devices) assert mock_devices[fd].open mock_devices[fd].open = False + @patch('os.open', mock_open) @patch('os.write', mock_write) @patch('os.close', mock_close) @@ -63,18 +69,32 @@ class TestWatchdog(unittest.TestCase): def setUp(self): mock_devices[:] = [None] + @patch('platform.system', Mock(return_value='Linux')) + @patch.object(LinuxWatchdogDevice, 'can_be_disabled', PropertyMock(return_value=True)) + def test_unsafe_timeout_disable_watchdog_and_exit(self): + self.assertRaises(SystemExit, Watchdog({'ttl': 30, 'loop_wait': 15, 'watchdog': {'mode': 'required'}}).activate) + + @patch('platform.system', Mock(return_value='Linux')) + @patch.object(LinuxWatchdogDevice, 'get_timeout', Mock(return_value=16)) + def test_timeout_does_not_ensure_safe_termination(self): + Watchdog({'ttl': 30, 'loop_wait': 15, 'watchdog': {'mode': 'auto'}}).activate() + self.assertEquals(len(mock_devices), 2) + + @patch('platform.system', Mock(return_value='Linux')) + @patch.object(Watchdog, 'is_running', PropertyMock(return_value=False)) + def test_watchdog_not_activated(self): + self.assertRaises(SystemExit, Watchdog({'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'required'}}).activate) + + @patch('platform.system', Mock(return_value='Linux')) def test_basic_operation(self): - if platform.system() != 'Linux': - return - watchdog = Watchdog({'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'required'}}) - watchdog.activate() + self.assertEquals(len(mock_devices), 2) device = mock_devices[-1] self.assertTrue(device.open) - self.assertEquals(device.timeout, 15) + self.assertEquals(device.timeout, 14) watchdog.keepalive() self.assertEquals(len(device.writes), 1) @@ -88,3 +108,56 @@ class TestWatchdog(unittest.TestCase): watchdog.activate() self.assertEquals(len(mock_devices), 1) self.assertFalse(watchdog.is_running) + + def test_parse_mode(self): + with patch('patroni.watchdog.base.logger.warning', new_callable=Mock()) as warning_mock: + watchdog = Watchdog({'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'bad'}}) + self.assertEquals(watchdog.mode, 'off') + warning_mock.assert_called_once() + + @patch('platform.system', Mock(return_value='Unknown')) + def test_unsupported_platform(self): + self.assertRaises(SystemExit, Watchdog, {'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'required'}}) + + def test_exceptions(self): + wd = Watchdog({'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'bad'}}) + wd.impl.close = wd.impl.keepalive = Mock(side_effect=WatchdogError('')) + self.assertIsNone(wd.disable()) + self.assertIsNone(wd.keepalive()) + + +class TestNullWatchdog(unittest.TestCase): + + def test_basics(self): + watchdog = NullWatchdog() + self.assertTrue(watchdog.can_be_disabled) + self.assertRaises(WatchdogError, watchdog.set_timeout, 1) + self.assertEquals(watchdog.describe(), 'NullWatchdog') + self.assertIsInstance(NullWatchdog.from_config({}), NullWatchdog) + + +class TestLinuxWatchdogDevice(unittest.TestCase): + + def setUp(self): + self.impl = LinuxWatchdogDevice.from_config({}) + + @patch('os.open', Mock(return_value=3)) + @patch('os.write', Mock(side_effect=OSError)) + @patch('fcntl.ioctl', Mock(return_value=0)) + def test_basics(self): + self.impl.open() + try: + if self.impl.get_support().has_foo: + self.assertFail() + except Exception as e: + self.assertTrue(isinstance(e, AttributeError)) + self.assertRaises(WatchdogError, self.impl.close) + self.assertRaises(WatchdogError, self.impl.keepalive) + self.assertRaises(WatchdogError, self.impl.set_timeout, -1) + + @patch('os.open', Mock(return_value=3)) + @patch('fcntl.ioctl', Mock(return_value=-1)) + def test__ioctl(self): + self.assertRaises(WatchdogError, self.impl.get_support) + self.impl.open() + self.assertRaises(IOError, self.impl.get_support)