From 03c2a85d23e4ab36a41c2d66b26a3c66545df5c0 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 5 Jan 2018 15:25:56 +0100 Subject: [PATCH] Expose current timeline in DCS and via API (#591) It is very easy to get current timeline on the master by executing ```sql SELECT ('x' || SUBSTR(pg_walfile_name(pg_current_wal_lsn()), 1, 8))::bit(32)::int ``` Unfortunately the same method doesn't work when postgres is_in_recovery. Therefore we will use replication connection for that on the replicas. In order to avoid opening and closing replication connection on every HA loop we will cache the result if its value matches with the timeline of the master. Also this PR introduces a new key in DCS: `/history`. It will contain a json serialized object with timeline history in a format similar to the usual history files. The differences are: * Second column is the absolute wal position in bytes, instead of LSN * Optionally there might be a fourth column - timestamp, (mtime of history file) --- patroni/api.py | 46 ++++++----- patroni/dcs/__init__.py | 35 ++++++++- patroni/dcs/consul.py | 14 +++- patroni/dcs/etcd.py | 14 +++- patroni/dcs/kubernetes.py | 13 +++- patroni/dcs/zookeeper.py | 44 ++++++----- patroni/ha.py | 42 ++++++++-- patroni/postgresql.py | 156 +++++++++++++++++++++++++------------- tests/test_api.py | 6 ++ tests/test_consul.py | 4 + tests/test_etcd.py | 3 + tests/test_ha.py | 29 +++++-- tests/test_kubernetes.py | 3 + tests/test_postgresql.py | 38 ++++++---- tests/test_zookeeper.py | 3 + 15 files changed, 323 insertions(+), 127 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index e70726dd..4e457765 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -402,40 +402,44 @@ class RestApiHandler(BaseHTTPRequestHandler): try: if self.server.patroni.postgresql.state not in ('running', 'restarting', 'starting'): raise RetryFailedError('') - row = self.query("""WITH replication_info AS ( - SELECT usename, application_name, client_addr, state, sync_state, sync_priority - FROM pg_stat_replication - ) - SELECT to_char(pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'), - pg_is_in_recovery(), - CASE WHEN pg_is_in_recovery() - THEN 0 - ELSE pg_{0}_{1}_diff(pg_current_{0}_{1}(), '0/0')::bigint - END, - pg_{0}_{1}_diff(COALESCE(pg_last_{0}_receive_{1}(), - pg_last_{0}_replay_{1}()), '0/0')::bigint, - pg_{0}_{1}_diff(pg_last_{0}_replay_{1}(), '0/0')::bigint, - to_char(pg_last_xact_replay_timestamp(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'), - pg_is_in_recovery() AND pg_is_{0}_replay_paused(), - (SELECT array_to_json(array_agg(row_to_json(ri))) - FROM replication_info ri)""".format(self.server.patroni.postgresql.wal_name, - self.server.patroni.postgresql.lsn_name), - retry=retry)[0] + stmt = ("WITH replication_info AS (" + "SELECT usename, application_name, client_addr, state, sync_state, sync_priority" + " FROM pg_stat_replication) SELECT" + " to_char(pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ')," + " CASE WHEN pg_is_in_recovery() THEN 0" + " ELSE ('x' || SUBSTR(pg_{0}file_name(pg_current_{0}_{1}()), 1, 8))::bit(32)::int END," + " CASE WHEN pg_is_in_recovery() THEN 0" + " ELSE pg_{0}_{1}_diff(pg_current_{0}_{1}(), '0/0')::bigint END," + " pg_{0}_{1}_diff(COALESCE(pg_last_{0}_receive_{1}(), pg_last_{0}_replay_{1}()), '0/0')::bigint," + " pg_{0}_{1}_diff(pg_last_{0}_replay_{1}(), '0/0')::bigint," + " to_char(pg_last_xact_replay_timestamp(), 'YYYY-MM-DD HH24:MI:SS.MS TZ')," + " pg_is_in_recovery() AND pg_is_{0}_replay_paused()," + " (SELECT array_to_json(array_agg(row_to_json(ri))) FROM replication_info ri)") + + row = self.query(stmt.format(self.server.patroni.postgresql.wal_name, + self.server.patroni.postgresql.lsn_name), retry=retry)[0] result = { 'state': self.server.patroni.postgresql.state, 'postmaster_start_time': row[0], - 'role': 'replica' if row[1] else 'master', + 'role': 'replica' if row[1] == 0 else 'master', 'server_version': self.server.patroni.postgresql.server_version, 'xlog': ({ 'received_location': row[3], 'replayed_location': row[4], 'replayed_timestamp': row[5], - 'paused': row[6]} if row[1] else { + 'paused': row[6]} if row[1] == 0 else { 'location': row[2] }) } + if row[1] > 0: + result['timeline'] = row[1] + else: + cluster = self.server.patroni.dcs.cluster + leader_timeline = None if not cluster or cluster.is_unlocked() else cluster.leader.timeline + result['timeline'] = self.server.patroni.postgresql.replica_cached_timeline(leader_timeline) + if row[7]: result['replication'] = row[7] diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 717a0801..4e450279 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -177,6 +177,10 @@ class Leader(namedtuple('Leader', 'index,session,member')): def conn_url(self): return self.member.conn_url + @property + def timeline(self): + return self.member.data.get('timeline') + class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')): @@ -297,7 +301,26 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')): return name is not None and name in (self.leader, self.sync_standby) -class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operation,members,failover,sync')): +class TimelineHistory(namedtuple('TimelineHistory', 'index,lines')): + """Object representing timeline history file""" + + @staticmethod + def from_node(index, value): + """ + >>> h = TimelineHistory.from_node(1, 2) + >>> h.lines + [] + """ + try: + lines = json.loads(value) + except (TypeError, ValueError): + lines = None + if not isinstance(lines, list): + lines = [] + return TimelineHistory(index, lines) + + +class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operation,members,failover,sync,history')): """Immutable object (namedtuple) which represents PostgreSQL cluster. Consists of the following fields: @@ -309,6 +332,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat :param members: list of Member object, all PostgreSQL cluster members including leader :param failover: reference to `Failover` object :param sync: reference to `SyncState` object, last observed synchronous replication state. + :param history: reference to `TimelineHistory` object """ def is_unlocked(self): @@ -342,6 +366,7 @@ class AbstractDCS(object): _CONFIG = 'config' _LEADER = 'leader' _FAILOVER = 'failover' + _HISTORY = 'history' _MEMBERS = 'members/' _OPTIME = 'optime' _LEADER_OPTIME = _OPTIME + '/' + _LEADER @@ -389,6 +414,10 @@ class AbstractDCS(object): def failover_path(self): return self.client_path(self._FAILOVER) + @property + def history_path(self): + return self.client_path(self._HISTORY) + @property def leader_optime_path(self): return self.client_path(self._LEADER_OPTIME) @@ -560,6 +589,10 @@ class AbstractDCS(object): sync_value = self.sync_state(leader, sync_standby) return self.set_sync_state_value(json.dumps(sync_value, separators=(',', ':')), index) + @abc.abstractmethod + def set_history_value(self, value): + """""" + @abc.abstractmethod def set_sync_state_value(self, value, index=None): """""" diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index d1120099..6c319744 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -8,7 +8,7 @@ import time import urllib3 from consul import ConsulException, NotFound, base -from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState +from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory from patroni.exceptions import DCSError from patroni.utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port from urllib3.exceptions import HTTPError @@ -264,6 +264,10 @@ class Consul(AbstractDCS): config = nodes.get(self._CONFIG) config = config and ClusterConfig.from_node(config['ModifyIndex'], config['Value']) + # get timeline history + history = nodes.get(self._HISTORY) + history = history and TimelineHistory.from_node(history['ModifyIndex'], history['Value']) + # get last leader operation last_leader_operation = nodes.get(self._LEADER_OPTIME) last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['Value']) @@ -293,9 +297,9 @@ class Consul(AbstractDCS): sync = nodes.get(self._SYNC) sync = SyncState.from_node(sync and sync['ModifyIndex'], sync and sync['Value']) - self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync) + self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) except NotFound: - self._cluster = Cluster(None, None, None, None, [], None, None) + self._cluster = Cluster(None, None, None, None, [], None, None, None) except Exception: logger.exception('get_cluster') raise ConsulError('Consul is not responding properly') @@ -372,6 +376,10 @@ class Consul(AbstractDCS): def delete_cluster(self): return self.retry(self._client.kv.delete, self.client_path(''), recurse=True) + @catch_consul_errors + def set_history_value(self, value): + return self._client.kv.put(self.history_path, value) + @catch_consul_errors def delete_leader(self): cluster = self.cluster diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 940654da..c945a798 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -12,7 +12,7 @@ import time from dns.exception import DNSException from dns import resolver -from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState +from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory from patroni.exceptions import DCSError from patroni.utils import Retry, RetryFailedError, split_host_port from urllib3.exceptions import HTTPError, ReadTimeoutError @@ -455,6 +455,10 @@ class Etcd(AbstractDCS): config = nodes.get(self._CONFIG) config = config and ClusterConfig.from_node(config.modifiedIndex, config.value) + # get timeline history + history = nodes.get(self._HISTORY) + history = history and TimelineHistory.from_node(history.modifiedIndex, history.value) + # get last leader operation last_leader_operation = nodes.get(self._LEADER_OPTIME) last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation.value) @@ -479,9 +483,9 @@ class Etcd(AbstractDCS): sync = nodes.get(self._SYNC) sync = SyncState.from_node(sync and sync.modifiedIndex, sync and sync.value) - self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync) + self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) except etcd.EtcdKeyNotFound: - self._cluster = Cluster(None, None, None, None, [], None, None) + self._cluster = Cluster(None, None, None, None, [], None, None, None) except Exception as e: self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly')) self._has_failed = False @@ -540,6 +544,10 @@ class Etcd(AbstractDCS): def delete_cluster(self): return self.retry(self._client.delete, self.client_path(''), recursive=True) + @catch_etcd_errors + def set_history_value(self, value): + return self._client.write(self.history_path, value) + @catch_etcd_errors def set_sync_state_value(self, value, index=None): return self.retry(self._client.write, self.sync_path, value, prevIndex=index or 0) diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 7d424848..335e284b 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -8,7 +8,7 @@ import sys import time from kubernetes import client as k8s_client, config as k8s_config, watch as k8s_watch -from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState +from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory from patroni.exceptions import DCSError from patroni.utils import deep_compare, tzutc, Retry, RetryFailedError from urllib3.exceptions import HTTPError @@ -149,6 +149,10 @@ class Kubernetes(AbstractDCS): config = ClusterConfig.from_node(metadata and metadata.resource_version, annotations.get(self._CONFIG) or '{}') + # get timeline history + history = TimelineHistory.from_node(metadata and metadata.resource_version, + annotations.get(self._HISTORY) or '[]') + leader = nodes.get(self.leader_path) metadata = leader and leader.metadata self._leader_resource_version = metadata.resource_version if metadata else None @@ -190,7 +194,7 @@ class Kubernetes(AbstractDCS): metadata = sync and sync.metadata sync = SyncState.from_node(metadata and metadata.resource_version, metadata and metadata.annotations) - self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync) + self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) except Exception: logger.exception('get_cluster') raise KubernetesError('Kubernetes API is not responding properly') @@ -358,6 +362,11 @@ class Kubernetes(AbstractDCS): def delete_cluster(self): self.retry(self._api.delete_collection_namespaced_kind, self._namespace, label_selector=self._label_selector) + @catch_kubernetes_errors + def set_history_value(self, value): + patch = bool(self.cluster and self.cluster.config and self.cluster.config.index) + return self.patch_or_create(self.config_path, {self._HISTORY: value}, None, patch, False) + def set_sync_state_value(self, value, index=None): """Unused""" diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 40ae3cb9..df476a4e 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -5,7 +5,7 @@ import time from kazoo.client import KazooClient, KazooState, KazooRetry from kazoo.exceptions import NoNodeError, NodeExistsError from kazoo.handlers.threading import SequentialThreadingHandler -from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState +from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory from patroni.exceptions import DCSError from patroni.utils import deep_compare @@ -161,6 +161,10 @@ class ZooKeeper(AbstractDCS): config = self.get_node(self.config_path, watch=self.cluster_watcher) if self._CONFIG in nodes else None config = config and ClusterConfig.from_node(config[1].version, config[0], config[1].mzxid) + # get timeline history + history = self.get_node(self.history_path, watch=self.cluster_watcher) if self._HISTORY in nodes else None + history = history and TimelineHistory.from_node(history[1].mzxid, history[0]) + # get last leader operation last_leader_operation = self._OPTIME in nodes and self._fetch_cluster and self.get_node(self.leader_optime_path) last_leader_operation = last_leader_operation and int(last_leader_operation[0]) or 0 @@ -193,7 +197,7 @@ class ZooKeeper(AbstractDCS): failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None failover = failover and Failover.from_node(failover[1].version, failover[0]) - self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync) + self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) def _load_cluster(self): if self._fetch_cluster or self._cluster is None: @@ -217,16 +221,19 @@ class ZooKeeper(AbstractDCS): logger.info('Could not take out TTL lock') return ret - def set_failover_value(self, value, index=None): + def __set_failover_or_sync_state_value(self, key, value, index=None): try: - self._client.retry(self._client.set, self.failover_path, value.encode('utf-8'), version=index or -1) + self._client.retry(self._client.set, key, value.encode('utf-8'), version=index or -1) return True except NoNodeError: - return value == '' or (index is None and self._create(self.failover_path, value)) + return value == '' or (index is None and self._create(key, value)) except Exception: logging.exception('set_failover_value') return False + def set_failover_value(self, value, index=None): + return self.__set_failover_or_sync_state_value(self.failover_path, value, index) + def set_config_value(self, value, index=None): try: self._client.retry(self._client.set, self.config_path, value.encode('utf-8'), version=index or -1) @@ -279,21 +286,24 @@ class ZooKeeper(AbstractDCS): def take_leader(self): return self.attempt_to_acquire_leader() - def _write_leader_optime(self, last_operation): - last_operation = last_operation.encode('utf-8') + def __write_leader_optime_or_history_value(self, key, value): + value = value.encode('utf-8') try: - self._client.set_async(self.leader_optime_path, last_operation).get(timeout=1) + self._client.set_async(key, value).get(timeout=1) return True except NoNodeError: try: - self._client.create_async(self.leader_optime_path, last_operation, makepath=True).get(timeout=1) + self._client.create_async(key, value, makepath=True).get(timeout=1) return True except Exception: - logger.exception('Failed to create %s', self.leader_optime_path) + logger.exception('Failed to create %s', key) except Exception: - logger.exception('Failed to update %s', self.leader_optime_path) + logger.exception('Failed to update %s', key) return False + def _write_leader_optime(self, last_operation): + return self.__write_leader_optime_or_history_value(self.leader_optime_path, last_operation) + def _update_leader(self): return True @@ -319,15 +329,11 @@ class ZooKeeper(AbstractDCS): except NoNodeError: return True + def set_history_value(self, value): + return self.__write_leader_optime_or_history_value(self.history_path, value) + def set_sync_state_value(self, value, index=None): - try: - self._client.retry(self._client.set, self.sync_path, value.encode('utf-8'), version=index or -1) - return True - except NoNodeError: - return value == '' or (index is None and self._create(self.sync_path, value)) - except Exception: - logging.exception('set_sync_state_value') - return False + return self.__set_failover_or_sync_state_value(self.sync_path, value, index) def delete_sync_state(self, index=None): return self.set_sync_state_value("{}", index) diff --git a/patroni/ha.py b/patroni/ha.py index 65845a89..6f37fdb9 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -57,6 +57,7 @@ class Ha(object): self.dcs = patroni.dcs self.cluster = None self.old_cluster = None + self._leader_timeline = None self.recovering = False self._post_bootstrap_task = None self._crash_recovery_executed = False @@ -82,6 +83,8 @@ class Ha(object): self.old_cluster = cluster self.cluster = cluster + self._leader_timeline = None if cluster.is_unlocked() else cluster.leader.timeline + def acquire_lock(self): return self.dcs.attempt_to_acquire_leader() @@ -123,9 +126,15 @@ class Ha(object): data['tags'] = tags if self.state_handler.pending_restart: data['pending_restart'] = True - if not self._async_executor.busy and data['state'] in ['running', 'restarting', 'starting']: + if self._async_executor.scheduled_action in (None, 'promote') \ + and data['state'] in ['running', 'restarting', 'starting']: try: - data['xlog_location'] = self.state_handler.wal_position() + timeline, wal_position = self.state_handler.timeline_wal_position() + data['xlog_location'] = wal_position + if not timeline: + timeline = self.state_handler.replica_cached_timeline(self._leader_timeline) + if timeline: + data['timeline'] = timeline except Exception: pass if self.patroni.scheduled_restart: @@ -366,6 +375,23 @@ class Ha(object): with self._member_state_lock: self._disable_sync -= 1 + def update_cluster_history(self): + master_timeline = self.state_handler.get_master_timeline() + cluster_history = self.cluster.history and self.cluster.history.lines + if cluster_history and master_timeline == 1: + self.dcs.set_history_value('[]') + elif not cluster_history or cluster_history[-1][0] != master_timeline - 1 or len(cluster_history[-1]) != 4: + cluster_history = {l[0]: l for l in cluster_history or []} + history = self.state_handler.get_history(master_timeline) + if history: + for line in history: + # enrich current history with promotion timestamps stored in DCS + if len(line) == 3 and line[0] in cluster_history \ + and len(cluster_history[line[0]]) == 4 \ + and cluster_history[line[0]][1] == line[1]: + line.append(cluster_history[line[0]][3]) + self.dcs.set_history_value(json.dumps(history, separators=(',', ':'))) + def enforce_master_role(self, message, promote_message): if not self.is_paused() and not self.watchdog.is_running and not self.watchdog.activate(): if self.state_handler.is_leader(): @@ -375,10 +401,14 @@ class Ha(object): self.release_leader_key_voluntarily() return 'Not promoting self because watchdog could not be activated' - if self.state_handler.is_leader() or self.state_handler.role == 'master': + if self.state_handler.is_leader(): # Inform the state handler about its master role. # It may be unaware of it if postgres is promoted manually. self.state_handler.set_role('master') + self.process_sync_replication() + self.update_cluster_history() + return message + elif self.state_handler.role == 'master': self.process_sync_replication() return message else: @@ -390,7 +420,9 @@ class Ha(object): # promotion until next cycle. TODO: trigger immediate retry of run_cycle return 'Postponing promotion because synchronous replication state was updated by somebody else' self.state_handler.set_synchronous_standby('*' if self.is_synchronous_mode_strict() else None) - self.state_handler.promote() + if self.state_handler.role != 'master': + self._async_executor.schedule('promote') + self._async_executor.run_async(self.state_handler.promote, args=(self.dcs.loop_wait,)) return promote_message @staticmethod @@ -426,7 +458,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.""" - my_wal_position = self.state_handler.wal_position() + _, my_wal_position = self.state_handler.timeline_wal_position() if check_replication_lag and self.is_lagging(my_wal_position): return False # Too far behind last reported wal position on master diff --git a/patroni/postgresql.py b/patroni/postgresql.py index d3feaa02..d0145b5f 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -36,11 +36,12 @@ STOP_POLLING_INTERVAL = 1 REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECK': 1, 'NEED': 2, 'NOT_NEED': 3, 'SUCCESS': 4, 'FAILED': 5}) sync_standby_name_re = re.compile('^[A-Za-z_][A-Za-z_0-9\$]*$') -wal_position_query = ("CASE WHEN pg_is_in_recovery() THEN GREATEST(" - " pg_{0}_{1}_diff(COALESCE(pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint," - " pg_{0}_{1}_diff(pg_last_{0}_replay_{1}(), '0/0')::bigint)" - " ELSE pg_{0}_{1}_diff(pg_current_{0}_{1}(), '0/0')::bigint " - "END") +cluster_info_query = ("SELECT CASE WHEN pg_is_in_recovery() THEN 0 " + "ELSE ('x' || SUBSTR(pg_{0}file_name(pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, " + "CASE WHEN pg_is_in_recovery() THEN GREATEST(" + " pg_{0}_{1}_diff(COALESCE(pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint," + " pg_{0}_{1}_diff(pg_last_{0}_replay_{1}(), '0/0')::bigint)" + "ELSE pg_{0}_{1}_diff(pg_current_{0}_{1}(), '0/0')::bigint END") def quote_ident(value): @@ -167,6 +168,7 @@ class Postgresql(object): self._state_entry_timestamp = None self._cluster_info_state = {} + self._cached_replica_timeline = None # Last known running process self._postmaster_proc = None @@ -710,11 +712,10 @@ class Postgresql(object): def _cluster_info_state_get(self, name): if not self._cluster_info_state: - stmt = "SELECT pg_is_in_recovery(), " + wal_position_query.format(self.wal_name, self.lsn_name) - + stmt = cluster_info_query.format(self.wal_name, self.lsn_name) try: result = self._is_leader_retry(self._query, stmt).fetchone() - self._cluster_info_state = dict(zip(['is_in_recovery', 'wal_position'], result)) + 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)} if not self.is_starting() and self.pg_isready() == STATE_REJECT: @@ -726,7 +727,7 @@ class Postgresql(object): return self._cluster_info_state.get(name) def is_leader(self): - return not self._cluster_info_state_get('is_in_recovery') + return bool(self._cluster_info_state_get('timeline')) def is_running(self): """Returns PostmasterProcess if one is running on the data directory or None. If most recently seen process @@ -1201,31 +1202,57 @@ class Postgresql(object): except Exception: return logger.exception('Exception when working with leader') - def _get_local_timeline_lsn(self): + def _get_local_timeline_lsn_from_replication_connection(self): timeline = lsn = None + try: + with self._get_replication_connection_cursor(**self._local_replication_address) as cur: + cur.execute('IDENTIFY_SYSTEM') + timeline, lsn = cur.fetchone()[1:3] + except Exception: + logger.exception('Can not fetch local timeline and lsn from replication connection') + return timeline, lsn + + def _get_local_timeline_lsn_from_controldata(self): + timeline = lsn = None + data = self.controldata() + try: + if data.get('Database cluster state') == 'shut down in recovery': + lsn = data.get('Minimum recovery ending location') + timeline = int(data.get("Min recovery ending loc's timeline")) + if lsn == '0/0' or timeline == 0: # it was a master when it crashed + data['Database cluster state'] = 'shut down' + if data.get('Database cluster state') == 'shut down': + lsn = data.get('Latest checkpoint location') + timeline = int(data.get("Latest checkpoint's TimeLineID")) + except (TypeError, ValueError): + logger.exception('Failed to get local timeline and lsn from pg_controldata output') + return timeline, lsn + + def _get_local_timeline_lsn(self): if self.is_running(): # if postgres is running - get timeline and lsn from replication connection - try: - with self._get_replication_connection_cursor(**self._local_replication_address) as cur: - cur.execute('IDENTIFY_SYSTEM') - timeline, lsn = cur.fetchone()[1:3] - except Exception: - logger.exception('Can not fetch local timeline and lsn from replication connection') + timeline, lsn = self._get_local_timeline_lsn_from_replication_connection() else: # otherwise analyze pg_controldata output - data = self.controldata() - try: - if data.get('Database cluster state') == 'shut down in recovery': - lsn = data.get('Minimum recovery ending location') - timeline = int(data.get("Min recovery ending loc's timeline")) - if lsn == '0/0' or timeline == 0: # it was a master when it crashed - data['Database cluster state'] = 'shut down' - if data.get('Database cluster state') == 'shut down': - lsn = data.get('Latest checkpoint location') - timeline = int(data.get("Latest checkpoint's TimeLineID")) - except (TypeError, ValueError): - logger.exception('Failed to get local timeline and lsn from pg_controldata output') + timeline, lsn = self._get_local_timeline_lsn_from_controldata() logger.info('Local timeline=%s lsn=%s', timeline, lsn) return timeline, lsn + @staticmethod + def parse_lsn(lsn): + t = lsn.split('/') + return int(t[0], 16) * 0x100000000 + int(t[1], 16) + + @staticmethod + def parse_history(data): + for line in data.split('\n'): + values = line.strip().split('\t') + if len(values) == 3: + try: + values[0] = int(values[0]) + values[1] = Postgresql.parse_lsn(values[1]) + yield values + except (IndexError, ValueError): + logger.exception('Exception when parsing timeline history line "%s"', values) + def _check_timeline_and_lsn(self, leader): local_timeline, local_lsn = self._get_local_timeline_lsn() if local_timeline is None or local_lsn is None: @@ -1252,28 +1279,44 @@ class Postgresql(object): return logger.exception('Exception when working with master via replication connection') if history is not None: - def parse_lsn(lsn): - t = lsn.split('/') - return int(t[0], 16) * 0x100000000 + int(t[1], 16) - - for line in history.split('\n'): - line = line.strip().split('\t') - if len(line) == 3: + for parent_timeline, switchpoint, _ in self.parse_history(history): + if parent_timeline == local_timeline: try: - timeline = int(line[0]) - if timeline == local_timeline: - try: - need_rewind = parse_lsn(local_lsn) >= parse_lsn(line[1]) - except ValueError: - logger.exception('Exception when parsing lsn') - break - elif timeline > local_timeline: - break - except ValueError: - continue + need_rewind = self.parse_lsn(local_lsn) >= switchpoint + except (IndexError, ValueError): + logger.exception('Exception when parsing lsn') + break + elif parent_timeline > local_timeline: + break self._rewind_state = need_rewind and REWIND_STATUS.NEED or REWIND_STATUS.NOT_NEED + def get_replica_timeline(self): + return self._get_local_timeline_lsn_from_replication_connection()[0] + + def replica_cached_timeline(self, master_timeline): + if not self._cached_replica_timeline or not master_timeline or self._cached_replica_timeline != master_timeline: + self._cached_replica_timeline = self.get_replica_timeline() + return self._cached_replica_timeline + + def get_master_timeline(self): + return self._cluster_info_state_get('timeline') + + def get_history(self, timeline): + history_path = 'pg_{0}/{1:08X}.history'.format(self.wal_name, timeline) + try: + cursor = self._cursor() + cursor.execute('SELECT isdir, modification FROM pg_stat_file(%s)', (history_path,)) + isdir, modification = cursor.fetchone() + if not isdir: + cursor.execute('SELECT pg_read_file(%s)', (history_path,)) + history = list(self.parse_history(cursor.fetchone()[0])) + if history[-1][0] == timeline - 1: + history[-1].append(modification.isoformat()) + return history + except Exception: + logger.exception('Failed to read and parse %s', (history_path,)) + def rewind(self, leader): if self.is_running() and not self.stop(checkpoint=False): return logger.warning('Can not run pg_rewind because postgres is still running') @@ -1370,15 +1413,22 @@ class Postgresql(object): except IOError: logger.exception('unable to restore configuration files from backup') - def promote(self): + def _wait_promote(self, wait_seconds): + for _ in polling_loop(wait_seconds - 1): + data = self.controldata() + if data.get('Database cluster state') == 'in production': + return True + + def promote(self, wait_seconds): if self.role == 'master': return True - ret = self.pg_ctl('promote') + ret = self.pg_ctl('promote', '-W') if ret: self.set_role('master') logger.info("cleared rewind state after becoming the leader") self._rewind_state = REWIND_STATUS.INITIAL self.call_nowait(ACTION_ON_ROLE_CHANGE) + ret = self._wait_promote(wait_seconds) return ret def create_or_update_role(self, name, password, options): @@ -1398,15 +1448,15 @@ BEGIN END; $$""".format(name, ' '.join(options)), name, password, password) - def wal_position(self): + def timeline_wal_position(self): # This method could be called from different threads (simultaneously with some other `_query` calls). # If it is called not from main thread we will create a new cursor to execute statement. if current_thread().ident == self.__thread_ident: - return self._cluster_info_state_get('wal_position') + return self._cluster_info_state_get('timeline'), self._cluster_info_state_get('wal_position') with self.connection().cursor() as cursor: - cursor.execute('SELECT ' + wal_position_query.format(self.wal_name, self.lsn_name)) - return cursor.fetchone()[0] + cursor.execute(cluster_info_query.format(self.wal_name, self.lsn_name)) + return cursor.fetchone()[:2] def load_replication_slots(self): if self.use_slots and self._schedule_load_slots: @@ -1469,7 +1519,7 @@ $$""".format(name, ' '.join(options)), name, password, password) self._schedule_load_slots = True def last_operation(self): - return str(self.wal_position()) + return str(self._cluster_info_state_get('wal_position')) def _post_restore(self): self.delete_trigger_file() diff --git a/tests/test_api.py b/tests/test_api.py index 9ed644ef..0945711c 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -37,6 +37,10 @@ class MockPostgresql(object): def postmaster_start_time(): return str(postmaster_start_time) + @staticmethod + def replica_cached_timeline(_): + return 2 + class MockWatchdog(object): is_healthy = False @@ -152,6 +156,8 @@ class TestRestApiHandler(unittest.TestCase): with patch.object(MockHa, 'restart_scheduled', Mock(return_value=True)): MockRestApiServer(RestApiHandler, 'GET /master') self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /master')) + with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, '')])): + self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni')) def test_do_OPTIONS(self): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'OPTIONS / HTTP/1.0')) diff --git a/tests/test_consul.py b/tests/test_consul.py index 47df9550..fad173d2 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -172,3 +172,7 @@ class TestConsul(unittest.TestCase): def test_sync_state(self): self.assertTrue(self.c.set_sync_state_value('{}')) self.assertTrue(self.c.delete_sync_state()) + + @patch.object(consul.Consul.KV, 'put', Mock(return_value=True)) + def test_set_history_value(self): + self.assertTrue(self.c.set_history_value('{}')) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index b3ed44c2..c4c11729 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -335,3 +335,6 @@ class TestEtcd(unittest.TestCase): def test_sync_state(self): self.assertFalse(self.etcd.write_sync_state('leader', None)) self.assertFalse(self.etcd.delete_sync_state()) + + def test_set_history_value(self): + self.assertFalse(self.etcd.set_history_value('{}')) diff --git a/tests/test_ha.py b/tests/test_ha.py index b037a41a..2312ddbe 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -5,7 +5,7 @@ import unittest from mock import Mock, MagicMock, PropertyMock, patch from patroni.config import Config -from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState +from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState, TimelineHistory from patroni.dcs.etcd import Client from patroni.exceptions import DCSError, PostgresConnectionException, PatroniException from patroni.ha import Ha, _MemberStatus @@ -25,7 +25,8 @@ def false(*args, **kwargs): def get_cluster(initialize, leader, members, failover, sync): - return Cluster(initialize, ClusterConfig(1, {1: 2}, 1), leader, 10, members, failover, sync) + history = TimelineHistory(1, [(1, 67197376, 'no recovery target specified', datetime.datetime.now().isoformat())]) + return Cluster(initialize, ClusterConfig(1, {1: 2}, 1), leader, 10, members, failover, sync, history) def get_cluster_not_initialized_without_leader(): @@ -115,7 +116,8 @@ def run_async(self, func, args=()): @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster())) @patch.object(Postgresql, 'is_leader', Mock(return_value=True)) -@patch.object(Postgresql, 'wal_position', Mock(return_value=10)) +@patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10))) +@patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=3)) @patch.object(Postgresql, 'call_nowait', Mock(return_value=True)) @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False)) @patch.object(Postgresql, 'controldata', Mock(return_value={'Database system identifier': '1234567890'})) @@ -130,6 +132,7 @@ def run_async(self, func, args=()): @patch.object(etcd.Client, 'write', etcd_write) @patch.object(etcd.Client, 'read', etcd_read) @patch.object(etcd.Client, 'delete', Mock(side_effect=etcd.EtcdException)) +@patch('patroni.postgresql.polling_loop', Mock(return_value=range(1))) @patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=False)) @patch('patroni.async_executor.AsyncExecutor.run_async', run_async) @patch('subprocess.call', Mock(return_value=0)) @@ -166,7 +169,8 @@ class TestHa(unittest.TestCase): self.assertTrue(self.ha.update_lock(True)) def test_touch_member(self): - self.p.wal_position = Mock(side_effect=Exception) + self.p.timeline_wal_position = Mock(return_value=(0, 1)) + self.p.replica_cached_timeline = Mock(side_effect=Exception) self.ha.touch_member() def test_start_as_replica(self): @@ -221,8 +225,10 @@ class TestHa(unittest.TestCase): self.p.is_leader = false self.p.is_healthy = true self.ha.has_lock = true + self.p.controldata = lambda: {'Database cluster state': 'in production'} self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock') + @patch('psycopg2.connect', psycopg2_connect) def test_acquire_lock_as_master(self): self.assertEquals(self.ha.run_cycle(), 'acquired session lock as a leader') @@ -231,6 +237,13 @@ class TestHa(unittest.TestCase): self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') + def test_long_promote(self): + self.ha.cluster.is_unlocked = false + self.ha.has_lock = true + self.p.is_leader = false + self.p.set_role('master') + self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') + def test_demote_after_failing_to_obtain_lock(self): self.ha.acquire_lock = false self.assertEquals(self.ha.run_cycle(), 'demoted self after trying and failing to obtain lock') @@ -526,7 +539,7 @@ class TestHa(unittest.TestCase): self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) self.ha.fetch_node_status = get_node_status(wal_position=11) # accessible, in_recovery, wal position ahead self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) - with patch('patroni.postgresql.Postgresql.wal_position', return_value=1): + with patch('patroni.postgresql.Postgresql.timeline_wal_position', return_value=(1, 1)): self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) self.ha.patroni.nofailover = True self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) @@ -877,3 +890,9 @@ 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'") + + def test_update_cluster_history(self): + self.p.get_master_timeline = Mock(return_value=1) + self.ha.has_lock = true + self.ha.cluster.is_unlocked = false + self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index c9c012e9..41b8e82b 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -101,3 +101,6 @@ class TestKubernetes(unittest.TestCase): self.assertFalse(self.k.watch('1', 2)) self.assertRaises(KeyboardInterrupt, self.k.watch, '1', 2) self.assertTrue(self.k.watch('1', 2)) + + def test_set_history_value(self): + self.k.set_history_value('{}') diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 4c1e48fd..e255818e 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -1,3 +1,4 @@ +import datetime import mock # for the mock.call method, importing it without a namespace breaks python3 import os import psycopg2 @@ -34,13 +35,13 @@ class MockCursor(object): elif sql.startswith('SELECT slot_name'): self.results = [('blabla',), ('foobar',)] elif sql.startswith('SELECT CASE WHEN pg_is_in_recovery()'): - self.results = [(2,)] + self.results = [(1, 2)] elif sql.startswith('SELECT pg_is_in_recovery()'): self.results = [(False, 2)] elif sql.startswith('WITH replication_info AS ('): replication_info = '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' +\ '"state":"streaming","sync_state":"async","sync_priority":0}]' - self.results = [('', True, '', '', '', '', False, replication_info)] + self.results = [('', 0, '', '', '', '', False, replication_info)] elif sql.startswith('SELECT name, setting'): self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'), ('search_path', 'public', None, 'string', 'user'), @@ -50,6 +51,11 @@ class MockCursor(object): ('unix_socket_directories', '/tmp', None, 'string', 'postmaster')] elif sql.startswith('IDENTIFY_SYSTEM'): self.results = [('1', 2, '0/402EEC0', '')] + elif sql.startswith('SELECT isdir, modification'): + self.results = [(False, datetime.datetime.now())] + elif sql.startswith('SELECT pg_read_file'): + self.results = [('1\t0/40159C0\tno recovery target specified\n\n' + + '2\t1/40159C0\tno recovery target specified\n',)] elif sql.startswith('TIMELINE_HISTORY '): self.results = [('', b'x\t0/40159C0\tno recovery target specified\n\n' + b'1\t0/40159C0\tno recovery target specified\n\n' + @@ -347,7 +353,7 @@ class TestPostgresql(unittest.TestCase): @patch.object(Postgresql, 'start', Mock()) @patch.object(Postgresql, 'can_rewind', PropertyMock(return_value=True)) - @patch.object(Postgresql, '_get_local_timeline_lsn', Mock(return_value=(2, '0/40159C1'))) + @patch.object(Postgresql, '_get_local_timeline_lsn', Mock(return_value=(2, '40159C1'))) @patch.object(Postgresql, 'check_leader_is_not_in_recovery') def test__check_timeline_and_lsn(self, mock_check_leader_is_not_in_recovery): mock_check_leader_is_not_in_recovery.return_value = False @@ -358,12 +364,8 @@ class TestPostgresql(unittest.TestCase): self.p.trigger_check_diverged_lsn() with patch('psycopg2.connect', Mock(side_effect=Exception)): self.assertFalse(self.p.rewind_needed_and_possible(self.leader)) - with patch.object(MockCursor, 'fetchone', - Mock(side_effect=[('', 2, '0/0'), ('', b'2\tG/40159C0\tno recovery target specified\n\n')])): - self.assertFalse(self.p.rewind_needed_and_possible(self.leader)) self.p.trigger_check_diverged_lsn() - with patch.object(MockCursor, 'fetchone', - Mock(side_effect=[('', 2, '0/0'), ('', b'3\t040159C0\tno recovery target specified\n')])): + with patch.object(MockCursor, 'fetchone', Mock(side_effect=[('', 2, '0/0'), ('', b'3\t0/40159C0\tn\n')])): self.assertFalse(self.p.rewind_needed_and_possible(self.leader)) self.p.trigger_check_diverged_lsn() with patch.object(MockCursor, 'fetchone', Mock(return_value=('', 1, '0/0'))): @@ -448,7 +450,7 @@ class TestPostgresql(unittest.TestCase): @patch.object(Postgresql, 'is_running', Mock(return_value=True)) def test_sync_replication_slots(self): self.p.start() - cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None, None) + cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None, None, None) with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg2.OperationalError)): self.p.sync_replication_slots(cluster) self.p.sync_replication_slots(cluster) @@ -494,12 +496,12 @@ class TestPostgresql(unittest.TestCase): def test_promote(self): self.p.set_role('replica') - self.assertTrue(self.p.promote()) - self.assertTrue(self.p.promote()) + self.assertIsNone(self.p.promote(0)) + self.assertTrue(self.p.promote(0)) - def test_last_operation(self): - self.assertEquals(self.p.last_operation(), '2') - Thread(target=self.p.last_operation).start() + def test_timeline_wal_position(self): + self.assertEquals(self.p.timeline_wal_position(), (1, 2)) + Thread(target=self.p.timeline_wal_position).start() @patch.object(PostmasterProcess, 'from_pidfile') def test_is_running(self, mock_frompidfile): @@ -817,7 +819,7 @@ class TestPostgresql(unittest.TestCase): def test_pick_sync_standby(self): cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None, - SyncState(0, self.me.name, self.leadermem.name)) + SyncState(0, self.me.name, self.leadermem.name), None) with patch.object(Postgresql, "query", return_value=[ (self.leadermem.name, 'streaming', 'sync'), @@ -935,6 +937,12 @@ class TestPostgresql(unittest.TestCase): def test_fix_cluster_state(self): self.assertTrue(self.p.fix_cluster_state()) + def test_replica_cached_timeline(self): + self.assertEquals(self.p.replica_cached_timeline(1), 2) + + def test_get_master_timeline(self): + self.assertEquals(self.p.get_master_timeline(), 1) + def test_cancellable_subprocess_call(self): self.p.cancel() self.assertRaises(PostgresException, self.p.cancellable_subprocess_call) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index c70949cd..452fd7df 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -217,3 +217,6 @@ class TestZooKeeper(unittest.TestCase): self.zk.set_sync_state_value('ok') self.zk.set_sync_state_value('Exception') self.zk.delete_sync_state() + + def test_set_history_value(self): + self.zk.set_history_value('{}')