Merge pull request #151 from zalando/feature/base_backup_from_the_replica

First implementation of cloning from the replica.
2026-08-25 14:53:37 +00:00 · 2016-03-11 16:57:06 +01:00
parent 25106e97f3 d965d21ada
commit e802bba5f9
6 changed files with 70 additions and 47 deletions
@@ -35,6 +35,10 @@ class Patroni(object):
    def replicatefrom(self):
        return self.tags.get('replicatefrom')

+    @property
+    def clonefrom(self):
+        return self.tags.get('clonefrom')
+
    @staticmethod
    def get_dcs(name, config):
        if 'etcd' in config:
@@ -146,6 +146,9 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem
    def has_member(self, member_name):
        return any(m for m in self.members if m.name == member_name)

+    def get_member(self, member_name):
+        return ([m for m in self.members if m.name == member_name] or [None])[0]
+

 class AbstractDCS(object):

@@ -65,19 +65,25 @@ class Ha(object):
                pass
        self.dcs.touch_member(json.dumps(data, separators=(',', ':')))

-    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')
+    def clone(self, clone_member, clone_member_name="leader"):
+        if self.state_handler.bootstrap(cluster_initialized=True, clone_member=clone_member):
+            logger.info('bootstrapped from {0}'.format(clone_member_name)
+                        if clone_member else 'bootstrapped without leader')
        else:
            self.state_handler.stop('immediate')
            self.state_handler.remove_data_directory()
-            logger.error('failed to bootstrap from leader' if leader else 'failed to bootstrap (without leader)')
+            logger.error('failed to bootstrap from {0}'.format(clone_member_name)
+                         if clone_member 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.clone, args=(self.cluster.leader, ))
-            return 'trying to bootstrap from leader'
+            clonefrom = self.patroni.clonefrom
+            clone_member = self.cluster.get_member(clonefrom)\
+                if self.cluster.has_member(clonefrom) else self.cluster.leader
+            clone_member_name = 'leader' if clone_member == self.cluster.leader else 'replica \'{0}\''.format(clonefrom)
+            self._async_executor.schedule('bootstrap from {0}'.format(clone_member_name))
+            self._async_executor.run_async(self.clone, args=(clone_member, clone_member_name))
+            return 'trying to bootstrap from {0}'.format(clone_member_name)
        elif not self.cluster.initialize and not self.patroni.nofailover:  # no initialize key
            if self.dcs.initialize(create_new=True):  # race for initialization
                try:
@@ -96,7 +102,7 @@ class Ha(object):
            else:
                return 'failed to acquire initialize lock'
        else:
-            if self.state_handler.can_create_replica_without_leader():
+            if self.state_handler.can_create_replica_without_replication_connection():
                self._async_executor.run_async(self.clone, args=(None, ))
                return "trying to bootstrap without leader"
            return 'waiting for leader to bootstrap'
@@ -233,9 +233,10 @@ class Postgresql(object):
        env['PGPASSFILE'] = self.pgpass
        return env

-    def sync_replica(self, leader):
-        env = self.write_pgpass(parseurl(leader.conn_url)) if leader else os.environ.copy()
-        if self.create_replica(leader, env) == 0:
+    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
@@ -248,33 +249,35 @@ class Postgresql(object):
        """
        return ' '.join('{0}={1}'.format(param, val) for param, val in sorted(conn.items()))

-    def replica_method_can_work_without_leader(self, method):
+    def replica_method_can_work_without_replication_connection(self, method):
        return method != 'basebackup' and self.config and self.config.get(method, {}).get('no_master')

-    def can_create_replica_without_leader(self):
+    def can_create_replica_without_replication_connection(self):
        """ go through the replication methods to see if there are ones
-            that does not require a running leader to create the replica.
+            that does not require a working replication connection.
        """
        replica_methods = self.config.get('create_replica_method', [])
-        return any(self.replica_method_can_work_without_leader(replica_method) for replica_method in replica_methods)
+        return any(self.replica_method_can_work_without_replication_connection(replica_method)
+                   for replica_method in replica_methods)

-    def create_replica(self, leader, env):
+    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 = leader.conn_url if leader else ""
+        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 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
+        # 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)]\
+            if not clone_member else replica_methods
        # 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(leader, env)
+                ret = self.basebackup(clone_member, env)
                if ret == 0:
                    logger.info("replica has been created using basebackup")
                    # if basebackup succeeds, exit with success
@@ -698,18 +701,19 @@ $$""".format(name, options), name, password, password)
    def last_operation(self):
        return str(self.xlog_position())

-    def bootstrap(self, cluster_initialized=False, current_leader=None):
+    def bootstrap(self, cluster_initialized=False, clone_member=None):
        """
            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 from an existing member (master or replica)
             - initialize the replica using the replica creation method that
-               works without the master (i.e. restore from on-disk base backup)
+               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 from from
-            the master-less replica creation method (in the latter case, there is
+            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).

@@ -719,7 +723,7 @@ $$""".format(name, options), name, password, password)
            that should be retried in the future.
        """
        ret = False
-        if not (cluster_initialized or current_leader):
+        if not (cluster_initialized or clone_member):
            ret = self.initialize() and self.start()
            if ret:
                self.create_replication_user()
@@ -727,9 +731,9 @@ $$""".format(name, options), name, password, password)
            else:
                raise PostgresException("Could not bootstrap master PostgreSQL")
        else:
-            if self.sync_replica(current_leader):
+            if self.sync_replica(clone_member):
                self.restore_configuration_files()
-                self.write_recovery_conf(current_leader, True)
+                self.write_recovery_conf(clone_member, True)
                ret = self.start()
        return ret

@@ -757,12 +761,12 @@ $$""".format(name, options), name, password, password)
            logger.exception('Could not remove data directory %s', self.data_dir)
            self.move_data_directory()

-    def basebackup(self, leader, env):
+    def basebackup(self, clone_member, 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 = leader.conn_url
+        master_connection = clone_member.conn_url
        maxfailures = 2
        ret = 1
        for bbfailures in range(0, maxfailures):
@@ -29,12 +29,12 @@ def get_cluster_not_initialized_without_leader():


 def get_cluster_initialized_without_leader(leader=False, failover=None):
-    m = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres',
-                                 'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4})
-    l = Leader(0, 0, m) if leader else None
-    o = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
-                                'api_url': 'http://127.0.0.1:8011/patroni'})
-    return get_cluster(True, l, [m, o], failover)
+    m1 = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres',
+                                  'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4})
+    l = Leader(0, 0, m1) if leader else None
+    m2 = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
+                                 'api_url': 'http://127.0.0.1:8011/patroni'})
+    return get_cluster(True, l, [m1, m2], failover)


 def get_cluster_initialized_with_leader(failover=None):
@@ -57,6 +57,7 @@ class MockPatroni(object):
        self.nap_time = 10
        self.replicatefrom = None
        self.api.connection_string = 'http://127.0.0.1:8008'
+        self.clonefrom = None


 def run_async(func, args=()):
@@ -87,7 +88,7 @@ class TestHa(unittest.TestCase):
                                 'replication': {'username': '', 'password': '', 'network': ''}})
            self.p.set_state('running')
            self.p.check_replication_lag = true
-            self.p.can_create_replica_without_leader = MagicMock(return_value=False)
+            self.p.can_create_replica_without_replication_connection = 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
@@ -205,13 +206,18 @@ class TestHa(unittest.TestCase):
        self.p.bootstrap = false
        self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap from leader')

+    def test_bootstrap_from_another_member(self):
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.ha.patroni.clonefrom = 'other'
+        self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap from replica \'other\'')
+
    def test_bootstrap_waiting_for_leader(self):
        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.p.can_create_replica_without_replication_connection = MagicMock(return_value=True)
        self.assertEquals(self.ha.bootstrap(), "trying to bootstrap without leader")

    def test_bootstrap_initialize_lock_failed(self):
@@ -471,17 +471,17 @@ class TestPostgresql(unittest.TestCase):
    def test_restore_configuration_files(self):
        self.p.restore_configuration_files()

-    def test_can_create_replica_without_leader(self):
+    def test_can_create_replica_without_replication_connection(self):
        self.p.config['create_replica_method'] = []
-        self.assertFalse(self.p.can_create_replica_without_leader())
+        self.assertFalse(self.p.can_create_replica_without_replication_connection())
        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())
+        self.assertTrue(self.p.can_create_replica_without_replication_connection())

-    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'))
+    def test_replica_method_can_work_without_replication_connection(self):
+        self.assertFalse(self.p.replica_method_can_work_without_replication_connection('basebackup'))
+        self.assertFalse(self.p.replica_method_can_work_without_replication_connection('foobar'))
        self.p.config['foo'] = {'command': 'bar', 'no_master': 1}
-        self.assertTrue(self.p.replica_method_can_work_without_leader('foo'))
+        self.assertTrue(self.p.replica_method_can_work_without_replication_connection('foo'))
        self.p.config['foo'] = {'command': 'bar'}
-        self.assertFalse(self.p.replica_method_can_work_without_leader('foo'))
+        self.assertFalse(self.p.replica_method_can_work_without_replication_connection('foo'))