Remove data directory only if replica creation failed

And follow the right node after replica creation (it was following
the same node from which it took the backup)

patroni/ha.py

@@ -67,11 +67,13 @@ class Ha(object):
         self.dcs.touch_member(json.dumps(data, separators=(',', ':')))
 
     def clone(self, clone_member=None, msg='(without leader)'):
-        if self.state_handler.bootstrap(cluster_initialized=True, clone_member=clone_member):
+        if self.state_handler.clone(clone_member):
             logger.info('bootstrapped %s', msg)
+            cluster = self.dcs.get_cluster()
+            node_to_follow = self._get_node_to_follow(cluster)
+            self.state_handler.follow(node_to_follow, True)
         else:
             logger.error('failed to bootstrap %s', msg)
-            self.state_handler.stop('immediate')
             self.state_handler.remove_data_directory()
 
     def bootstrap(self):
@@ -109,22 +111,27 @@ class Ha(object):
         self.recovering = True
         return self.follow("starting as readonly because i had the session lock", "starting as a secondary", True, True)
 
+    def _get_node_to_follow(self, cluster):
+        # determine the node to follow. If replicatefrom tag is set,
+        # try to follow the node mentioned there, otherwise, follow the leader.
+        if not self.patroni.replicatefrom or self.patroni.replicatefrom == self.state_handler.name:
+            node_to_follow = cluster.leader
+        else:
+            node_to_follow = cluster.get_member(self.patroni.replicatefrom)
+
+        return node_to_follow if node_to_follow and node_to_follow.name != self.state_handler.name else None
+
     def follow(self, demote_reason, follow_reason, refresh=True, recovery=False):
         if refresh:
             self.load_cluster_from_dcs()
 
-        ret = demote_reason if not recovery and self.state_handler.is_leader() 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 = self.cluster.get_member(self.patroni.replicatefrom, fallback_to_leader=True)
+        if recovery:
+            ret = demote_reason if self.has_lock() else follow_reason
         else:
-            node_to_follow = self.cluster.leader
-        if node_to_follow and node_to_follow.name == self.state_handler.name:
-            ret = demote_reason
-            node_to_follow = None
+            ret = demote_reason if self.state_handler.is_leader() else follow_reason
+
+        node_to_follow = self._get_node_to_follow(self.cluster)
+
         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, (node_to_follow, recovery))

patroni/postgresql.py

@@ -126,7 +126,12 @@ class Postgresql(object):
         return local_address + ':' + self.port
 
     def get_postgres_role_from_data_directory(self):
-        return 'replica' if os.path.exists(self.recovery_conf) else 'master'
+        if self.data_directory_empty():
+            return 'uninitialized'
+        elif os.path.exists(self.recovery_conf):
+            return 'replica'
+        else:
+            return 'master'
 
     @property
     def _connect_kwargs(self):
@@ -233,14 +238,6 @@ class Postgresql(object):
         env['PGPASSFILE'] = self.pgpass
         return env
 
-    def sync_replica(self, clone_member):
-        # add the credentials to connect to the replica origin to pgpass.
-        env = self.write_pgpass(parseurl(clone_member.conn_url)) if clone_member else os.environ.copy()
-        if self.create_replica(clone_member, env) == 0:
-            self.delete_trigger_file()
-            return True
-        return False
-
     @staticmethod
     def build_connstring(conn):
         """
@@ -257,26 +254,36 @@ class Postgresql(object):
             that does not require a working replication connection.
         """
         replica_methods = self.config.get('create_replica_method', [])
-        return any(self.replica_method_can_work_without_replication_connection(replica_method)
-                   for replica_method in replica_methods)
+        return any(self.replica_method_can_work_without_replication_connection(method) for method in replica_methods)
+
+    def create_replica(self, clone_member):
+        """
+            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
+        """
 
-    def create_replica(self, clone_member, 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 = clone_member.conn_url if clone_member 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 source, leave only replica methods that work without it
-        replica_methods = replica_methods if clone_member else \
-            [r for r in replica_methods if self.replica_method_can_work_without_replication_connection(r)]
+
+        if clone_member:
+            connstring = clone_member.conn_url
+            # add the credentials to connect to the replica origin to pgpass.
+            env = self.write_pgpass(parseurl(clone_member.conn_url))
+        else:
+            connstring = ''
+            env = os.environ.copy()
+            # if we don't have any source, leave only replica methods that work without it
+            replica_methods = \
+                [r for r in replica_methods if self.replica_method_can_work_without_replication_connection(r)]
+
         # go through them in priority order
         ret = 1
         for replica_method in replica_methods:
             # if the method is basebackup, then use the built-in
             if replica_method == "basebackup":
-                ret = self.basebackup(clone_member, env)
+                ret = self.basebackup(connstring, env)
                 if ret == 0:
                     logger.info("replica has been created using basebackup")
                     # if basebackup succeeds, exit with success
@@ -302,10 +309,10 @@ class Postgresql(object):
                     ret = subprocess.call(shlex.split(cmd) + params, env=env)
                     # if we succeeded, stop
                     if ret == 0:
-                        logger.info("replica has been created using {0}".format(replica_method))
+                        logger.info('replica has been created using %s', replica_method)
                         break
-                except Exception as e:
-                    logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, str(e)))
+                except Exception:
+                    logger.exception('Error creating replica using method %s', replica_method)
                     ret = 1
 
         return ret
@@ -693,41 +700,32 @@ $$""".format(name, options), name, password, password)
     def last_operation(self):
         return str(self.xlog_position())
 
-    def bootstrap(self, cluster_initialized=False, clone_member=None):
+    def clone(self, clone_member):
         """
-            Populate PostgreSQL data directory by doing one of the following:
-             - create with initdb if there is no master.
              - initialize the replica from an existing member (master or replica)
              - initialize the replica using the replica creation method that
                works without the replication connection (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 using 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).
+        ret = self.create_replica(clone_member) == 0
+        if ret:
+            self.delete_trigger_file()
+            self.restore_configuration_files()
+        return ret
 
+    def bootstrap(self):
+        """
             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 (cluster_initialized or clone_member):
-            ret = self.initialize() and self.start()
-            if ret:
-                self.create_replication_user()
-                self.create_connection_user()
-            else:
-                raise PostgresException("Could not bootstrap master PostgreSQL")
-        else:
-            if self.sync_replica(clone_member):
-                self.restore_configuration_files()
-                self.write_recovery_conf(clone_member)
-                ret = self.start()
-        return ret
+        if self.initialize() and self.start():
+            self.create_replication_user()
+            self.create_connection_user()
+            return True
+        raise PostgresException("Could not bootstrap master PostgreSQL")
 
     def move_data_directory(self):
         if os.path.isdir(self.data_dir) and not self.is_running():
@@ -753,18 +751,17 @@ $$""".format(name, options), name, password, password)
             logger.exception('Could not remove data directory %s', self.data_dir)
             self.move_data_directory()
 
-    def basebackup(self, clone_member, env):
+    def basebackup(self, conn_url, env):
         # creates a replica data dir using pg_basebackup.
         # this is the default, built-in create_replica_method
         # tries twice, then returns failure (as 1)
         # uses "stream" as the xlog-method to avoid sync issues
-        master_connection = clone_member.conn_url
         maxfailures = 2
         ret = 1
         for bbfailures in range(0, maxfailures):
             try:
                 ret = subprocess.call(['pg_basebackup', '--pgdata=' + self.data_dir,
-                                       '--xlog-method=stream', "--dbname=" + master_connection], env=env)
+                                       '--xlog-method=stream', "--dbname=" + conn_url], env=env)
                 if ret == 0:
                     break
 

tests/test_postgresql.py

@@ -239,12 +239,6 @@ class TestPostgresql(unittest.TestCase):
     def test_write_pgpass(self):
         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_replica(self):
-        self.assertTrue(self.p.sync_replica(self.leader))
-        self.p.create_replica = Mock(return_value=1)
-        self.assertFalse(self.p.sync_replica(self.leader))
-
     @patch('subprocess.call', side_effect=OSError)
     @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
     def test_pg_rewind(self, mock_call):
@@ -294,19 +288,19 @@ class TestPostgresql(unittest.TestCase):
     def test_create_replica(self):
         self.p.delete_trigger_file = Mock(side_effect=OSError)
         with patch('subprocess.call', Mock(side_effect=[1, 0])):
-            self.assertEquals(self.p.create_replica(self.leader, ''), 0)
+            self.assertEquals(self.p.create_replica(self.leader), 0)
         with patch('subprocess.call', Mock(side_effect=[Exception(), 0])):
-            self.assertEquals(self.p.create_replica(self.leader, ''), 0)
+            self.assertEquals(self.p.create_replica(self.leader), 0)
 
         self.p.config['create_replica_method'] = ['wale', 'basebackup']
         self.p.config['wale'] = {'command': 'foo'}
         with patch('subprocess.call', Mock(return_value=0)):
-            self.assertEquals(self.p.create_replica(self.leader, ''), 0)
+            self.assertEquals(self.p.create_replica(self.leader), 0)
             del self.p.config['wale']
-            self.assertEquals(self.p.create_replica(self.leader, ''), 0)
+            self.assertEquals(self.p.create_replica(self.leader), 0)
 
         with patch('subprocess.call', Mock(side_effect=Exception("foo"))):
-            self.assertEquals(self.p.create_replica(self.leader, ''), 1)
+            self.assertEquals(self.p.create_replica(self.leader), 1)
 
     def test_sync_replication_slots(self):
         self.p.start()
@@ -372,13 +366,19 @@ class TestPostgresql(unittest.TestCase):
         with patch('os.rename', Mock(side_effect=OSError)):
             self.p.move_data_directory()
 
-    @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
     def test_bootstrap(self):
         with patch('subprocess.call', Mock(return_value=1)):
             self.assertRaises(PostgresException, self.p.bootstrap)
         self.p.bootstrap()
-        with patch('patroni.postgresql.Postgresql.sync_replica', MagicMock(return_value=True)):
-            self.p.bootstrap(self.leader)
+
+    @patch('patroni.postgresql.Postgresql.create_replica', Mock(return_value=0))
+    def test_clone(self):
+        self.p.clone(self.leader)
+
+    @patch('os.listdir', Mock(return_value=['recovery.conf']))
+    @patch('os.path.exists', Mock(return_value=True))
+    def test_get_postgres_role_from_data_directory(self):
+        self.assertEquals(self.p.get_postgres_role_from_data_directory(), 'replica')
 
     def test_remove_data_directory(self):
         self.p.data_dir = 'data_dir'