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