From 15bec1e28c0ca2ca82015fa8fa06b3ae5c8a9c94 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 26 Jan 2016 15:24:32 +0100 Subject: [PATCH] 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 | 9 ++++-- patroni/postgresql.py | 57 ++++++++++++++++++++++++--------- patroni/scripts/wale_restore.py | 37 ++++++++++++--------- tests/test_ha.py | 16 +++++++++ tests/test_postgresql.py | 23 ++++++++++++- tests/test_wale_restore.py | 4 ++- 6 files changed, 111 insertions(+), 35 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 3856366d..0feb172b 100644 --- a/patroni/ha.py +++ b/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): diff --git a/patroni/postgresql.py b/patroni/postgresql.py index db47cbff..f329fad5 100644 --- a/patroni/postgresql.py +++ b/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 diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py index f54c37a3..58fdd420 100755 --- a/patroni/scripts/wale_restore.py +++ b/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 diff --git a/tests/test_ha.py b/tests/test_ha.py index 336b78a3..4c8500f4 100644 --- a/tests/test_ha.py +++ b/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()) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 8114272b..3681562f 100644 --- a/tests/test_postgresql.py +++ b/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')) diff --git a/tests/test_wale_restore.py b/tests/test_wale_restore.py index 05f34187..ba4c3bea 100644 --- a/tests/test_wale_restore.py +++ b/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)):