diff --git a/features/patroni_api.feature b/features/patroni_api.feature index 4351bab9..e8b9eb5d 100644 --- a/features/patroni_api.feature +++ b/features/patroni_api.feature @@ -70,6 +70,7 @@ Scenario: check the switchover via the API in the pause mode And postgres1 role is the primary after 10 seconds And postgres0 role is the secondary after 10 seconds And replication works from postgres1 to postgres0 after 20 seconds + And "members/postgres0" key in DCS has state=running after 10 seconds When I issue a GET request to http://127.0.0.1:8008/master Then I receive a response code 503 When I issue a GET request to http://127.0.0.1:8008/replica @@ -91,6 +92,7 @@ Scenario: check the scheduled switchover And postgres0 role is the primary after 10 seconds And postgres1 role is the secondary after 10 seconds And replication works from postgres0 to postgres1 after 25 seconds + And "members/postgres1" key in DCS has state=running after 10 seconds When I issue a GET request to http://127.0.0.1:8008/master Then I receive a response code 200 When I issue a GET request to http://127.0.0.1:8008/replica diff --git a/patroni/api.py b/patroni/api.py index b14347c4..86e46d4e 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -13,7 +13,7 @@ import socket from patroni.postgresql import PostgresConnectionException from patroni.postgresql.misc import postgres_version_to_int, PostgresException from patroni.utils import deep_compare, parse_bool, patch_config, Retry, \ - RetryFailedError, parse_int, split_host_port, tzutc, uri + RetryFailedError, parse_int, split_host_port, tzutc, uri, cluster_as_json from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer from six.moves.socketserver import ThreadingMixIn from threading import Thread @@ -133,6 +133,14 @@ class RestApiHandler(BaseHTTPRequestHandler): response = self.get_postgresql_status(True) self._write_status_response(200, response) + def do_GET_cluster(self): + cluster = self.server.patroni.dcs.cluster or self.server.patroni.dcs.get_cluster() + self._write_json_response(200, cluster_as_json(cluster)) + + def do_GET_history(self): + cluster = self.server.patroni.dcs.cluster or self.server.patroni.dcs.get_cluster() + self._write_json_response(200, cluster.history and cluster.history.lines or []) + def do_GET_config(self): cluster = self.server.patroni.dcs.cluster or self.server.patroni.dcs.get_cluster() if cluster.config: @@ -415,10 +423,7 @@ class RestApiHandler(BaseHTTPRequestHandler): if self.server.patroni.postgresql.state not in ('running', 'restarting', 'starting'): raise RetryFailedError('') - stmt = ("WITH replication_info AS (" - "SELECT usename, application_name, client_addr, state, sync_state, sync_priority" - " FROM pg_catalog.pg_stat_replication) SELECT" - " pg_catalog.to_char(pg_catalog.pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ')," + stmt = ("SELECT pg_catalog.to_char(pg_catalog.pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ')," " CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0" " ELSE ('x' || pg_catalog.substr(pg_catalog.pg_{0}file_name(" "pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END," @@ -429,8 +434,10 @@ class RestApiHandler(BaseHTTPRequestHandler): " pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint," " pg_catalog.to_char(pg_catalog.pg_last_xact_replay_timestamp(), 'YYYY-MM-DD HH24:MI:SS.MS TZ')," " pg_catalog.pg_is_in_recovery() AND pg_catalog.pg_is_{0}_replay_paused(), " - "(SELECT pg_catalog.array_to_json(pg_catalog.array_agg(" - "pg_catalog.row_to_json(ri))) FROM replication_info ri)") + " pg_catalog.array_to_json(pg_catalog.array_agg(pg_catalog.row_to_json(ri))) " + "FROM (SELECT (SELECT rolname FROM pg_authid WHERE oid = usesysid) AS usename," + " application_name, client_addr, w.state, sync_state, sync_priority" + " FROM pg_catalog.pg_stat_get_wal_senders() w, pg_catalog.pg_stat_get_activity(pid)) AS ri") row = self.query(stmt.format(self.server.patroni.postgresql.wal_name, self.server.patroni.postgresql.lsn_name), retry=retry)[0] diff --git a/patroni/ctl.py b/patroni/ctl.py index fe2a6828..ffc9fc11 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -14,6 +14,7 @@ import json import logging import os import random +import six import subprocess import sys import tempfile @@ -28,12 +29,11 @@ from patroni.dcs import get_dcs as _get_dcs from patroni.exceptions import PatroniException from patroni.postgresql import Postgresql from patroni.postgresql.misc import postgres_version_to_int +from patroni.utils import cluster_as_json, patch_config, polling_loop from patroni.request import PatroniRequest -from patroni.utils import patch_config, polling_loop from patroni.version import __version__ from prettytable import PrettyTable from six.moves.urllib_parse import urlparse -from six import text_type CONFIG_DIR_PATH = click.get_app_dir('patroni') CONFIG_FILE_PATH = os.path.join(CONFIG_DIR_PATH, 'patronictl.yaml') @@ -705,80 +705,51 @@ def switchover(obj, cluster_name, master, candidate, force, scheduled): def output_members(cluster, name, extended=False, fmt='pretty'): rows = [] logging.debug(cluster) - leader_name = None - if cluster.leader: - leader_name = cluster.leader.name - - xlog_location_cluster = cluster.last_leader_operation or 0 - - # Mainly for consistent pretty printing and watching we sort the output - cluster.members.sort(key=lambda x: x.name) - - has_scheduled_restarts = any(m.data.get('scheduled_restart') for m in cluster.members) - has_pending_restarts = any(m.data.get('pending_restart') for m in cluster.members) - - # Show Host as 'host:port' if somebody is running on non-standard port or two nodes are running on the same host - append_port = any(str(m.conn_kwargs()['port']) != '5432' for m in cluster.members) or\ - len(set(m.conn_kwargs()['host'] for m in cluster.members)) < len(cluster.members) - - for m in cluster.members: - logging.debug(m) - - role = '' - if m.name == leader_name: - role = 'Leader' - elif m.name == cluster.sync.sync_standby: - role = 'Sync standby' - - xlog_location = m.data.get('xlog_location') - lag = '' - if xlog_location is None: - lag = 'unknown' - elif xlog_location_cluster >= xlog_location: - lag = round((xlog_location_cluster - xlog_location)/1024/1024) - - host = m.conn_kwargs()['host'] - if append_port: - host += ':{0}'.format(m.conn_kwargs()['port']) - - row = [name, m.name, host, role, m.data.get('state', ''), m.data.get('timeline', ''), lag] - - if extended or has_pending_restarts: - row.append('*' if m.data.get('pending_restart') else '') - - if extended or has_scheduled_restarts: - value = '' - scheduled_restart = m.data.get('scheduled_restart') - if scheduled_restart: - value = scheduled_restart['schedule'] - if 'postgres_version' in scheduled_restart: - value += ' if version < {0}'.format(scheduled_restart['postgres_version']) - - row.append(value) - - rows.append(row) + cluster = cluster_as_json(cluster) columns = ['Cluster', 'Member', 'Host', 'Role', 'State', 'TL', 'Lag in MB'] - alignment = {'Lag in MB': 'r', 'TL': 'r'} + for c in ('Pending restart', 'Scheduled restart'): + if extended or any(m.get(c.lower().replace(' ', '_')) for m in cluster['members']): + columns.append(c) - if extended or has_pending_restarts: - columns.append('Pending restart') + # Show Host as 'host:port' if somebody is running on non-standard port or two nodes are running on the same host + append_port = any(m['port'] != 5432 for m in cluster['members']) or\ + len(set(m['host'] for m in cluster['members'])) < len(cluster['members']) - if extended or has_scheduled_restarts: - columns.append('Scheduled restart') + for m in cluster['members']: + logging.debug(m) - print_output(columns, rows, alignment, fmt) + lag = m.get('lag', '') + m.update(cluster=name, member=m['name'], tl=m.get('timeline', ''), + role='' if m['role'] == 'replica' else m['role'].replace('_', ' ').title(), + lag_in_mb=round(lag/1024/1024) if isinstance(lag, six.integer_types) else lag, + pending_restart='*' if m.get('pending_restart') else '') + + if append_port: + m['host'] = ':'.join([m['host'], str(m['port'])]) + + if 'scheduled_restart' in m: + value = m['scheduled_restart']['schedule'] + if 'postgres_version' in m['scheduled_restart']: + value += ' if version < {0}'.format(m['scheduled_restart']['postgres_version']) + m['scheduled_restart'] = value + + rows.append([m.get(n.lower().replace(' ', '_'), '') for n in columns]) + + print_output(columns, rows, {'Lag in MB': 'r', 'TL': 'r'}, fmt) + + if fmt != 'pretty': # Omit service info when using machine-readable formats + return service_info = [] - if cluster.is_paused(): + if cluster.get('pause'): service_info.append('Maintenance mode: on') - if cluster.failover and cluster.failover.scheduled_at: - info = 'Switchover scheduled at: ' + cluster.failover.scheduled_at.isoformat() - if cluster.failover.leader: - info += '\n from: ' + cluster.failover.leader - if cluster.failover.candidate: - info += '\n to: ' + cluster.failover.candidate + if 'scheduled_switchover' in cluster: + info = 'Switchover scheduled at: ' + cluster['scheduled_switchover']['at'] + for name in ('from', 'to'): + if name in cluster['scheduled_switchover']: + info += '\n{0:>24}: {1}'.format(name, cluster['scheduled_switchover'][name]) service_info.append(info) if service_info: @@ -998,7 +969,7 @@ def show_diff(before_editing, after_editing): buf = io.StringIO() for line in unified_diff: # Force cast to unicode as difflib on Python 2.7 returns a mix of unicode and str. - buf.write(text_type(line)) + buf.write(six.text_type(line)) buf.seek(0) class opts: @@ -1217,6 +1188,19 @@ def version(obj, cluster_name, member_names): click.echo("{0}: failed to get version: {1}".format(m.name, e)) +@ctl.command('history', help="Show the history of failovers/switchovers") +@arg_cluster_name +@option_format +@click.pass_obj +def history(obj, cluster_name, fmt): + cluster = get_dcs(obj, cluster_name).get_cluster() + history = cluster.history and cluster.history.lines or [] + for line in history: + if len(line) < 4: + line.append('') + print_output(['TL', 'LSN', 'Reason', 'Timestamp'], history, {'TL': 'r', 'LSN': 'r'}, fmt) + + def format_pg_version(version): if version < 100000: return "{0}.{1}.{2}".format(version // 10000, version // 100 % 100, version % 100) diff --git a/patroni/ha.py b/patroni/ha.py index e2a3828a..1002f440 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -14,7 +14,7 @@ from patroni.exceptions import DCSError, PostgresConnectionException, PatroniExc from patroni.postgresql import ACTION_ON_START, ACTION_ON_ROLE_CHANGE from patroni.postgresql.misc import postgres_version_to_int from patroni.postgresql.rewind import Rewind -from patroni.utils import polling_loop, tzutc +from patroni.utils import polling_loop, tzutc, is_standby_cluster as _is_standby_cluster from patroni.dcs import RemoteMember from threading import RLock @@ -108,9 +108,7 @@ class Ha(object): return config.get('standby_cluster') def is_standby_cluster(self): - config = self.get_standby_cluster_config() - # Check whether or not provided configuration describes a standby cluster - return isinstance(config, dict) and (config.get('host') or config.get('port') or config.get('restore_command')) + return _is_standby_cluster(self.get_standby_cluster_config()) def is_leader(self): with self._is_leader_lock: @@ -580,6 +578,7 @@ class Ha(object): def _is_healthiest_node(self, members, check_replication_lag=True): """This method tries to determine whether I am healthy enough to became a new leader candidate or not.""" + # We don't call `last_operation()` here because it returns a string _, my_wal_position = self.state_handler.timeline_wal_position() if check_replication_lag and self.is_lagging(my_wal_position): logger.info('My wal position exceeds maximum replication lag') diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 0deb7a16..ba5eb9da 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -37,15 +37,6 @@ STATE_UNKNOWN = 'unknown' STOP_POLLING_INTERVAL = 1 -cluster_info_query = ("SELECT CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 " - "ELSE ('x' || pg_catalog.substr(pg_catalog.pg_{0}file_name(" - "pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, " - "CASE WHEN pg_catalog.pg_is_in_recovery() THEN GREATEST(" - " pg_catalog.pg_{0}_{1}_diff(COALESCE(" - "pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint," - " pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint)" - "ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), '0/0')::bigint END") - @contextmanager def null_context(): @@ -139,6 +130,18 @@ class Postgresql(object): def lsn_name(self): return 'lsn' if self._major_version >= 100000 else 'location' + @property + def cluster_info_query(self): + return ("SELECT CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 " + "ELSE ('x' || pg_catalog.substr(pg_catalog.pg_{0}file_name(" + "pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, " + "CASE WHEN pg_catalog.pg_is_in_recovery() THEN GREATEST(" + " pg_catalog.pg_{0}_{1}_diff(COALESCE(" + "pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint," + " pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint)" + "ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), '0/0')::bigint " + "END").format(self.wal_name, self.lsn_name) + def _version_file_exists(self): return not self.data_directory_empty() and os.path.isfile(self._version_file) @@ -275,9 +278,8 @@ class Postgresql(object): def _cluster_info_state_get(self, name): if not self._cluster_info_state: - stmt = cluster_info_query.format(self.wal_name, self.lsn_name) try: - result = self._is_leader_retry(self._query, stmt).fetchone() + result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone() self._cluster_info_state = dict(zip(['timeline', 'wal_position'], result)) except RetryFailedError as e: # SELECT failed two times self._cluster_info_state = {'error': str(e)} @@ -721,7 +723,7 @@ class Postgresql(object): return self._cluster_info_state_get('timeline'), self._cluster_info_state_get('wal_position') with self.connection().cursor() as cursor: - cursor.execute(cluster_info_query.format(self.wal_name, self.lsn_name)) + cursor.execute(self.cluster_info_query) return cursor.fetchone()[:2] def postmaster_start_time(self): diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 9bcea92f..9951148c 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -118,7 +118,7 @@ def parse_dsn(value): >>> r == {'application_name': 'mya/pp', 'host': ',/host2', 'sslmode': 'require',\ 'password': 'pass', 'port': '/123', 'user': 'u/se'} True - >>> r = parse_dsn(" host = 'host' dbname = db\\ name requiressl=1 ") + >>> r = parse_dsn(" host = 'host' dbname = db\\\\ name requiressl=1 ") >>> r == {'host': 'host', 'sslmode': 'require'} True >>> parse_dsn('requiressl = 0\\\\') == {'sslmode': 'prefer'} diff --git a/patroni/utils.py b/patroni/utils.py index 6fd48ba0..8ecb8786 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -352,3 +352,52 @@ def uri(proto, netloc, path='', user=None): path = '/{0}'.format(path) if path and not path.startswith('/') else path user = '{0}@'.format(user) if user else '' return '{0}://{1}{2}{3}{4}'.format(proto, user, host, port, path) + + +def is_standby_cluster(config): + # Check whether or not provided configuration describes a standby cluster + return isinstance(config, dict) and (config.get('host') or config.get('port') or config.get('restore_command')) + + +def cluster_as_json(cluster): + leader_name = cluster.leader.name if cluster.leader else None + xlog_location_cluster = cluster.last_leader_operation or 0 + + ret = {'members': []} + for m in cluster.members: + if m.name == leader_name: + config = cluster.config.data if cluster.config and cluster.config.modify_index else {} + role = 'standby_leader' if is_standby_cluster(config.get('standby_cluster')) else 'leader' + elif m.name == cluster.sync.sync_standby: + role = 'sync_standby' + else: + role = 'replica' + + conn_kwargs = m.conn_kwargs() + member = {'name': m.name, 'host': conn_kwargs['host'], 'port': int(conn_kwargs['port']), + 'role': role, 'state': m.data.get('state', ''), 'api_url': m.api_url} + optional_attributes = ('timeline', 'pending_restart', 'scheduled_restart', 'tags') + member.update({n: m.data[n] for n in optional_attributes if n in m.data}) + + if m.name != leader_name: + xlog_location = m.data.get('xlog_location') + if xlog_location is None: + member['lag'] = 'unknown' + elif xlog_location_cluster >= xlog_location: + member['lag'] = xlog_location_cluster - xlog_location + else: + member['lag'] = 0 + + ret['members'].append(member) + + # sort members by name for consistency + ret['members'].sort(key=lambda m: m['name']) + if cluster.is_paused(): + ret['pause'] = True + if cluster.failover and cluster.failover.scheduled_at: + ret['scheduled_switchover'] = {'at': cluster.failover.scheduled_at.isoformat()} + if cluster.failover.leader: + ret['scheduled_switchover']['from'] = cluster.failover.leader + if cluster.failover.candidate: + ret['scheduled_switchover']['to'] = cluster.failover.candidate + return ret diff --git a/tests/__init__.py b/tests/__init__.py index 14d9d4a5..2f2118ba 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -100,7 +100,7 @@ class MockCursor(object): self.results = [(1, 2)] elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'): self.results = [(False, 2)] - elif sql.startswith('WITH replication_info AS ('): + elif sql.startswith('SELECT pg_catalog.to_char'): replication_info = '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' +\ '"state":"streaming","sync_state":"async","sync_priority":0}]' self.results = [('', 0, '', '', '', '', False, replication_info)] diff --git a/tests/test_api.py b/tests/test_api.py index 7e4d5b12..a6230ecd 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -12,6 +12,7 @@ from patroni.utils import tzutc from six import BytesIO as IO from six.moves import BaseHTTPServer from . import psycopg2_connect, MockCursor +from .test_ha import get_cluster_initialized_without_leader future_restart_time = datetime.datetime.now(tzutc) + datetime.timedelta(days=5) @@ -159,6 +160,7 @@ class TestRestApiHandler(unittest.TestCase): def test_do_GET(self): MockRestApiServer(RestApiHandler, 'GET /replica') + MockRestApiServer(RestApiHandler, 'GET /read-only') with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={})): MockRestApiServer(RestApiHandler, 'GET /replica') with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})): @@ -197,6 +199,17 @@ class TestRestApiHandler(unittest.TestCase): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /restart HTTP/1.0')) MockRestApiServer(RestApiHandler, 'POST /restart HTTP/1.0\nAuthorization:') + @patch.object(MockPatroni, 'dcs') + def test_do_GET_cluster(self, mock_dcs): + mock_dcs.cluster = get_cluster_initialized_without_leader() + mock_dcs.cluster.members[1].data['xlog_location'] = 11 + self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /cluster')) + + @patch.object(MockPatroni, 'dcs') + def test_do_GET_history(self, mock_dcs): + mock_dcs.cluster = get_cluster_initialized_without_leader() + self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /history')) + @patch.object(MockPatroni, 'dcs') def test_do_GET_config(self, mock_dcs): mock_dcs.cluster.config.data = {} diff --git a/tests/test_ctl.py b/tests/test_ctl.py index a190ef3e..0733d88f 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -566,6 +566,13 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['version', 'dummy']) assert 'failed to get version' in result.output + @patch('patroni.ctl.get_dcs') + def test_history(self, mock_get_dcs): + mock_get_dcs.return_value.get_cluster = Mock() + mock_get_dcs.return_value.get_cluster.return_value.history.lines = [[1, 67176, 'no recovery target specified']] + result = self.runner.invoke(ctl, ['history']) + assert 'Reason' in result.output + def test_format_pg_version(self): self.assertEqual(format_pg_version(100001), '10.1') self.assertEqual(format_pg_version(90605), '9.6.5') diff --git a/tests/test_log.py b/tests/test_log.py index 76fb5704..b86ae4cf 100644 --- a/tests/test_log.py +++ b/tests/test_log.py @@ -19,7 +19,6 @@ class TestPatroniLogger(unittest.TestCase): logging.getLogger().handlers[:] = self._handlers @patch('logging.FileHandler._open', Mock()) - @patch('logging.Handler.close', Mock(side_effect=Exception)) def test_patroni_logger(self): config = { 'log': { @@ -47,13 +46,14 @@ class TestPatroniLogger(unittest.TestCase): self.assertEqual(logger.log_handler.backupCount, config['log']['file_num']) config['log'].pop('dir') - logger.reload_config(config['log']) - with patch.object(logging.Logger, 'makeRecord', - Mock(side_effect=[logging.LogRecord('', logging.INFO, '', 0, '', (), None), Exception])): + with patch('logging.Handler.close', Mock(side_effect=Exception)): + logger.reload_config(config['log']) + with patch.object(logging.Logger, 'makeRecord', + Mock(side_effect=[logging.LogRecord('', logging.INFO, '', 0, '', (), None), Exception])): + logging.error('test') logging.error('test') - logging.error('test') - with patch.object(Queue, 'put_nowait', Mock(side_effect=Full)): - self.assertRaises(SystemExit, logger.shutdown) - self.assertRaises(Exception, logger.shutdown) + with patch.object(Queue, 'put_nowait', Mock(side_effect=Full)): + self.assertRaises(SystemExit, logger.shutdown) + self.assertRaises(Exception, logger.shutdown) self.assertLessEqual(logger.queue_size, 2) # "Failed to close the old log handler" could be still in the queue self.assertEqual(logger.records_lost, 0)