Support new replicas without the master.

The replication method should have no_master flag set and
support getting the base backup from some external storage
(i.e. S3). At the moment we only support initialization of
replicas in the already existing cluster when no master is
present, since there is no 'one fits all' way to decide whether
to run initidb or wait for the replica data storage to become
available when dealing with the new cluster.

patroni/ha.py

@@ -63,12 +63,12 @@ class Ha:
         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')
+        if self.state_handler.bootstrap(True, 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
@@ -92,6 +92,9 @@ 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.copy_backup_from_leader, args=(None, ))
+                return "trying to bootstrap without leader"
             return 'waiting for leader to bootstrap'
 
     def recover(self):

patroni/postgresql.py

@@ -220,9 +220,9 @@ class Postgresql:
         return env
 
     def sync_from_leader(self, leader):
-        r = parseurl(leader.conn_url)
-
-        env = self.write_pgpass(r)
+        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
@@ -235,14 +235,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:
@@ -435,7 +451,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'
@@ -444,6 +460,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))
 
@@ -658,18 +675,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, initialize=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 (initialize or current_leader):
             ret = self.initialize() and self.start()
             if ret:
                 self.create_replication_user()
@@ -679,7 +706,7 @@ $$""".format(name, options), name, password, password)
         else:
             if self.sync_from_leader(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

tests/test_ha.py

@@ -103,6 +103,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
@@ -139,6 +140,7 @@ class TestHa(unittest.TestCase):
         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')
 
@@ -223,6 +225,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')
@@ -342,3 +349,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_postgresql.py

@@ -259,6 +259,8 @@ class TestPostgresql(unittest.TestCase):
             self.p.follow(self.leader, recovery=True)
             self.p.rewind.return_value = False
             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
@@ -305,6 +307,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):
@@ -364,7 +369,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_from_leader', MagicMock(return_value=True)):
+            self.p.bootstrap(self.leader)
 
     def test_remove_data_directory(self):
         self.p.data_dir = 'data_dir'
@@ -478,3 +484,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

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