diff --git a/features/environment.py b/features/environment.py index eb307452..83bc83ea 100644 --- a/features/environment.py +++ b/features/environment.py @@ -14,6 +14,7 @@ import yaml import patroni.psycopg as psycopg +from patroni.request import PatroniRequest from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer @@ -138,12 +139,24 @@ class PatroniController(AbstractController): def _start(self): if self.watchdog: self.watchdog.start() + env = os.environ.copy() if isinstance(self._context.dcs_ctl, KubernetesController): self._context.dcs_ctl.create_pod(self._name[8:], self._scope) - os.environ['PATRONI_KUBERNETES_POD_IP'] = '10.0.0.' + self._name[-1] - return subprocess.Popen([sys.executable, '-m', 'coverage', 'run', - '--source=patroni', '-p', 'patroni.py', self._config], - stdout=self._log, stderr=subprocess.STDOUT, cwd=self._work_directory) + env['PATRONI_KUBERNETES_POD_IP'] = '10.0.0.' + self._name[-1] + if os.name == 'nt': + env['BEHAVE_DEBUG'] = 'true' + patroni = subprocess.Popen([sys.executable, '-m', 'coverage', 'run', + '--source=patroni', '-p', 'patroni.py', self._config], env=env, + stdout=self._log, stderr=subprocess.STDOUT, cwd=self._work_directory) + if os.name == 'nt': + patroni.terminate = self.terminate + return patroni + + def terminate(self): + try: + self._context.request_executor.request('POST', self._restapi_url + '/sigterm') + except Exception: + pass def stop(self, kill=False, timeout=15, postgres=False): if postgres: @@ -178,15 +191,16 @@ class PatroniController(AbstractController): config['postgresql']['listen'] = config['postgresql']['connect_address'] = '{0}:{1}'.format(host, self.__PORT) config['name'] = name - config['postgresql']['data_dir'] = self._data_dir + config['postgresql']['data_dir'] = self._data_dir.replace('\\', '/') config['postgresql']['basebackup'] = [{'checkpoint': 'fast'}] config['postgresql']['use_unix_socket'] = os.name != 'nt' # windows doesn't yet support unix-domain sockets config['postgresql']['use_unix_socket_repl'] = os.name != 'nt' - config['postgresql']['pgpass'] = os.path.join(tempfile.gettempdir(), 'pgpass_' + name) + config['postgresql']['pgpass'] = os.path.join(tempfile.gettempdir(), 'pgpass_' + name).replace('\\', '/') config['postgresql']['parameters'].update({ - 'logging_collector': 'on', 'log_destination': 'csvlog', 'log_directory': self._output_dir, + 'logging_collector': 'on', 'log_destination': 'csvlog', + 'log_directory': self._output_dir.replace('\\', '/'), 'log_filename': name + '.log', 'log_statement': 'all', 'log_min_messages': 'debug1', - 'unix_socket_directories': tempfile.gettempdir()}) + 'unix_socket_directories': tempfile.gettempdir().replace('\\', '/')}) if 'bootstrap' in config: config['bootstrap']['post_bootstrap'] = 'psql -w -c "SELECT 1"' @@ -210,6 +224,7 @@ class PatroniController(AbstractController): self._replication = config['postgresql'].get('authentication', config['postgresql']).get('replication', {}) self._replication.update({'host': host, 'port': self.__PORT, 'dbname': 'postgres'}) + self._restapi_url = 'http://{0}'.format(config['restapi']['connect_address']) return patroni_config_path @@ -394,7 +409,7 @@ class AbstractEtcdController(AbstractDcsController): self._client_cls = client_cls def _start(self): - return subprocess.Popen(["etcd", "--debug", "--data-dir", self._work_directory], + return subprocess.Popen(["etcd", "--data-dir", self._work_directory], stdout=self._log, stderr=subprocess.STDOUT) def _is_running(self): @@ -634,8 +649,9 @@ class RaftController(AbstractDcsController): class PatroniPoolController(object): - BACKUP_SCRIPT = [sys.executable, 'features/backup_create.py'] - ARCHIVE_RESTORE_SCRIPT = ' '.join((sys.executable, os.path.abspath('features/archive-restore.py'))) + PYTHON = sys.executable.replace('\\', '/') + BACKUP_SCRIPT = [PYTHON, 'features/backup_create.py'] + ARCHIVE_RESTORE_SCRIPT = ' '.join((PYTHON, os.path.abspath('features/archive-restore.py'))) def __init__(self, context): self._context = context @@ -711,7 +727,7 @@ class PatroniPoolController(object): 'archive_mode': 'on', 'archive_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode archive ' + '--dirname {} --filename %f --pathname %p').format( - os.path.join(self.patroni_path, 'data', 'wal_archive')) + os.path.join(self.patroni_path, 'data', 'wal_archive').replace('\\', '/')) }, 'authentication': { 'superuser': {'password': 'zalando1'}, @@ -727,14 +743,14 @@ class PatroniPoolController(object): 'bootstrap': { 'method': 'backup_restore', 'backup_restore': { - 'command': (sys.executable + ' features/backup_restore.py --sourcedir=' + - os.path.join(self.patroni_path, 'data', 'basebackup')), + 'command': (self.PYTHON + ' features/backup_restore.py --sourcedir=' + + os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')), 'recovery_conf': { 'recovery_target_action': 'promote', 'recovery_target_timeline': 'latest', 'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore ' + '--dirname {} --filename %f --pathname %p').format( - os.path.join(self.patroni_path, 'data', 'wal_archive')) + os.path.join(self.patroni_path, 'data', 'wal_archive').replace('\\', '/')) } } }, @@ -874,7 +890,10 @@ class WatchdogMonitor(object): # actions to execute on start/stop of the tests and before running individual features def before_all(context): os.environ.update({'PATRONI_RESTAPI_USERNAME': 'username', 'PATRONI_RESTAPI_PASSWORD': 'password'}) - context.ci = any(a in os.environ for a in ('TRAVIS_BUILD_NUMBER', 'BUILD_NUMBER', 'GITHUB_ACTIONS')) + context.request_executor = PatroniRequest({'ctl': {'auth': os.environ['PATRONI_RESTAPI_USERNAME'] + + ':' + os.environ['PATRONI_RESTAPI_PASSWORD']}}) + context.ci = os.name == 'nt' or\ + any(a in os.environ for a in ('TRAVIS_BUILD_NUMBER', 'BUILD_NUMBER', 'GITHUB_ACTIONS')) context.timeout_multiplier = 5 if context.ci else 1 # MacOS sometimes is VERY slow context.pctl = PatroniPoolController(context) context.dcs_ctl = context.pctl.known_dcs[context.pctl.dcs](context) @@ -894,13 +913,18 @@ def after_all(context): def before_feature(context, feature): """ create per-feature output directory to collect Patroni and PostgreSQL logs """ - context.pctl.create_and_set_output_directory(feature.name) + if feature.name == 'watchdog' and os.name == 'nt': + feature.skip("Watchdog isn't supported on Windows") + else: + context.pctl.create_and_set_output_directory(feature.name) def after_feature(context, feature): """ stop all Patronis, remove their data directory and cleanup the keys in etcd """ context.pctl.stop_all() - shutil.rmtree(os.path.join(context.pctl.patroni_path, 'data')) + data = os.path.join(context.pctl.patroni_path, 'data') + if os.path.exists(data): + shutil.rmtree(data) context.dcs_ctl.cleanup_service_tree() if feature.status == 'failed': shutil.copytree(context.pctl.output_dir, context.pctl.output_dir + '_failed') diff --git a/features/patroni_api.feature b/features/patroni_api.feature index ade92200..2d663643 100644 --- a/features/patroni_api.feature +++ b/features/patroni_api.feature @@ -109,7 +109,7 @@ Scenario: check the scheduled switchover And I receive a response output "Can't schedule switchover in the paused state" When I run patronictl.py resume batman Then I receive a response returncode 0 - Given I issue a scheduled switchover from postgres1 to postgres0 in 5 seconds + Given I issue a scheduled switchover from postgres1 to postgres0 in 10 seconds Then I receive a response returncode 0 And postgres0 is a leader after 20 seconds And postgres0 role is the primary after 10 seconds diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py index c0523ba3..31e2a184 100644 --- a/features/steps/patroni_api.py +++ b/features/steps/patroni_api.py @@ -10,10 +10,8 @@ import yaml from behave import register_type, step, then from dateutil import tz from datetime import datetime, timedelta -from patroni.request import PatroniRequest tzutc = tz.tzutc() -request_executor = PatroniRequest({'ctl': {'auth': 'username:password'}}) @parse.with_pattern(r'https?://(?:\w|\.|:|/)+') @@ -75,9 +73,9 @@ def do_post_empty(context, url): def do_request(context, request_method, url, data): data = data and json.loads(data) try: - r = request_executor.request(request_method, url, data) + r = context.request_executor.request(request_method, url, data) if request_method == 'PATCH' and r.status == 409: - r = request_executor.request(request_method, url, data) + r = context.request_executor.request(request_method, url, data) except Exception: context.status_code = context.response = None else: @@ -139,7 +137,7 @@ def add_tag_to_config(context, tag, value, pg_name): def check_http_response(context, url, value, timeout, negate=False): timeout *= context.timeout_multiplier for _ in range(int(timeout)): - r = request_executor.request('GET', url) + r = context.request_executor.request('GET', url) if (value in r.data.decode('utf-8')) != negate: break time.sleep(1) diff --git a/features/steps/standby_cluster.py b/features/steps/standby_cluster.py index c3ce80e2..19256c26 100644 --- a/features/steps/standby_cluster.py +++ b/features/steps/standby_cluster.py @@ -1,17 +1,12 @@ import os -import sys import time from behave import step -select_replication_query = """ -SELECT * FROM pg_catalog.pg_stat_replication -WHERE application_name = '{0}' -""" - -executable = sys.executable if os.name != 'nt' else sys.executable.replace('\\', '/') -callback = executable + " features/callback2.py " +def callbacks(context, name): + return {c: '{0} features/callback2.py {1}'.format(context.pctl.PYTHON, name) + for c in ('on_start', 'on_stop', 'on_restart', 'on_role_change')} @step('I start {name:w} in a cluster {cluster_name:w}') @@ -19,10 +14,10 @@ def start_patroni(context, name, cluster_name): return context.pctl.start(name, custom_config={ "scope": cluster_name, "postgresql": { - "callbacks": {c: callback + name for c in ('on_start', 'on_stop', 'on_restart', 'on_role_change')}, + "callbacks": callbacks(context, name), "backup_restore": { - "command": (executable + " features/backup_restore.py --sourcedir=" + - os.path.join(context.pctl.patroni_path, 'data', 'basebackup'))} + "command": (context.pctl.PYTHON + " features/backup_restore.py --sourcedir=" + + os.path.join(context.pctl.patroni_path, 'data', 'basebackup').replace('\\', '/'))} } }) @@ -49,7 +44,7 @@ def start_patroni_standby_cluster(context, name, cluster_name, name2): } }, "postgresql": { - "callbacks": {c: callback + name for c in ('on_start', 'on_stop', 'on_restart', 'on_role_change')} + "callbacks": callbacks(context, name) } }) return context.pctl.start(name) @@ -62,7 +57,7 @@ def check_replication_status(context, pg_name1, pg_name2, timeout): while time.time() < bound_time: cur = context.pctl.query( pg_name2, - select_replication_query.format(pg_name1), + "SELECT * FROM pg_catalog.pg_stat_replication WHERE application_name = '{0}'".format(pg_name1), fail_ok=True ) diff --git a/patroni/api.py b/patroni/api.py index 5acd1fa1..657744fd 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -368,6 +368,14 @@ class RestApiHandler(BaseHTTPRequestHandler): self.server.patroni.sighup_handler() self._write_response(202, 'reload scheduled') + @check_access + def do_POST_sigterm(self): + """Only for behave testing on windows""" + + if os.name == 'nt' and os.getenv('BEHAVE_DEBUG'): + self.server.patroni.api_sigterm() + self._write_response(202, 'shutdown scheduled') + @staticmethod def parse_schedule(schedule, action): """ parses the given schedule and validates at """ diff --git a/patroni/daemon.py b/patroni/daemon.py index ca118acf..3f4f93ac 100644 --- a/patroni/daemon.py +++ b/patroni/daemon.py @@ -24,11 +24,15 @@ class AbstractPatroniDaemon(object): def sighup_handler(self, *args): self._received_sighup = True - def sigterm_handler(self, *args): + def api_sigterm(self): with self._sigterm_lock: if not self._received_sigterm: self._received_sigterm = True - sys.exit() + return True + + def sigterm_handler(self, *args): + if self.api_sigterm(): + sys.exit() def setup_signal_handlers(self): self._received_sighup = False diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index d536baca..5be6839d 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -216,10 +216,13 @@ class AbstractEtcdClientWithFailover(etcd.Client): return response except (HTTPError, HTTPException, socket.error, socket.timeout) as e: self.http.clear() - # switch to the next etcd node because we don't know exactly what happened, - # whether the key didn't received an update or there is a network problem. - if not retry and i + 1 < len(machines_cache): - self.set_base_uri(machines_cache[i + 1]) + if not retry: + if len(machines_cache) == 1: + self.set_base_uri(self._base_uri) # trigger Etcd3 watcher restart + # switch to the next etcd node because we don't know exactly what happened, + # whether the key didn't received an update or there is a network problem. + elif i + 1 < len(machines_cache): + self.set_base_uri(machines_cache[i + 1]) if (isinstance(fields, dict) and fields.get("wait") == "true" and isinstance(e, (ReadTimeoutError, ProtocolError))): logger.debug("Watch timed out.") diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 5437a255..a507c421 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -11,6 +11,7 @@ import time import urllib3 from threading import Condition, Lock, Thread +from urllib3.exceptions import ReadTimeoutError, ProtocolError from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors @@ -350,7 +351,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover): def deleteprefix(self, key, retry=None): return self.deleterange(key, prefix_range_end(key), retry=retry) - def watchrange(self, key, range_end=None, start_revision=None, filters=None): + def watchrange(self, key, range_end=None, start_revision=None, filters=None, read_timeout=None): """returns: response object""" params = build_range_request(key, range_end) if start_revision is not None: @@ -358,11 +359,11 @@ class Etcd3Client(AbstractEtcdClientWithFailover): params['filters'] = filters or [] kwargs = self._prepare_common_parameters(1, self.read_timeout) request_executor = self._prepare_request(kwargs, {'create_request': params}) - kwargs.update(timeout=urllib3.Timeout(connect=kwargs['timeout']), retries=0) + kwargs.update(timeout=urllib3.Timeout(connect=kwargs['timeout'], read=read_timeout), retries=0) return request_executor(self._MPOST, self._base_uri + self.version_prefix + '/watch', **kwargs) - def watchprefix(self, key, start_revision=None, filters=None): - return self.watchrange(key, prefix_range_end(key), start_revision, filters) + def watchprefix(self, key, start_revision=None, filters=None, read_timeout=None): + return self.watchrange(key, prefix_range_end(key), start_revision, filters, read_timeout) class KVCache(Thread): @@ -451,7 +452,14 @@ class KVCache(Thread): def _do_watch(self, revision): with self._response_lock: self._response = None - response = self._client.watchprefix(self._dcs.cluster_prefix, revision) + # We do most of requests with timeouts. The only exception /watch requests to Etcd v3. + # In order to interrupt the /watch request we do socket.shutdown() from the main thread, + # which doesn't work on Windows. Therefore we want to use the last resort, `read_timeout`. + # Setting it to TTL will help to partially mitigate the problem. + # Setting it to lower value is not nice because for idling clusters it will increase + # the numbers of interrupts and reconnects. + read_timeout = self._dcs.ttl if os.name == 'nt' else None + response = self._client.watchprefix(self._dcs.cluster_prefix, revision, read_timeout=read_timeout) with self._response_lock: if self._response is None: self._response = response @@ -473,7 +481,9 @@ class KVCache(Thread): try: self._do_watch(result['header']['revision']) except Exception as e: - logger.error('watchprefix failed: %r', e) + # Following exceptions are expected on Windows because the /watch request is done with `read_timeout` + if not (os.name == 'nt' and isinstance(e, (ReadTimeoutError, ProtocolError))): + logger.error('watchprefix failed: %r', e) finally: with self.condition: self._is_ready = False diff --git a/tests/test_api.py b/tests/test_api.py index 641871af..a1db15f3 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -134,6 +134,10 @@ class MockPatroni(object): def sighup_handler(): pass + @staticmethod + def api_sigterm(): + pass + class MockRequest(object): @@ -371,6 +375,11 @@ class TestRestApiHandler(unittest.TestCase): def test_do_POST_reload(self): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /reload HTTP/1.0' + self._authorization)) + @patch('os.environ', {'BEHAVE_DEBUG': 'true'}) + @patch('os.name', 'nt') + def test_do_POST_sigterm(self): + self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /sigterm HTTP/1.0' + self._authorization)) + @patch.object(MockPatroni, 'dcs') def test_do_POST_restart(self, mock_dcs): mock_dcs.get_cluster.return_value.is_paused.return_value = False diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 8df03dfd..ed6e68a1 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -562,7 +562,8 @@ class TestCtl(unittest.TestCase): @patch('sys.stdout.isatty', return_value=False) @patch('patroni.ctl.markup_to_pager') - def test_show_diff(self, mock_markup_to_pager, mock_isatty): + @patch('patroni.ctl.find_executable', return_value=None) + def test_show_diff(self, mock_find_executable, mock_markup_to_pager, mock_isatty): show_diff("foo:\n bar: 1\n", "foo:\n bar: 2\n") mock_markup_to_pager.assert_not_called() @@ -570,10 +571,10 @@ class TestCtl(unittest.TestCase): show_diff("foo:\n bar: 1\n", "foo:\n bar: 2\n") mock_markup_to_pager.assert_called_once() - with patch('patroni.ctl.find_executable', Mock(return_value=None)): - show_diff("foo:\n bar: 1\n", "foo:\n bar: 2\n") + show_diff("foo:\n bar: 1\n", "foo:\n bar: 2\n") # Test that unicode handling doesn't fail with an exception + mock_find_executable.return_value = '/usr/bin/less' show_diff(b"foo:\n bar: \xc3\xb6\xc3\xb6\n".decode('utf-8'), b"foo:\n bar: \xc3\xbc\xc3\xbc\n".decode('utf-8')) @@ -591,6 +592,7 @@ class TestCtl(unittest.TestCase): self.runner.invoke(ctl, ['show-config', 'dummy']) @patch('patroni.ctl.get_dcs') + @patch('subprocess.call', Mock(return_value=0)) def test_edit_config(self, mock_get_dcs): mock_get_dcs.return_value = self.e mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 44e4c0c9..ee79e0e7 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -50,6 +50,7 @@ class MockFrozenImporter(object): @patch.object(etcd.Client, 'read', etcd_read) class TestPatroni(unittest.TestCase): + @patch('sys.argv', ['patroni.py']) def test_no_config(self): self.assertRaises(SystemExit, patroni_main) @@ -57,6 +58,8 @@ class TestPatroni(unittest.TestCase): @patch('socket.socket.connect_ex', Mock(return_value=1)) def test_validate_config(self): self.assertRaises(SystemExit, patroni_main) + with patch.object(config.Config, '__init__', Mock(return_value=None)): + self.assertRaises(SystemExit, patroni_main) @patch('pkgutil.iter_importers', Mock(return_value=[MockFrozenImporter()])) @patch('sys.frozen', Mock(return_value=True), create=True) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 055ee2e0..9ee2d9a2 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -454,24 +454,20 @@ class TestPostgresql(BaseTestPostgresql): def test_get_postgres_role_from_data_directory(self): self.assertEqual(self.p.get_postgres_role_from_data_directory(), 'replica') + @patch('os.remove', Mock()) + @patch('shutil.rmtree', Mock()) + @patch('os.unlink', Mock(side_effect=OSError)) + @patch('os.path.isdir', Mock(return_value=True)) + @patch('os.path.exists', Mock(return_value=True)) def test_remove_data_directory(self): - def _symlink(src, dst): - if os.name != 'nt': # os.symlink under Windows needs admin rights skip it - os.symlink(src, dst) - - os.makedirs(os.path.join(self.p.data_dir, 'foo')) - _symlink('foo', os.path.join(self.p.data_dir, 'pg_wal')) - os.makedirs(os.path.join(self.p.data_dir, 'foo_tsp')) - pg_tblspc = os.path.join(self.p.data_dir, 'pg_tblspc') - os.makedirs(pg_tblspc) - _symlink('../foo_tsp', os.path.join(pg_tblspc, '12345')) - self.p.remove_data_directory() - open(self.p.data_dir, 'w').close() - self.p.remove_data_directory() - _symlink('unexisting', self.p.data_dir) - with patch('os.unlink', Mock(side_effect=OSError)): + with patch('os.path.islink', Mock(return_value=True)): + self.p.remove_data_directory() + with patch('os.path.isfile', Mock(return_value=True)): + self.p.remove_data_directory() + with patch('os.path.islink', Mock(side_effect=[False, False, True, True])),\ + patch('os.listdir', Mock(return_value=['12345'])),\ + patch('os.path.realpath', Mock(side_effect=['../foo', '../foo_tsp'])): self.p.remove_data_directory() - self.p.remove_data_directory() @patch('patroni.postgresql.Postgresql._version_file_exists', Mock(return_value=True)) def test_controldata(self):