diff --git a/helpers/dcs.py b/helpers/dcs.py index c7140c22..fa812a08 100644 --- a/helpers/dcs.py +++ b/helpers/dcs.py @@ -66,6 +66,7 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem class AbstractDCS: __metaclass__ = abc.ABCMeta + initialize_key = '/initialize' def __init__(self, name, config): """ @@ -131,7 +132,7 @@ class AbstractDCS: overwriting the key if necessary.""" @abc.abstractmethod - def race(self, path): + def initialize(self): """Race for cluster initialization. :param path: usually this is just '/initialize' :returns: `!True` if key has been created successfully. @@ -144,5 +145,9 @@ class AbstractDCS: """Voluntarily remove leader key from DCS This method should remove leader key if current instance is the leader""" + @abc.abstractmethod + def cancel_initialization(self): + """ Removes the initialize key for a cluster """ + def sleep(self, timeout): sleep(timeout) diff --git a/helpers/etcd.py b/helpers/etcd.py index add736c5..93be5a05 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -206,9 +206,13 @@ class Etcd(AbstractDCS): return ret @catch_etcd_errors - def race(self, path): - return self.client.write(self.client_path(path), self._name, prevExist=False) + def initialize(self): + return self.client.write(self.client_path(self.initialize_key), self._name, prevExist=False) @catch_etcd_errors def delete_leader(self): return self.client.delete(self.client_path('/leader'), prevValue=self._name) + + @catch_etcd_errors + def cancel_initialization(self): + return self.client.delete(self.client_path(self.initialize_key), prevValue=self._name) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index e451d844..4e425dec 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -391,3 +391,27 @@ recovery_target_timeline = 'latest' def last_operation(self): return str(self.xlog_position()) + + def bootstrap(self, 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. + """ + ret = False + if not current_leader: + ret = self.initialize() and self.start() + if ret: + self.create_replication_user() + self.create_connection_users() + else: + raise Exception("Could not bootstrap master PostgreSQL") + else: + if self.sync_from_leader(current_leader): + self.write_recovery_conf(current_leader) + ret = self.start() + return ret diff --git a/helpers/zookeeper.py b/helpers/zookeeper.py index cb2918cd..98d68ad1 100644 --- a/helpers/zookeeper.py +++ b/helpers/zookeeper.py @@ -180,8 +180,8 @@ class ZooKeeper(AbstractDCS): ret or logger.info('Could not take out TTL lock') return ret - def race(self, path): - return self._create(path, self._name, makepath=True) + def initialize(self): + return self._create(self.initialize_key, self._name, makepath=True) def touch_member(self, connection_string, ttl=None): for m in self.members: @@ -223,6 +223,11 @@ class ZooKeeper(AbstractDCS): if isinstance(self.leader, Member) and self.leader.name == self._name: self.client.delete(self.client_path('/leader')) + def cancel_initialization(self): + node = self.get_node(self.initialize_key) + if node and node == self._name: + self.client.delete(self.client_path(self.initialize_key)) + def sleep(self, timeout): self.cluster_event.wait(timeout) if self.cluster_event.isSet(): diff --git a/patroni.py b/patroni.py index 3be2aa57..64430318 100755 --- a/patroni.py +++ b/patroni.py @@ -43,6 +43,11 @@ class Patroni: return True return self.ha.dcs.touch_member(connection_string, ttl) + def cleanup_on_failed_initialization(self): + """ cleanup the DCS if initialization was not successfull """ + logger.info("removing initialize key after failed attempt to initialize the cluster") + self.ha.dcs.cancel_initialization() + def initialize(self): # wait for etcd to be available while not self.touch_member(): @@ -52,18 +57,18 @@ class Patroni: # is data directory empty? if self.postgresql.data_directory_empty(): # racing to initialize - if self.ha.dcs.race('/initialize'): - self.postgresql.initialize() + if self.ha.dcs.initialize(): + try: + self.postgresql.bootstrap() + except: + # bail out and clean the initialize flag. + self.cleanup_on_failed_initialization() + raise self.ha.dcs.take_leader() - self.postgresql.start() - self.postgresql.create_replication_user() - self.postgresql.create_connection_users() else: while True: leader = self.ha.dcs.current_leader() - if leader and self.postgresql.sync_from_leader(leader): - self.postgresql.write_recovery_conf(leader) - self.postgresql.start() + if leader and self.postgresql.bootstrap(leader): break sleep(5) elif self.postgresql.is_running(): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 38692c52..fa534d15 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -227,8 +227,8 @@ class TestEtcd(unittest.TestCase): def test_update_leader(self): self.assertTrue(self.etcd.update_leader(MockPostgresql())) - def test_race(self): - self.assertFalse(self.etcd.race('')) + def test_initialize(self): + self.assertFalse(self.etcd.initialize()) def test_delete_leader(self): self.etcd.client.delete = etcd_delete diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 67ac1ffd..eb41591c 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -46,6 +46,7 @@ class TestPatroni(unittest.TestCase): def set_up(self): self.touched = False + self.init_cancelled = False subprocess.call = subprocess_call psycopg2.connect = psycopg2_connect self.time_sleep = time.sleep @@ -121,14 +122,15 @@ class TestPatroni(unittest.TestCase): self.p.touch_member() def test_patroni_initialize(self): - self.p.postgresql.should_use_s3_to_create_replica = false self.p.ha.dcs.client.write = etcd_write self.p.touch_member = self.touch_member self.p.postgresql.data_directory_empty = true - self.p.ha.dcs.race = true + self.p.ha.dcs.initialize = true + self.p.postgresql.initialize = true + self.p.postgresql.start = true self.p.initialize() - self.p.ha.dcs.race = false + self.p.ha.dcs.initialize = false time.sleep = time_sleep self.p.ha.dcs.client.read = etcd_read self.p.initialize() @@ -142,3 +144,18 @@ class TestPatroni(unittest.TestCase): def test_schedule_next_run(self): self.p.next_run = time.time() - self.p.nap_time - 1 self.p.schedule_next_run() + + def cancel_initialization(self): + self.init_cancelled = True + + def test_cleanup_on_initialization(self): + self.p.ha.dcs.client.write = etcd_write + self.p.touch_member = self.touch_member + self.p.postgresql.data_directory_empty = true + self.p.ha.dcs.initialize = true + self.p.postgresql.initialize = true + self.p.postgresql.start = false + + self.p.ha.dcs.cancel_initialization = self.cancel_initialization + self.assertRaises(Exception, self.p.initialize) + self.assertTrue(self.init_cancelled) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 53f5ab82..28b337a9 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -141,7 +141,7 @@ class TestZooKeeper(unittest.TestCase): self.zk.delete_leader() def test_race(self): - self.assertFalse(self.zk.race('/initialize')) + self.assertFalse(self.zk.initialize()) def test_touch_member(self): self.zk.touch_member('new')