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

@@ -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]
 

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):
         """"""

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

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)

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"""
 

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)

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
 

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()

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'))

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

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

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')

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

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)

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