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('{}')