Merge pull request #122 from zalando/feature/replica_without_the_master

Run replicas without the master

patroni/__init__.py

@@ -31,6 +31,10 @@ class Patroni:
     def nofailover(self):
         return self.tags.get('nofailover', False)
 
+    @property
+    def replicatefrom(self):
+        return self.tags.get('replicatefrom')
+
     @staticmethod
     def get_dcs(name, config):
         if 'etcd' in config:

patroni/dcs.py

@@ -67,6 +67,10 @@ class Member(namedtuple('Member', 'index,name,session,data')):
     def nofailover(self):
         return self.data.get('tags', {}).get('nofailover', False)
 
+    @property
+    def replicatefrom(self):
+        return self.data.get('tags', {}).get('replicatefrom')
+
 
 class Leader(namedtuple('Leader', 'index,session,member')):
 
@@ -107,6 +111,9 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem
     def is_unlocked(self):
         return not (self.leader and self.leader.name)
 
+    def has_member(self, member_name):
+        return any(m for m in self.members if m.name == member_name)
+
 
 class AbstractDCS:
 

patroni/ha.py

@@ -19,12 +19,13 @@ class Ha:
         self.dcs = patroni.dcs
         self.cluster = None
         self.old_cluster = None
+        self.recovering = False
         self._async_executor = AsyncExecutor()
 
     def load_cluster_from_dcs(self):
         cluster = self.dcs.get_cluster()
 
-        # We want to keep the state of cluster when it was healhy
+        # We want to keep the state of cluster when it was healthy
         if not cluster.is_unlocked() or not self.old_cluster:
             self.old_cluster = cluster
         self.cluster = cluster
@@ -61,18 +62,18 @@ class Ha:
                 pass
         self.dcs.touch_member(json.dumps(data, separators=(',', ':')))
 
-    def copy_backup_from_leader(self, leader):
-        if self.state_handler.bootstrap(leader):
-            logger.info('bootstrapped from leader')
+    def clone(self, leader):
+        if self.state_handler.bootstrap(cluster_initialized=True, current_leader=leader):
+            logger.info('bootstrapped from leader' if leader else 'bootstrapped without leader')
         else:
             self.state_handler.stop('immediate')
             self.state_handler.remove_data_directory()
-            logger.error('failed to bootstrap from leader')
+            logger.error('failed to bootstrap from leader' if leader else 'failed to bootstrap (without leader)')
 
     def bootstrap(self):
         if not self.cluster.is_unlocked():  # cluster already has leader
             self._async_executor.schedule('bootstrap from leader')
-            self._async_executor.run_async(self.copy_backup_from_leader, args=(self.cluster.leader, ))
+            self._async_executor.run_async(self.clone, args=(self.cluster.leader, ))
             return 'trying to bootstrap from leader'
         elif not self.cluster.initialize and not self.patroni.nofailover:  # no initialize key
             if self.dcs.initialize(create_new=True):  # race for initialization
@@ -91,41 +92,38 @@ class Ha:
             else:
                 return 'failed to acquire initialize lock'
         else:
+            if self.state_handler.can_create_replica_without_leader():
+                self._async_executor.run_async(self.clone, args=(None, ))
+                return "trying to bootstrap without leader"
             return 'waiting for leader to bootstrap'
 
     def recover(self):
-        has_lock = self.has_lock()
-
         # try to see if we are the former master that crashed. If so - we likely need to run pg_rewind
         # in order to join the former standby being promoted.
         pg_controldata = self.state_handler.controldata()
-        if not has_lock and pg_controldata and\
+        if (self.state_handler.role == 'master') and pg_controldata and\
                 pg_controldata.get('Database cluster state', '') == 'in production':  # crashed master
             self.state_handler.require_rewind()
+        self.recovering = True
+        return self.follow("started as readonly because i had the session lock",
+                           "started as a secondary",
+                           refresh=True, recovery=True)
 
-        # XXX: follow the leader calls stop, which might take quite some time.
-        # perhaps we should run sync asynchronously
-        # (we still need the exit code from follow_the_leader)
-        ret = self.state_handler.follow_the_leader(None if has_lock else self.cluster.leader, recovery=True)
-        if not ret:
-            if not has_lock:
-                return 'failed to start postgres'
-            self.dcs.delete_leader()
-            self.dcs.reset_cluster()
-            return 'removed leader key after trying and failing to start postgres'
-        if not has_lock:
-            return 'started as a secondary'
-        logger.info('started as readonly because i had the session lock')
-        self.load_cluster_from_dcs()
-
-    def follow_the_leader(self, demote_reason, follow_reason, refresh=True):
+    def follow(self, demote_reason, follow_reason, refresh=True, recovery=False):
         refresh and self.load_cluster_from_dcs()
-        ret = demote_reason if self.state_handler.is_leader() else follow_reason
-        leader = self.cluster.leader
-        leader = None if (leader and leader.name) == self.state_handler.name else leader
-        if not self.state_handler.check_recovery_conf(leader):
+        ret = demote_reason if (not recovery and self.state_handler.is_leader() or
+                                recovery and self.state_handler.role == 'master') else follow_reason
+        # determine the node to follow. If replicatefrom tag is set,
+        # try to follow the node mentioned there, otherwise, follow the leader.
+        if self.patroni.replicatefrom:
+            node_to_follow = [m for m in self.cluster.members if m.name == self.patroni.replicatefrom]
+            node_to_follow = node_to_follow[0] if node_to_follow else self.cluster.leader
+        else:
+            node_to_follow = self.cluster.leader
+        node_to_follow = None if node_to_follow and node_to_follow.name == self.state_handler.name else node_to_follow
+        if not self.state_handler.check_recovery_conf(node_to_follow) or recovery:
             self._async_executor.schedule('changing primary_conninfo and restarting')
-            self._async_executor.run_async(self.state_handler.follow_the_leader, (leader, ))
+            self._async_executor.run_async(self.state_handler.follow, (node_to_follow, recovery))
         return ret
 
     def enforce_master_role(self, message, promote_message):
@@ -298,14 +296,14 @@ class Ha:
                 return self.enforce_master_role('acquired session lock as a leader',
                                                 'promoted self to leader by acquiring session lock')
             else:
-                return self.follow_the_leader('demoted self due after trying and failing to obtain lock',
-                                              'following new leader after trying and failing to obtain lock')
+                return self.follow('demoted self after trying and failing to obtain lock',
+                                   'following new leader after trying and failing to obtain lock')
         else:
             if self.patroni.nofailover:
-                return self.follow_the_leader('demoting self because I am not allowed to become master',
-                                              'following a different leader because I am not allowed to promote')
-            return self.follow_the_leader('demoting self because i am not the healthiest node',
-                                          'following a different leader because i am not the healthiest node')
+                return self.follow('demoting self because I am not allowed to become master',
+                                   'following a different leader because I am not allowed to promote')
+            return self.follow('demoting self because i am not the healthiest node',
+                               'following a different leader because i am not the healthiest node')
 
     def process_healthy_cluster(self):
         if self.has_lock():
@@ -323,8 +321,8 @@ class Ha:
                 self.load_cluster_from_dcs()
         else:
             logger.info('does not have lock')
-        return self.follow_the_leader('demoting self because i do not have the lock and i was a leader',
-                                      'no action.  i am a secondary and i am following a leader', False)
+        return self.follow('demoting self because i do not have the lock and i was a leader',
+                           'no action.  i am a secondary and i am following a leader', False)
 
     def schedule(self, action):
         with self._async_executor:
@@ -352,7 +350,7 @@ class Ha:
     def reinitialize(self, cluster):
         self.state_handler.stop('immediate')
         self.state_handler.remove_data_directory()
-        self.copy_backup_from_leader(cluster.leader)
+        self.clone(cluster.leader)
 
     def process_scheduled_action(self):
         if self.reinitialize_scheduled():
@@ -382,6 +380,15 @@ class Ha:
         # so even 1 << 32 would have 10 digits.
         return str(sysid) and len(str(sysid)) >= 10 and str(sysid).isdigit()
 
+    def post_recover(self):
+        if not self.state_handler.is_running():
+            if self.has_lock():
+                self.dcs.delete_leader()
+                self.dcs.reset_cluster()
+                return 'removed leader key after trying and failing to start postgres'
+            return 'failed to start postgres'
+        return None
+
     def _run_cycle(self):
         try:
             self.load_cluster_from_dcs()
@@ -395,6 +402,13 @@ class Ha:
             if self._async_executor.busy:
                 return self.handle_long_action_in_progress()
 
+            # we've got here, so any async action has finished. Check if we tried to recover and failed
+            if self.recovering:
+                self.recovering = False
+                msg = self.post_recover()
+                if msg is not None:
+                    return msg
+
             # currently it can trigger only reinitialize
             msg = self.process_scheduled_action()
             if msg is not None:
@@ -425,14 +439,18 @@ class Ha:
                 else:
                     return self.process_healthy_cluster()
             finally:
-                self.state_handler.sync_replication_slots(self.cluster)
+                # we might not have a valid PostgreSQL connection here if another thread
+                # stops PostgreSQL, therefore, we only reload replication slots if no
+                # asynchronous processes are running (should be always the case for the master)
+                if not self._async_executor.busy:
+                    self.state_handler.sync_replication_slots(self.cluster)
         except DCSError:
             logger.error('Error communicating with DCS')
             if self.state_handler.is_running() and self.state_handler.is_leader():
                 self.demote(delete_leader=False)
                 return 'demoted self because DCS is not accessible and i was a leader'
         except (psycopg2.Error, PostgresConnectionException):
-            logger.exception('Error communicating with Postgresql. Will try again later')
+            logger.exception('Error communicating with PostgreSQL. Will try again later')
 
     def run_cycle(self):
         with self._async_executor:

patroni/postgresql.py

@@ -221,10 +221,10 @@ class Postgresql:
         env['PGPASSFILE'] = self.pgpass
         return env
 
-    def sync_from_leader(self, leader):
-        r = parseurl(leader.conn_url)
-
-        env = self.write_pgpass(r)
+    def sync_replica(self, leader):
+        if leader:
+            r = parseurl(leader.conn_url)
+        env = self.write_pgpass(r) if leader else os.environ.copy()
         ret = self.create_replica(leader, env) == 0
         ret and self.delete_trigger_file()
         return ret
@@ -237,14 +237,30 @@ class Postgresql:
         """
         return ' '.join('{}={}'.format(param, val) for param, val in sorted(conn.items()))
 
+    def replica_method_can_work_without_leader(self, method):
+        return method != 'basebackup' and self.config and self.config.get(method, {}).get('no_master')
+
+    def can_create_replica_without_leader(self):
+        """ go through the replication methods to see if there are ones
+            that does not require a running leader to create the replica.
+        """
+        replica_methods = self.config.get('create_replica_method', [])
+        for replica_method in replica_methods:
+            if self.replica_method_can_work_without_leader(replica_method):
+                return True
+        return False
+
     def create_replica(self, leader, env):
         # create the replica according to the replica_method
         # defined by the user.  this is a list, so we need to
         # loop through all methods the user supplies
-        connstring = leader.conn_url
+        connstring = leader.conn_url if leader else ""
         # get list of replica methods from config.
         # If there is no configuration key, or no value is specified, use basebackup
         replica_methods = self.config.get('create_replica_method') or ['basebackup']
+        # if we don't have any leader, leave only replica methods that work without it
+        replica_methods = [r for r in replica_methods if self.replica_method_can_work_without_leader(r)] if not leader \
+            else replica_methods
         # go through them in priority order
         ret = 1
         for replica_method in replica_methods:
@@ -437,7 +453,7 @@ class Postgresql:
                     return pattern and (pattern in line)
         return not pattern
 
-    def write_recovery_conf(self, leader):
+    def write_recovery_conf(self, leader, bootstrap=False):
         with open(self.recovery_conf, 'w') as f:
             f.write("""standby_mode = 'on'
 recovery_target_timeline = 'latest'
@@ -446,6 +462,7 @@ recovery_target_timeline = 'latest'
                 f.write("""primary_conninfo = '{}'\n""".format(self.primary_conninfo(leader.conn_url)))
                 if self.use_slots:
                     f.write("""primary_slot_name = '{}'\n""".format(self.name))
+            if (leader and leader.conn_url) or bootstrap:
                 for name, value in self.config.get('recovery_conf', {}).items():
                     f.write("{} = '{}'\n".format(name, value))
 
@@ -526,7 +543,7 @@ recovery_target_timeline = 'latest'
                 except:
                     logger.exception("Unable to remove {}".format(path))
 
-    def follow_the_leader(self, leader, recovery=False):
+    def follow(self, leader, recovery=False):
         if not self.check_recovery_conf(leader) or recovery:
             change_role = (self.role == 'master')
 
@@ -598,9 +615,6 @@ recovery_target_timeline = 'latest'
             self.call_nowait(ACTION_ON_ROLE_CHANGE)
         return ret
 
-    def demote(self):
-        self.follow_the_leader(None)
-
     def create_or_update_role(self, name, password, options):
         self.query("""DO $$
 BEGIN
@@ -639,7 +653,18 @@ $$""".format(name, options), name, password, password)
         if self.use_slots:
             try:
                 self.load_replication_slots()
-                slots = [m.name for m in cluster.members if m.name != self.name] if self.role == 'master' else []
+                # if the replicatefrom tag is set on the member - we should not create the replication slot for it on
+                # the current master, because that member would replicate from elsewhere. We still create the slot if
+                # the replicatefrom destination member is currently not a member of the cluster (fallback to the
+                # master), or if replicatefrom destination member happens to be the current master
+                if self.role == 'master':
+                    slots = [m.name for m in cluster.members if m.name != self.name and
+                             (m.replicatefrom is None or m.replicatefrom == self.name or
+                              not cluster.has_member(m.replicatefrom))]
+                else:
+                    # only manage slots for replicas that replicate from this one, except for the leader among them
+                    slots = [m.name for m in cluster.members if m.replicatefrom == self.name and
+                             m.name != cluster.leader.name]
                 # drop unused slots
                 for slot in set(self.replication_slots) - set(slots):
                     self.query("""SELECT pg_drop_replication_slot(%s)
@@ -659,18 +684,28 @@ $$""".format(name, options), name, password, password)
     def last_operation(self):
         return str(self.xlog_position())
 
-    def bootstrap(self, current_leader=None):
+    def bootstrap(self, cluster_initialized=False, current_leader=None):
         """
-            Initially bootstrap PostgreSQL, either by creating a data
-            directory with initdb, or by initalizing a replica from an
-            exiting leader. Failure in the first case always leads to
-            exception, since there is no point in continuing if initdb failed.
-            In the second case, however, a False is returned on failure, since
-            it is normal for the replica to retry a failed attempt to initialize
-            from the master.
+            Populate PostgreSQL data directory by doing one of the following:
+             - create with initdb if there is no master.
+             - initialize the replica from an existing master
+             - initialize the replica using the replica creation method that
+               works without the master (i.e. restore from on-disk base backup)
+
+            The choice between the last 2 is triggered by the initialize flag.
+            We should never try to initdb an already initialized cluster, nor
+            try to bootstrap the cluster that lacks the initialize key from from
+            the master-less replica creation method (in the latter case, there is
+            no clear inidicator of the moment we should abandon our attempts and
+            swich to initdb).
+
+            Failure during initdb always leads to an exception, since there is
+            no point in continuing if initdb fails. For the rest of the cases,
+            the function returns False in order to inidicate a failed attempt
+            that should be retried in the future.
         """
         ret = False
-        if not current_leader:
+        if not (cluster_initialized or current_leader):
             ret = self.initialize() and self.start()
             if ret:
                 self.create_replication_user()
@@ -678,9 +713,9 @@ $$""".format(name, options), name, password, password)
             else:
                 raise PostgresException("Could not bootstrap master PostgreSQL")
         else:
-            if self.sync_from_leader(current_leader):
+            if self.sync_replica(current_leader):
                 self.restore_configuration_files()
-                self.write_recovery_conf(current_leader)
+                self.write_recovery_conf(current_leader, True)
                 ret = self.start()
         return ret
 

patroni/scripts/wale_restore.py

@@ -42,7 +42,7 @@ logger = logging.getLogger(__name__)
 
 class WALERestore(object):
 
-    def __init__(self, scope, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam):
+    def __init__(self, scope, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam, no_master):
         self.scope = scope
         self.master_connection = connstring
         self.data_dir = datadir
@@ -51,6 +51,7 @@ class WALERestore(object):
         self.wal_e.threshold_mb = threshold_mb
         self.wal_e.threshold_pct = threshold_pct
         self.wal_e.iam_string = ' --aws-instance-profile ' if use_iam == 1 else ''
+        self.no_master = no_master
         self.wal_e.cmd = 'envdir {0} wal-e {1} '.format(self.wal_e.dir, self.wal_e.iam_string)
         self.init_error = (not os.path.exists(self.wal_e.dir))
 
@@ -109,19 +110,23 @@ class WALERestore(object):
         conn = None
         cursor = None
         diff_in_bytes = long(backup_size)
-        try:
-            # get the difference in bytes between the current WAL location and the backup start offset
-            conn = psycopg2.connect(self.master_connection)
-            conn.autocommit = True
-            cursor = conn.cursor()
-            cursor.execute("SELECT pg_xlog_location_diff(pg_current_xlog_location(), %s)", (backup_start_lsn,))
-            diff_in_bytes = long(cursor.fetchone()[0])
-        except psycopg2.Error as e:
-            logger.error('could not determine difference with the master location: {}'.format(e))
-            return False
-        finally:
-            cursor and cursor.close()
-            conn and conn.close()
+        if not self.no_master:
+            try:
+                # get the difference in bytes between the current WAL location and the backup start offset
+                conn = psycopg2.connect(self.master_connection)
+                conn.autocommit = True
+                cursor = conn.cursor()
+                cursor.execute("SELECT pg_xlog_location_diff(pg_current_xlog_location(), %s)", (backup_start_lsn,))
+                diff_in_bytes = long(cursor.fetchone()[0])
+            except psycopg2.Error as e:
+                logger.error('could not determine difference with the master location: {}'.format(e))
+                return False
+            finally:
+                cursor and cursor.close()
+                conn and conn.close()
+        else:
+            # always try to use WAL-E if base backup is available
+            diff_in_bytes = 0
 
         # if the size of the accumulated WAL segments is more than a certan percentage of the backup size
         # or exceeds the pre-determined size - pg_basebackup is chosen instead.
@@ -150,13 +155,15 @@ def main():
     parser.add_argument('--threshold_megabytes', type=int, default=10240)
     parser.add_argument('--threshold_backup_size_percentage', type=int, default=30)
     parser.add_argument('--use_iam', type=int, default=0)
+    parser.add_argument('--no_master', type=int, default=0)
     args = parser.parse_args()
 
     # retry cloning in a loop
     for retry in range(0, args.retries + 1):
         restore = WALERestore(scope=args.scope, datadir=args.datadir, connstring=args.connstring,
                               env_dir=args.envdir, threshold_mb=args.threshold_megabytes,
-                              threshold_pct=args.threshold_backup_size_percentage, use_iam=args.use_iam)
+                              threshold_pct=args.threshold_backup_size_percentage, use_iam=args.use_iam,
+                              no_master=args.no_master)
         ret = restore.run()
         if ret == 0:
             break

postgres0.yml

@@ -87,14 +87,13 @@ postgresql:
     archive_mode: "on"
     wal_level: hot_standby
     archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f
-    max_wal_senders: 5
+    max_wal_senders: 10
     wal_keep_segments: 8
     archive_timeout: 1800s
-    max_replication_slots: 5
+    max_replication_slots: 10
     hot_standby: "on"
     wal_log_hints: "on"
 tags:
     nofailover: False
     noloadbalance: False
     clonefrom: False
-    replicatefrom: 127.0.0.1

postgres1.yml

@@ -88,14 +88,13 @@ postgresql:
     archive_mode: "on"
     wal_level: hot_standby
     archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f
-    max_wal_senders: 5
+    max_wal_senders: 10
     wal_keep_segments: 8
     archive_timeout: 1800s
-    max_replication_slots: 5
+    max_replication_slots: 10
     hot_standby: "on"
     wal_log_hints: "on"
 tags:
     nofailover: False
     noloadbalance: False
     clonefrom: False
-    replicatefrom: 127.0.0.1

postgres2.yml

@@ -0,0 +1,101 @@
+ttl: &ttl 30
+loop_wait: &loop_wait 10
+scope: &scope batman
+restapi:
+  listen: 127.0.0.1:8010
+  connect_address: 127.0.0.1:8010
+  auth: 'username:password'
+#  certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
+#  keyfile: /etc/ssl/private/ssl-cert-snakeoil.key
+etcd:
+  scope: *scope
+  ttl: *ttl
+  host: 127.0.0.1:4001
+  #discovery_srv: my-etcd.domain
+#zookeeper:
+#  scope: *scope
+#  session_timeout: *ttl
+#  reconnect_timeout: *loop_wait
+#  hosts:
+#    - 127.0.0.1:2181
+#    - 127.0.0.2:2181
+#  exhibitor:
+#    poll_interval: 300
+#    port: 8181
+#    hosts:
+#      - host1
+#      - host2
+#      - host3
+postgresql:
+  name: postgresql2
+  scope: *scope
+  listen: 127.0.0.1:5434
+  connect_address: 127.0.0.1:5434
+  data_dir: data/postgresql2
+  maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
+  use_slots: True
+  pgpass: /tmp/pgpass2
+  initdb:  ## We allow the following options to be passed on to initdb
+  # - auth: authmethod
+  # - auth-host: authmethod
+  # - auth-local: authmethod
+  - encoding: UTF8
+  # - data-checksums  # When pg_rewind is needed on 9.3, this needs to be enabled
+  # - locale: locale
+  # - lc-collate: locale
+  # - lc-ctype: locale
+  # - lc-messages: locale
+  # - lc-monetary: locale
+  # - lc-numeric: locale
+  # - lc-time: locale
+  # - text-search-config: CFG
+  # - xlogdir: directory
+  # - debug
+  # - noclean
+  pg_rewind:
+    username: postgres
+    password: zalando
+  pg_hba:
+  - host all all 0.0.0.0/0 md5
+  - hostssl all all 0.0.0.0/0 md5
+  replication:
+    username: replicator
+    password: rep-pass
+    network: 127.0.0.1/32
+  superuser:
+    user: postgres
+    password: zalando
+  admin:
+    username: admin
+    password: admin
+# commented-out example for wal-e provisioning
+  create_replica_method:
+    - basebackup
+#   - wal_e
+# commented-out example for wal-e provisioning
+  #wal_e:
+    #command: /patroni/scripts/wale_restore.py
+    #env_dir: /home/postgres/etc/wal-e.d/env
+    #threshold_megabytes: 10240
+    #threshold_backup_size_percentage: 30
+    #retries: 2
+    #use_iam: 1
+  #recovery_conf:
+    #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -p 1
+  recovery_conf:
+    restore_command: cp ../wal_archive/%f %p
+  parameters:
+    archive_mode: "on"
+    wal_level: hot_standby
+    archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f
+    max_wal_senders: 10
+    wal_keep_segments: 8
+    archive_timeout: 1800s
+    max_replication_slots: 10
+    hot_standby: "on"
+    wal_log_hints: "on"
+tags:
+    nofailover: False
+    noloadbalance: False
+    clonefrom: False
+    replicatefrom: postgresql1

tests/test_ha.py

@@ -90,6 +90,7 @@ class MockPatroni:
         self.api = Mock()
         self.tags = {}
         self.nofailover = None
+        self.replicatefrom = None
         self.api.connection_string = 'http://127.0.0.1:8008'
 
 
@@ -104,6 +105,7 @@ class TestHa(unittest.TestCase):
     def setUp(self, mock_machines):
         mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
         self.p = MockPostgresql()
+        self.p.can_create_replica_without_leader = MagicMock(return_value=False)
         self.e = Etcd('foo', {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'})
         self.e.client.read = etcd_read
         self.e.client.write = etcd_write
@@ -129,13 +131,19 @@ class TestHa(unittest.TestCase):
     def test_recover_replica_failed(self):
         self.p.controldata = lambda: {'Database cluster state': 'in production'}
         self.p.is_healthy = false
-        self.p.follow_the_leader = false
+        self.p.is_running = false
+        self.p.follow = false
+        self.assertEquals(self.ha.run_cycle(), 'started as a secondary')
         self.assertEquals(self.ha.run_cycle(), 'failed to start postgres')
 
     def test_recover_master_failed(self):
-        self.p.follow_the_leader = false
+        self.p.follow = false
         self.p.is_healthy = false
+        self.p.is_running = false
         self.ha.has_lock = true
+        self.p.role = 'master'
+        self.p.controldata = lambda: {'Database cluster state': 'in production'}
+        self.assertEquals(self.ha.run_cycle(), 'started as readonly because i had the session lock')
         self.assertEquals(self.ha.run_cycle(), 'removed leader key after trying and failing to start postgres')
 
     @patch('sys.exit', return_value=1)
@@ -146,7 +154,8 @@ class TestHa(unittest.TestCase):
 
     @patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
     def test_start_as_readonly(self):
-        self.p.is_leader = self.p.is_healthy = false
+        self.p.is_leader = false
+        self.p.is_healthy = true
         self.ha.has_lock = true
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock')
 
@@ -160,7 +169,7 @@ class TestHa(unittest.TestCase):
 
     def test_demote_after_failing_to_obtain_lock(self):
         self.ha.acquire_lock = false
-        self.assertEquals(self.ha.run_cycle(), 'demoted self due after trying and failing to obtain lock')
+        self.assertEquals(self.ha.run_cycle(), 'demoted self after trying and failing to obtain lock')
 
     def test_follow_new_leader_after_failing_to_obtain_lock(self):
         self.ha.is_healthiest_node = true
@@ -198,10 +207,12 @@ class TestHa(unittest.TestCase):
         self.ha.update_lock = false
         self.assertEquals(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader')
 
-    def test_follow_the_leader(self):
+    def test_follow(self):
         self.ha.cluster.is_unlocked = false
         self.p.is_leader = false
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
+        self.ha.patroni.replicatefrom = "foo"
+        self.assertEquals(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
 
     def test_no_etcd_connection_master_demote(self):
         self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
@@ -216,6 +227,11 @@ class TestHa(unittest.TestCase):
         self.ha.cluster = get_cluster_initialized_without_leader()
         self.assertEquals(self.ha.bootstrap(), 'waiting for leader to bootstrap')
 
+    def test_bootstrap_without_leader(self):
+        self.ha.cluster = get_cluster_initialized_without_leader()
+        self.p.can_create_replica_without_leader = MagicMock(return_value=True)
+        self.assertEquals(self.ha.bootstrap(), "trying to bootstrap without leader")
+
     def test_bootstrap_initialize_lock_failed(self):
         self.ha.cluster = get_cluster_not_initialized_without_leader()
         self.assertEquals(self.ha.bootstrap(), 'failed to acquire initialize lock')
@@ -335,3 +351,12 @@ class TestHa(unittest.TestCase):
         self.ha.fetch_node_status(member)
         member = Member(0, 'test', 1, {'api_url': 'http://localhost:8011/patroni'})
         self.ha.fetch_node_status(member)
+
+    def test_post_recover(self):
+        self.p.is_running = false
+        self.ha.has_lock = true
+        self.assertEqual(self.ha.post_recover(), 'removed leader key after trying and failing to start postgres')
+        self.ha.has_lock = false
+        self.assertEqual(self.ha.post_recover(), 'failed to start postgres')
+        self.p.is_running = true
+        self.assertIsNone(self.ha.post_recover())

tests/test_patroni.py

@@ -80,3 +80,8 @@ class TestPatroni(unittest.TestCase):
         self.assertTrue(self.p.nofailover)
         self.p.tags['nofailover'] = None
         self.assertFalse(self.p.nofailover)
+
+    def test_replicatefrom(self):
+        self.assertIsNone(self.p.replicatefrom)
+        self.p.tags['replicatefrom'] = 'foo'
+        self.assertEqual(self.p.replicatefrom, 'foo')

tests/test_postgresql.py

@@ -181,7 +181,8 @@ class TestPostgresql(unittest.TestCase):
             os.makedirs(self.p.data_dir)
         self.leadermem = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres'})
         self.leader = Leader(-1, 28, self.leadermem)
-        self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres'})
+        self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres',
+                            'tags': {'replicatefrom': 'leader'}})
         self.me = Member(0, 'test0', 28, {'conn_url': 'postgres://replicator:[email protected]:5434/postgres'})
 
     def tearDown(self):
@@ -228,8 +229,8 @@ class TestPostgresql(unittest.TestCase):
         self.p.write_pgpass({'host': 'localhost', 'port': '5432', 'user': 'foo', 'password': 'bar'})
 
     @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
-    def test_sync_from_leader(self):
-        self.assertTrue(self.p.sync_from_leader(self.leader))
+    def test_sync_replica(self):
+        self.assertTrue(self.p.sync_replica(self.leader))
 
     @patch('subprocess.call', side_effect=Exception("Test"))
     @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
@@ -242,25 +243,25 @@ class TestPostgresql(unittest.TestCase):
     @patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True))
     @patch('patroni.postgresql.Postgresql.single_user_mode', MagicMock(return_value=1))
     @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
-    def test_follow_the_leader(self, mock_pg_rewind):
-        self.p.demote()
-        self.p.follow_the_leader(None)
-        self.p.demote()
-        self.p.follow_the_leader(self.leader)
-        self.p.follow_the_leader(Leader(-1, 28, self.other))
+    def test_follow(self, mock_pg_rewind):
+        self.p.follow(None)
+        self.p.follow(self.leader)
+        self.p.follow(Leader(-1, 28, self.other))
         self.p.rewind = mock_pg_rewind
-        self.p.follow_the_leader(self.leader)
+        self.p.follow(self.leader)
         self.p.require_rewind()
         with mock.patch('os.path.islink', MagicMock(return_value=True)):
             with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)):
                 with mock.patch('os.unlink', MagicMock(return_value=True)):
-                    self.p.follow_the_leader(self.leader, recovery=True)
+                    self.p.follow(self.leader, recovery=True)
         self.p.require_rewind()
         with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)):
             self.p.rewind.return_value = True
-            self.p.follow_the_leader(self.leader, recovery=True)
+            self.p.follow(self.leader, recovery=True)
             self.p.rewind.return_value = False
-            self.p.follow_the_leader(self.leader, recovery=True)
+            self.p.follow(self.leader, recovery=True)
+        with mock.patch('patroni.postgresql.Postgresql.check_recovery_conf', MagicMock(return_value=True)):
+            self.assertTrue(self.p.follow(None))
 
     def test_can_rewind(self):
         tmp = self.p.pg_rewind
@@ -307,6 +308,9 @@ class TestPostgresql(unittest.TestCase):
         self.p.query = Mock(side_effect=psycopg2.OperationalError)
         self.p.schedule_load_slots = True
         self.p.sync_replication_slots(cluster)
+        self.p.schedule_load_slots = False
+        with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')):
+            self.p.sync_replication_slots(cluster)
 
     @patch.object(MockConnect, 'closed', 2)
     def test__query(self):
@@ -366,7 +370,8 @@ class TestPostgresql(unittest.TestCase):
         with patch('subprocess.call', Mock(return_value=1)):
             self.assertRaises(PostgresException, self.p.bootstrap)
         self.p.bootstrap()
-        self.p.bootstrap(self.leader)
+        with patch('patroni.postgresql.Postgresql.sync_replica', MagicMock(return_value=True)):
+            self.p.bootstrap(self.leader)
 
     def test_remove_data_directory(self):
         self.p.data_dir = 'data_dir'
@@ -480,3 +485,18 @@ class TestPostgresql(unittest.TestCase):
     def test_restore_configuration_files(self, mock_copy):
         shutil.copy = mock_copy
         self.p.restore_configuration_files()
+
+    def test_can_create_replica_without_leader(self):
+        self.p.config['create_replica_method'] = []
+        self.assertFalse(self.p.can_create_replica_without_leader())
+        self.p.config['create_replica_method'] = ['wale', 'basebackup']
+        self.p.config['wale'] = {'command': 'foo', 'no_master': 1}
+        self.assertTrue(self.p.can_create_replica_without_leader())
+
+    def test_replica_method_can_work_without_leader(self):
+        self.assertFalse(self.p.replica_method_can_work_without_leader('basebackup'))
+        self.assertFalse(self.p.replica_method_can_work_without_leader('foobar'))
+        self.p.config['foo'] = {'command': 'bar', 'no_master': 1}
+        self.assertTrue(self.p.replica_method_can_work_without_leader('foo'))
+        self.p.config['foo'] = {'command': 'bar'}
+        self.assertFalse(self.p.replica_method_can_work_without_leader('foo'))

tests/test_wale_restore.py

@@ -3,7 +3,7 @@ from mock import MagicMock, patch, PropertyMock
 import os
 import psycopg2
 import subprocess
-from patroni.scripts.wale_restore import WALERestore
+from patroni.scripts.wale_restore import WALERestore, main
 
 
 def fake_cursor_fetchone(*args, **kwargs):
@@ -58,7 +58,7 @@ class TestWALERestore(unittest.TestCase):
 
     def setUp(self):
         self.wale_restore = WALERestore("batman", "/data",
-                                        "host=batman port=5432 user=batman", "/etc", 100, 100, 1)
+                                        "host=batman port=5432 user=batman", "/etc", 100, 100, 1, 0)
 
     def tearDown(self):
         pass
@@ -76,6 +76,8 @@ class TestWALERestore(unittest.TestCase):
             self.assertFalse(self.wale_restore.should_use_s3_to_create_replica())
 
         self.wale_restore.should_use_s3_to_create_replica()
+        self.wale_restore.no_master = 1
+        self.assertTrue(self.wale_restore.should_use_s3_to_create_replica())
 
     def test_create_replica_with_s3(self):
         with patch('subprocess.call', MagicMock(return_value=0)):
@@ -89,3 +91,7 @@ class TestWALERestore(unittest.TestCase):
         with patch.object(self.wale_restore, 'should_use_s3_to_create_replica', MagicMock(return_value=True)):
             with patch.object(self.wale_restore, 'create_replica_with_s3', MagicMock(return_value=0)):
                 self.assertEqual(self.wale_restore.run(), 0)
+
+    def test_main(self):
+        with patch('sys.exit', MagicMock(return_value=0)):
+            self.assertEqual(main(), None)