diff --git a/patroni/__init__.py b/patroni/__init__.py index 1ded41aa..b30c2239 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -31,6 +31,10 @@ class Patroni: def nofailover(self): return self.tags.get('nofailover', False) + @property + def replicatefrom(self): + return self.tags.get('replicatefrom') + @staticmethod def get_dcs(name, config): if 'etcd' in config: diff --git a/patroni/dcs.py b/patroni/dcs.py index f640787c..a6b9d856 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -67,6 +67,10 @@ class Member(namedtuple('Member', 'index,name,session,data')): def nofailover(self): return self.data.get('tags', {}).get('nofailover', False) + @property + def replicatefrom(self): + return self.data.get('tags', {}).get('replicatefrom') + class Leader(namedtuple('Leader', 'index,session,member')): @@ -107,6 +111,9 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem def is_unlocked(self): return not (self.leader and self.leader.name) + def has_member(self, member_name): + return any(m for m in self.members if m.name == member_name) + class AbstractDCS: diff --git a/patroni/ha.py b/patroni/ha.py index 3a8d46d7..7e78eb77 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -19,12 +19,13 @@ class Ha: self.dcs = patroni.dcs self.cluster = None self.old_cluster = None + self.recovering = False self._async_executor = AsyncExecutor() def load_cluster_from_dcs(self): cluster = self.dcs.get_cluster() - # We want to keep the state of cluster when it was healhy + # We want to keep the state of cluster when it was healthy if not cluster.is_unlocked() or not self.old_cluster: self.old_cluster = cluster self.cluster = cluster @@ -61,18 +62,18 @@ class Ha: pass 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') + 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') 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 self._async_executor.schedule('bootstrap from leader') - self._async_executor.run_async(self.copy_backup_from_leader, args=(self.cluster.leader, )) + self._async_executor.run_async(self.clone, args=(self.cluster.leader, )) return 'trying to bootstrap from leader' elif not self.cluster.initialize and not self.patroni.nofailover: # no initialize key if self.dcs.initialize(create_new=True): # race for initialization @@ -91,41 +92,38 @@ 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.clone, args=(None, )) + return "trying to bootstrap without leader" return 'waiting for leader to bootstrap' def recover(self): - has_lock = self.has_lock() - # try to see if we are the former master that crashed. If so - we likely need to run pg_rewind # in order to join the former standby being promoted. pg_controldata = self.state_handler.controldata() - if not has_lock and pg_controldata and\ + if (self.state_handler.role == 'master') and pg_controldata and\ pg_controldata.get('Database cluster state', '') == 'in production': # crashed master self.state_handler.require_rewind() + self.recovering = True + return self.follow("started as readonly because i had the session lock", + "started as a secondary", + refresh=True, recovery=True) - # XXX: follow the leader calls stop, which might take quite some time. - # perhaps we should run sync asynchronously - # (we still need the exit code from follow_the_leader) - ret = self.state_handler.follow_the_leader(None if has_lock else self.cluster.leader, recovery=True) - if not ret: - if not has_lock: - return 'failed to start postgres' - self.dcs.delete_leader() - self.dcs.reset_cluster() - return 'removed leader key after trying and failing to start postgres' - if not has_lock: - return 'started as a secondary' - logger.info('started as readonly because i had the session lock') - self.load_cluster_from_dcs() - - def follow_the_leader(self, demote_reason, follow_reason, refresh=True): + def follow(self, demote_reason, follow_reason, refresh=True, recovery=False): refresh and self.load_cluster_from_dcs() - ret = demote_reason if self.state_handler.is_leader() else follow_reason - leader = self.cluster.leader - leader = None if (leader and leader.name) == self.state_handler.name else leader - if not self.state_handler.check_recovery_conf(leader): + ret = demote_reason if (not recovery and self.state_handler.is_leader() or + recovery and self.state_handler.role == 'master') 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 = [m for m in self.cluster.members if m.name == self.patroni.replicatefrom] + node_to_follow = node_to_follow[0] if node_to_follow else self.cluster.leader + else: + node_to_follow = self.cluster.leader + node_to_follow = None if node_to_follow and node_to_follow.name == self.state_handler.name else node_to_follow + 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_the_leader, (leader, )) + self._async_executor.run_async(self.state_handler.follow, (node_to_follow, recovery)) return ret def enforce_master_role(self, message, promote_message): @@ -298,14 +296,14 @@ class Ha: return self.enforce_master_role('acquired session lock as a leader', 'promoted self to leader by acquiring session lock') else: - return self.follow_the_leader('demoted self due after trying and failing to obtain lock', - 'following new leader after trying and failing to obtain lock') + return self.follow('demoted self after trying and failing to obtain lock', + 'following new leader after trying and failing to obtain lock') else: if self.patroni.nofailover: - return self.follow_the_leader('demoting self because I am not allowed to become master', - 'following a different leader because I am not allowed to promote') - return self.follow_the_leader('demoting self because i am not the healthiest node', - 'following a different leader because i am not the healthiest node') + return self.follow('demoting self because I am not allowed to become master', + 'following a different leader because I am not allowed to promote') + return self.follow('demoting self because i am not the healthiest node', + 'following a different leader because i am not the healthiest node') def process_healthy_cluster(self): if self.has_lock(): @@ -323,8 +321,8 @@ class Ha: self.load_cluster_from_dcs() else: logger.info('does not have lock') - return self.follow_the_leader('demoting self because i do not have the lock and i was a leader', - 'no action. i am a secondary and i am following a leader', False) + return self.follow('demoting self because i do not have the lock and i was a leader', + 'no action. i am a secondary and i am following a leader', False) def schedule(self, action): with self._async_executor: @@ -352,7 +350,7 @@ class Ha: def reinitialize(self, cluster): self.state_handler.stop('immediate') self.state_handler.remove_data_directory() - self.copy_backup_from_leader(cluster.leader) + self.clone(cluster.leader) def process_scheduled_action(self): if self.reinitialize_scheduled(): @@ -382,6 +380,15 @@ class Ha: # so even 1 << 32 would have 10 digits. return str(sysid) and len(str(sysid)) >= 10 and str(sysid).isdigit() + def post_recover(self): + if not self.state_handler.is_running(): + if self.has_lock(): + self.dcs.delete_leader() + self.dcs.reset_cluster() + return 'removed leader key after trying and failing to start postgres' + return 'failed to start postgres' + return None + def _run_cycle(self): try: self.load_cluster_from_dcs() @@ -395,6 +402,13 @@ class Ha: if self._async_executor.busy: return self.handle_long_action_in_progress() + # we've got here, so any async action has finished. Check if we tried to recover and failed + if self.recovering: + self.recovering = False + msg = self.post_recover() + if msg is not None: + return msg + # currently it can trigger only reinitialize msg = self.process_scheduled_action() if msg is not None: @@ -425,14 +439,18 @@ class Ha: else: return self.process_healthy_cluster() finally: - self.state_handler.sync_replication_slots(self.cluster) + # we might not have a valid PostgreSQL connection here if another thread + # stops PostgreSQL, therefore, we only reload replication slots if no + # asynchronous processes are running (should be always the case for the master) + if not self._async_executor.busy: + self.state_handler.sync_replication_slots(self.cluster) except DCSError: logger.error('Error communicating with DCS') if self.state_handler.is_running() and self.state_handler.is_leader(): self.demote(delete_leader=False) return 'demoted self because DCS is not accessible and i was a leader' except (psycopg2.Error, PostgresConnectionException): - logger.exception('Error communicating with Postgresql. Will try again later') + logger.exception('Error communicating with PostgreSQL. Will try again later') def run_cycle(self): with self._async_executor: diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 5e2312cb..0e279402 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -221,10 +221,10 @@ class Postgresql: env['PGPASSFILE'] = self.pgpass return env - def sync_from_leader(self, leader): - r = parseurl(leader.conn_url) - - env = self.write_pgpass(r) + def sync_replica(self, leader): + 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 @@ -237,14 +237,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: @@ -437,7 +453,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' @@ -446,6 +462,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)) @@ -526,7 +543,7 @@ recovery_target_timeline = 'latest' except: logger.exception("Unable to remove {}".format(path)) - def follow_the_leader(self, leader, recovery=False): + def follow(self, leader, recovery=False): if not self.check_recovery_conf(leader) or recovery: change_role = (self.role == 'master') @@ -598,9 +615,6 @@ recovery_target_timeline = 'latest' self.call_nowait(ACTION_ON_ROLE_CHANGE) return ret - def demote(self): - self.follow_the_leader(None) - def create_or_update_role(self, name, password, options): self.query("""DO $$ BEGIN @@ -639,7 +653,18 @@ $$""".format(name, options), name, password, password) if self.use_slots: try: self.load_replication_slots() - slots = [m.name for m in cluster.members if m.name != self.name] if self.role == 'master' else [] + # if the replicatefrom tag is set on the member - we should not create the replication slot for it on + # the current master, because that member would replicate from elsewhere. We still create the slot if + # the replicatefrom destination member is currently not a member of the cluster (fallback to the + # master), or if replicatefrom destination member happens to be the current master + if self.role == 'master': + slots = [m.name for m in cluster.members if m.name != self.name and + (m.replicatefrom is None or m.replicatefrom == self.name or + not cluster.has_member(m.replicatefrom))] + else: + # only manage slots for replicas that replicate from this one, except for the leader among them + slots = [m.name for m in cluster.members if m.replicatefrom == self.name and + m.name != cluster.leader.name] # drop unused slots for slot in set(self.replication_slots) - set(slots): self.query("""SELECT pg_drop_replication_slot(%s) @@ -659,18 +684,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, cluster_initialized=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 (cluster_initialized or current_leader): ret = self.initialize() and self.start() if ret: self.create_replication_user() @@ -678,9 +713,9 @@ $$""".format(name, options), name, password, password) else: raise PostgresException("Could not bootstrap master PostgreSQL") else: - if self.sync_from_leader(current_leader): + if self.sync_replica(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/postgres0.yml b/postgres0.yml index 244a0f19..8b77259e 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -87,14 +87,13 @@ postgresql: archive_mode: "on" wal_level: hot_standby archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f - max_wal_senders: 5 + max_wal_senders: 10 wal_keep_segments: 8 archive_timeout: 1800s - max_replication_slots: 5 + max_replication_slots: 10 hot_standby: "on" wal_log_hints: "on" tags: nofailover: False noloadbalance: False clonefrom: False - replicatefrom: 127.0.0.1 diff --git a/postgres1.yml b/postgres1.yml index 58fc308c..a8802226 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -88,14 +88,13 @@ postgresql: archive_mode: "on" wal_level: hot_standby archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f - max_wal_senders: 5 + max_wal_senders: 10 wal_keep_segments: 8 archive_timeout: 1800s - max_replication_slots: 5 + max_replication_slots: 10 hot_standby: "on" wal_log_hints: "on" tags: nofailover: False noloadbalance: False clonefrom: False - replicatefrom: 127.0.0.1 diff --git a/postgres2.yml b/postgres2.yml new file mode 100644 index 00000000..99e8b47e --- /dev/null +++ b/postgres2.yml @@ -0,0 +1,101 @@ +ttl: &ttl 30 +loop_wait: &loop_wait 10 +scope: &scope batman +restapi: + listen: 127.0.0.1:8010 + connect_address: 127.0.0.1:8010 + auth: 'username:password' +# certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem +# keyfile: /etc/ssl/private/ssl-cert-snakeoil.key +etcd: + scope: *scope + ttl: *ttl + host: 127.0.0.1:4001 + #discovery_srv: my-etcd.domain +#zookeeper: +# scope: *scope +# session_timeout: *ttl +# reconnect_timeout: *loop_wait +# hosts: +# - 127.0.0.1:2181 +# - 127.0.0.2:2181 +# exhibitor: +# poll_interval: 300 +# port: 8181 +# hosts: +# - host1 +# - host2 +# - host3 +postgresql: + name: postgresql2 + scope: *scope + listen: 127.0.0.1:5434 + connect_address: 127.0.0.1:5434 + data_dir: data/postgresql2 + maximum_lag_on_failover: 1048576 # 1 megabyte in bytes + use_slots: True + pgpass: /tmp/pgpass2 + initdb: ## We allow the following options to be passed on to initdb + # - auth: authmethod + # - auth-host: authmethod + # - auth-local: authmethod + - encoding: UTF8 + # - data-checksums # When pg_rewind is needed on 9.3, this needs to be enabled + # - locale: locale + # - lc-collate: locale + # - lc-ctype: locale + # - lc-messages: locale + # - lc-monetary: locale + # - lc-numeric: locale + # - lc-time: locale + # - text-search-config: CFG + # - xlogdir: directory + # - debug + # - noclean + pg_rewind: + username: postgres + password: zalando + pg_hba: + - host all all 0.0.0.0/0 md5 + - hostssl all all 0.0.0.0/0 md5 + replication: + username: replicator + password: rep-pass + network: 127.0.0.1/32 + superuser: + user: postgres + password: zalando + admin: + username: admin + password: admin +# commented-out example for wal-e provisioning + create_replica_method: + - basebackup +# - wal_e +# commented-out example for wal-e provisioning + #wal_e: + #command: /patroni/scripts/wale_restore.py + #env_dir: /home/postgres/etc/wal-e.d/env + #threshold_megabytes: 10240 + #threshold_backup_size_percentage: 30 + #retries: 2 + #use_iam: 1 + #recovery_conf: + #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -p 1 + recovery_conf: + restore_command: cp ../wal_archive/%f %p + parameters: + archive_mode: "on" + wal_level: hot_standby + archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f + max_wal_senders: 10 + wal_keep_segments: 8 + archive_timeout: 1800s + max_replication_slots: 10 + hot_standby: "on" + wal_log_hints: "on" +tags: + nofailover: False + noloadbalance: False + clonefrom: False + replicatefrom: postgresql1 diff --git a/tests/test_ha.py b/tests/test_ha.py index 614f4943..1ac164cf 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -90,6 +90,7 @@ class MockPatroni: self.api = Mock() self.tags = {} self.nofailover = None + self.replicatefrom = None self.api.connection_string = 'http://127.0.0.1:8008' @@ -104,6 +105,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 @@ -129,13 +131,19 @@ class TestHa(unittest.TestCase): def test_recover_replica_failed(self): self.p.controldata = lambda: {'Database cluster state': 'in production'} self.p.is_healthy = false - self.p.follow_the_leader = false + self.p.is_running = false + self.p.follow = false + self.assertEquals(self.ha.run_cycle(), 'started as a secondary') self.assertEquals(self.ha.run_cycle(), 'failed to start postgres') def test_recover_master_failed(self): - self.p.follow_the_leader = false + self.p.follow = false self.p.is_healthy = false + 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') @patch('sys.exit', return_value=1) @@ -146,7 +154,8 @@ class TestHa(unittest.TestCase): @patch.object(Cluster, 'is_unlocked', Mock(return_value=False)) def test_start_as_readonly(self): - self.p.is_leader = self.p.is_healthy = false + self.p.is_leader = false + self.p.is_healthy = true self.ha.has_lock = true self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock') @@ -160,7 +169,7 @@ class TestHa(unittest.TestCase): def test_demote_after_failing_to_obtain_lock(self): self.ha.acquire_lock = false - self.assertEquals(self.ha.run_cycle(), 'demoted self due after trying and failing to obtain lock') + self.assertEquals(self.ha.run_cycle(), 'demoted self after trying and failing to obtain lock') def test_follow_new_leader_after_failing_to_obtain_lock(self): self.ha.is_healthiest_node = true @@ -198,10 +207,12 @@ class TestHa(unittest.TestCase): self.ha.update_lock = false self.assertEquals(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader') - def test_follow_the_leader(self): + def test_follow(self): self.ha.cluster.is_unlocked = false self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'no action. i am a secondary and i am following a leader') + self.ha.patroni.replicatefrom = "foo" + self.assertEquals(self.ha.run_cycle(), 'no action. i am a secondary and i am following a leader') def test_no_etcd_connection_master_demote(self): self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly')) @@ -216,6 +227,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') @@ -335,3 +351,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_patroni.py b/tests/test_patroni.py index 18f5c14b..d1db99cb 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -80,3 +80,8 @@ class TestPatroni(unittest.TestCase): self.assertTrue(self.p.nofailover) self.p.tags['nofailover'] = None self.assertFalse(self.p.nofailover) + + def test_replicatefrom(self): + self.assertIsNone(self.p.replicatefrom) + self.p.tags['replicatefrom'] = 'foo' + self.assertEqual(self.p.replicatefrom, 'foo') diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 2097bcb7..a88f150c 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -181,7 +181,8 @@ class TestPostgresql(unittest.TestCase): os.makedirs(self.p.data_dir) self.leadermem = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres'}) self.leader = Leader(-1, 28, self.leadermem) - self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres'}) + self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', + 'tags': {'replicatefrom': 'leader'}}) self.me = Member(0, 'test0', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres'}) def tearDown(self): @@ -228,8 +229,8 @@ class TestPostgresql(unittest.TestCase): 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_from_leader(self): - self.assertTrue(self.p.sync_from_leader(self.leader)) + def test_sync_replica(self): + self.assertTrue(self.p.sync_replica(self.leader)) @patch('subprocess.call', side_effect=Exception("Test")) @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) @@ -242,25 +243,25 @@ class TestPostgresql(unittest.TestCase): @patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True)) @patch('patroni.postgresql.Postgresql.single_user_mode', MagicMock(return_value=1)) @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) - def test_follow_the_leader(self, mock_pg_rewind): - self.p.demote() - self.p.follow_the_leader(None) - self.p.demote() - self.p.follow_the_leader(self.leader) - self.p.follow_the_leader(Leader(-1, 28, self.other)) + def test_follow(self, mock_pg_rewind): + self.p.follow(None) + self.p.follow(self.leader) + self.p.follow(Leader(-1, 28, self.other)) self.p.rewind = mock_pg_rewind - self.p.follow_the_leader(self.leader) + self.p.follow(self.leader) self.p.require_rewind() with mock.patch('os.path.islink', MagicMock(return_value=True)): with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): with mock.patch('os.unlink', MagicMock(return_value=True)): - self.p.follow_the_leader(self.leader, recovery=True) + self.p.follow(self.leader, recovery=True) self.p.require_rewind() with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): self.p.rewind.return_value = True - self.p.follow_the_leader(self.leader, recovery=True) + self.p.follow(self.leader, recovery=True) self.p.rewind.return_value = False - self.p.follow_the_leader(self.leader, recovery=True) + 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 @@ -307,6 +308,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): @@ -366,7 +370,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_replica', MagicMock(return_value=True)): + self.p.bootstrap(self.leader) def test_remove_data_directory(self): self.p.data_dir = 'data_dir' @@ -480,3 +485,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..aa834b96 100644 --- a/tests/test_wale_restore.py +++ b/tests/test_wale_restore.py @@ -3,7 +3,7 @@ from mock import MagicMock, patch, PropertyMock import os import psycopg2 import subprocess -from patroni.scripts.wale_restore import WALERestore +from patroni.scripts.wale_restore import WALERestore, main def fake_cursor_fetchone(*args, **kwargs): @@ -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)): @@ -89,3 +91,7 @@ class TestWALERestore(unittest.TestCase): with patch.object(self.wale_restore, 'should_use_s3_to_create_replica', MagicMock(return_value=True)): with patch.object(self.wale_restore, 'create_replica_with_s3', MagicMock(return_value=0)): self.assertEqual(self.wale_restore.run(), 0) + + def test_main(self): + with patch('sys.exit', MagicMock(return_value=0)): + self.assertEqual(main(), None)