Merge pull request #30 from zalando/feature/cleanup_on_failed_initialization

Make sure initialize flag is reset on failure.

patroni/__init__.py

@@ -6,6 +6,7 @@ import yaml
 
 from patroni.api import RestApiServer
 from patroni.etcd import Etcd
+from patroni.exceptions import DCSError
 from patroni.ha import Ha
 from patroni.postgresql import Postgresql
 from patroni.utils import setup_signal_handlers, sleep, reap_children
@@ -42,6 +43,14 @@ 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()
+        self.touch_member(self.shutdown_member_ttl)
+        self.postgresql.stop()
+        self.postgresql.move_data_directory()
+
     def initialize(self):
         # wait for etcd to be available
         while not self.touch_member():
@@ -50,21 +59,27 @@ class Patroni:
 
         # is data directory empty?
         if self.postgresql.data_directory_empty():
-            # racing to initialize
+            while True:
-            if self.ha.dcs.race('/initialize'):
+                try:
-                self.postgresql.initialize()
+                    cluster = self.ha.dcs.get_cluster()
-                self.ha.dcs.take_leader()
+                    if not cluster.is_unlocked():  # the leader already exists
-                self.postgresql.start()
+                        if not cluster.initialize:
-                self.postgresql.create_replication_user()
+                            self.ha.dcs.initialize()
-                self.postgresql.create_connection_users()
+                        self.postgresql.bootstrap(cluster.leader)
-            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()
                         break
-                    sleep(5)
+                    # racing to initialize
+                    elif not cluster.initialize and 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()
+                        break
+                except DCSError:
+                    logger.info('waiting on DCS')
+                sleep(5)
         elif self.postgresql.is_running():
             self.postgresql.load_replication_slots()
 
@@ -110,8 +125,8 @@ def main():
         config = yaml.load(f)
 
     patroni = Patroni(config)
+    patroni.initialize()
     try:
-        patroni.initialize()
         patroni.run()
     except KeyboardInterrupt:
         pass

patroni/dcs.py

@@ -74,6 +74,12 @@ class AbstractDCS:
 
     __metaclass__ = abc.ABCMeta
 
+    _INITIALIZE = 'initialize'
+    _LEADER = 'leader'
+    _MEMBERS = 'members/'
+    _OPTIME = 'optime'
+    _LEADER_OPTIME = _OPTIME + '/' + _LEADER
+
     def __init__(self, name, config):
         """
         :param name: name of current instance (the same value as `~Postgresql.name`)
@@ -85,7 +91,27 @@ class AbstractDCS:
         self._base_path = '/service/' + self._scope
 
     def client_path(self, path):
-        return self._base_path + path
+        return '/'.join([self._base_path, path.lstrip('/')])
+
+    @property
+    def initialize_path(self):
+        return self.client_path(self._INITIALIZE)
+
+    @property
+    def members_path(self):
+        return self.client_path(self._MEMBERS)
+
+    @property
+    def member_path(self):
+        return self.client_path(self._MEMBERS + self._name)
+
+    @property
+    def leader_path(self):
+        return self.client_path(self._LEADER)
+
+    @property
+    def leader_optime_path(self):
+        return self.client_path(self._LEADER_OPTIME)
 
     @abc.abstractmethod
     def get_cluster(self):
@@ -139,12 +165,11 @@ 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.
 
-        this method should create atomically `path` key and return `!True`
+        this method should create atomically initialize key and return `!True`
         otherwise it should return `!False`"""
 
     @abc.abstractmethod
@@ -152,5 +177,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 watch(self, timeout):
         sleep(timeout)

patroni/etcd.py

@@ -179,17 +179,17 @@ class Etcd(AbstractDCS):
             nodes = {os.path.relpath(node.key, result.key): node for node in result.leaves}
 
             # get initialize flag
-            initialize = bool(nodes.get('initialize', False))
+            initialize = bool(nodes.get(self._INITIALIZE, False))
 
             # get last leader operation
-            last_leader_operation = nodes.get('optime/leader', None)
+            last_leader_operation = nodes.get(self._LEADER_OPTIME, None)
             last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation.value)
 
             # get list of members
-            members = [self.member(n) for k, n in nodes.items() if k.startswith('members/') and len(k.split('/')) == 2]
+            members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
 
             # get leader
-            leader = nodes.get('leader', None)
+            leader = nodes.get(self._LEADER, None)
             if leader:
                 member = Member(-1, leader.value, None, None, None, None)
                 member = ([m for m in members if m.name == leader.value] or [member])[0]
@@ -206,17 +206,15 @@ class Etcd(AbstractDCS):
 
     @catch_etcd_errors
     def touch_member(self, connection_string, ttl=None):
-        return self.retry(self.client.set, self.client_path('/members/' + self._name),
+        return self.retry(self.client.set, self.member_path, connection_string, ttl or self.member_ttl)
-                          connection_string, ttl or self.member_ttl)
 
     @catch_etcd_errors
     def take_leader(self):
-        return self.retry(self.client.set, self.client_path('/leader'), self._name, self.ttl)
+        return self.retry(self.client.set, self.leader_path, self._name, self.ttl)
 
     def attempt_to_acquire_leader(self):
         try:
-            return not self.retry(self.client.write, self.client_path('/leader'),
+            return bool(self.retry(self.client.write, self.leader_path, self._name, ttl=self.ttl, prevExist=False))
-                                  self._name, ttl=self.ttl, prevExist=False) is None
         except etcd.EtcdAlreadyExist:
             logger.info('Could not take out TTL lock')
         except (RetryFailedError, etcd.EtcdException):
@@ -225,31 +223,34 @@ class Etcd(AbstractDCS):
 
     @catch_etcd_errors
     def write_leader_optime(self, state_handler):
-        return self.client.set(self.client_path('/optime/leader'), state_handler.last_operation())
+        return self.client.set(self.leader_optime_path, state_handler.last_operation())
 
     @catch_etcd_errors
     def update_leader(self, state_handler):
-        ret = self.retry(self.client.test_and_set, self.client_path('/leader'), self._name, self._name, self.ttl)
+        ret = self.retry(self.client.test_and_set, self.leader_path, self._name, self._name, self.ttl)
         ret and self.write_leader_optime(state_handler)
         return ret
 
     @catch_etcd_errors
-    def race(self, path):
+    def initialize(self):
-        return self.retry(self.client.write, self.client_path(path), self._name, prevExist=False)
+        return self.client.write(self.initialize_path, self._name, prevExist=False)
 
     @catch_etcd_errors
     def delete_leader(self):
-        return self.client.delete(self.client_path('/leader'), prevValue=self._name)
+        return self.client.delete(self.leader_path, prevValue=self._name)
+
+    @catch_etcd_errors
+    def cancel_initialization(self):
+        return self.client.delete(self.initialize_path, prevValue=self._name)
 
     def watch(self, timeout):
         # watch on leader key changes if it is defined and current node is not lock owner
         if self.cluster and self.cluster.leader and self.cluster.leader.name != self._name:
             end_time = time.time() + timeout
             index = self.cluster.leader.index
-
             while index and timeout >= 1:  # when timeout is too small urllib3 doesn't have enough time to connect
                 try:
-                    res = self.client.watch(self.client_path('/leader'), index=index + 1, timeout=timeout)
+                    res = self.client.watch(self.leader_path, index=index + 1, timeout=timeout)
                     if res.action not in ['set', 'compareAndSwap'] or res.value != self.cluster.leader.name:
                         return
                     index = res.modifiedIndex

patroni/exceptions.py

@@ -13,5 +13,9 @@ class PatroniException(Exception):
         return repr(self.value)
 
 
+class PostgresException(PatroniException):
+    pass
+
+
 class DCSError(PatroniException):
     pass

patroni/postgresql.py

@@ -4,7 +4,9 @@ import psycopg2
 import shlex
 import shutil
 import subprocess
+import time
 
+from patroni.exceptions import PostgresException
 from patroni.utils import sleep
 from six.moves.urllib_parse import urlparse
 
@@ -159,7 +161,7 @@ class Postgresql:
         return ret
 
     def is_running(self):
-        return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null', shell=True) == 0
+        return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null 2>&1', shell=True) == 0
 
     def call_nowait(self, cb_name, is_leader=None):
         """ pick a callback command and call it without waiting for it to finish """
@@ -391,3 +393,34 @@ 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 PostgresException("Could not bootstrap master PostgreSQL")
+        else:
+            if self.sync_from_leader(current_leader):
+                self.write_recovery_conf(current_leader)
+                ret = self.start()
+        return ret
+
+    def move_data_directory(self):
+        if os.path.isdir(self.data_dir) and not self.is_running():
+            try:
+                os.rename(self.data_dir, '{0}_{1}'.format(self.data_dir, time.strftime('%Y-%m-%d-%H-%M-%S')))
+            except:
+                logger.exception("Could not rename data directory {0}".format(self.data_dir))

patroni/zookeeper.py

@@ -92,9 +92,8 @@ class ZooKeeper(AbstractDCS):
         self.client.add_listener(self.session_listener)
         self.cluster_event = self.client.handler.event_object()
 
+        self.cluster = None
         self.fetch_cluster = True
-        self.members = []
-        self.leader = None
         self.last_leader_operation = 0
 
         self.client.start(None)
@@ -107,50 +106,60 @@ class ZooKeeper(AbstractDCS):
         self.fetch_cluster = True
         self.cluster_event.set()
 
-    def get_node(self, name, watch=None):
+    def get_node(self, key, watch=None):
         try:
-            return self.client.get(self.client_path(name), watch)
+            return self.client.get(key, watch)
         except NoNodeError:
-            pass
+            return None
-        except:
-            logger.exception('get_node')
-        return None
 
     @staticmethod
     def member(name, value, znode):
         conn_url, api_url = parse_connection_string(value)
         return Member(znode.mzxid, name, conn_url, api_url, None, None)
 
+    def get_children(self, key, watch=None):
+        try:
+            return self.client.get_children(key, watch)
+        except NoNodeError:
+            return []
+
     def load_members(self):
         members = []
-        for member in self.client.get_children(self.client_path('/members'), self.cluster_watcher):
+        for member in self.get_children(self.members_path, self.cluster_watcher):
-            data = self.get_node('/members/' + member)
+            data = self.get_node(self.members_path + member)
             if data is not None:
                 members.append(self.member(member, *data))
         return members
 
     def _inner_load_cluster(self):
         self.cluster_event.clear()
-        leader = self.get_node('/leader', self.cluster_watcher)
+        nodes = set(self.get_children(self.client_path('')))
-        self.members = self.load_members()
+
+        # get initialize flag
+        initialize = self._INITIALIZE in nodes
+
+        # get list of members
+        members = self.load_members() if self._MEMBERS[:-1] in nodes else []
+
+        # get leader
+        leader = self.get_node(self.leader_path, self.cluster_watcher) if self._LEADER in nodes else None
         if leader:
             client_id = self.client.client_id
             if leader[0] == self._name and client_id is not None and client_id[0] != leader[1].ephemeralOwner:
                 logger.info('I am leader but not owner of the session. Removing leader node')
-                self.client.delete(self.client_path('/leader'))
+                self.client.delete(self.leader_path)
                 leader = None
 
             if leader:
                 member = Member(-1, leader[0], None, None, None, None)
-                member = ([m for m in self.members if m.name == leader[0]] or [member])[0]
+                member = ([m for m in members if m.name == leader[0]] or [member])[0]
                 leader = Leader(leader[1].mzxid, None, None, member)
                 self.fetch_cluster = member.index == -1
 
-        self.leader = leader
+        # get last leader operation
-        if self.fetch_cluster:
+        self.last_leader_operation = self.get_node(self.leader_optime_path) if self.fetch_cluster else None
-            last_leader_operation = self.get_node('/optime/leader')
+        self.last_leader_operation = 0 if self.last_leader_operation is None else int(self.last_leader_operation[0])
-            if last_leader_operation:
+        self.cluster = Cluster(initialize, leader, self.last_leader_operation, members)
-                self.last_leader_operation = int(last_leader_operation[0])
 
     def get_cluster(self):
         if self.exhibitor and self.exhibitor.poll():
@@ -163,28 +172,27 @@ class ZooKeeper(AbstractDCS):
                 logger.exception('get_cluster')
                 self.session_listener(KazooState.LOST)
                 raise ZooKeeperError('ZooKeeper in not responding properly')
-        return Cluster(True, self.leader, self.last_leader_operation, self.members)
+        return self.cluster
 
     def _create(self, path, value, **kwargs):
         try:
-            self.client.retry(self.client.create, self.client_path(path), value, **kwargs)
+            self.client.retry(self.client.create, path, value, **kwargs)
             return True
         except:
             return False
 
     def attempt_to_acquire_leader(self):
-        ret = self._create('/leader', self._name, makepath=True, ephemeral=True)
+        ret = self._create(self.leader_path, self._name, makepath=True, ephemeral=True)
         ret or logger.info('Could not take out TTL lock')
         return ret
 
-    def race(self, path):
+    def initialize(self):
-        return self._create(path, self._name, makepath=True)
+        return self._create(self.initialize_path, self._name, makepath=True)
 
     def touch_member(self, connection_string, ttl=None):
-        for m in self.members:
+        if self.cluster and any(m.name == self._name for m in self.cluster.members):
-            if m.name == self._name:
+            return True
-                return True
+        path = self.member_path
-        path = self.client_path('/members/' + self._name)
         try:
             self.client.retry(self.client.create, path, connection_string, makepath=True, ephemeral=True)
             return True
@@ -204,7 +212,7 @@ class ZooKeeper(AbstractDCS):
         last_operation = state_handler.last_operation()
         if last_operation != self.last_leader_operation:
             self.last_leader_operation = last_operation
-            path = self.client_path('/optime/leader')
+            path = self.leader_optime_path
             try:
                 self.client.retry(self.client.set, path, last_operation)
             except NoNodeError:
@@ -217,8 +225,19 @@ class ZooKeeper(AbstractDCS):
         return True
 
     def delete_leader(self):
-        if isinstance(self.leader, Leader) and self.leader.name == self._name:
+        if isinstance(self.cluster, Cluster) and self.cluster.leader.name == self._name:
-            self.client.delete(self.client_path('/leader'))
+            self.client.delete(self.leader_path, version=self.cluster.leader.index)
+
+    def _cancel_initialization(self):
+        node = self.get_node(self.initialize_path)
+        if node and node[0] == self._name:
+            self.client.delete(self.initialize_path, version=node[1].mzxid)
+
+    def cancel_initialization(self):
+        try:
+            self.client.retry(self._cancel_initialization)
+        except:
+            logger.exception("Unable to delete initialize key")
 
     def watch(self, timeout):
         self.cluster_event.wait(timeout)

postgres0.yml

@@ -47,7 +47,7 @@ postgresql:
     env_dir: /home/postgres/etc/wal-e.d/env
     threshold_megabytes: 10240
     threshold_backup_size_percentage: 30
-  restore: scripts/restore.py
+  restore: patroni/scripts/restore.py
   #recovery_conf:
     #restore_command: cp ../wal_archive/%f %p
   parameters:

postgres1.yml

@@ -49,7 +49,7 @@ postgresql:
     env_dir: /home/postgres/etc/wal-e.d/env
     threshold_megabytes: 10240
     threshold_backup_size_percentage: 30
-  restore: scripts/restore.py
+  restore: patroni/scripts/restore.py
   parameters:
     archive_mode: "on"
     wal_level: hot_standby

release.sh

@@ -27,5 +27,5 @@ git push
 
 python3 setup.py sdist bdist_wheel upload
 
-git tag ${version}
+git tag v${version}
 git push --tags

tests/test_etcd.py

@@ -93,9 +93,9 @@ def etcd_delete(key, **kwargs):
 
 
 def etcd_read(key, **kwargs):
-    if key == '/service/noleader':
+    if key == '/service/noleader/':
         raise DCSError('noleader')
-    elif key == '/service/nocluster':
+    elif key == '/service/nocluster/':
         raise etcd.EtcdKeyNotFound
 
     response = {"action": "get", "node": {"key": "/service/batman5", "dir": True, "nodes": [
@@ -266,8 +266,12 @@ class TestEtcd(unittest.TestCase):
     def test_update_leader(self):
         self.assertTrue(self.etcd.update_leader(MockPostgresql()))
 
-    def test_race(self):
+    def test_initialize(self):
-        self.assertFalse(self.etcd.race(''))
+        self.assertFalse(self.etcd.initialize())
+
+    def test_cancel_initializion(self):
+        self.etcd.client.delete = etcd_delete
+        self.assertFalse(self.etcd.cancel_initialization())
 
     def test_delete_leader(self):
         self.etcd.client.delete = etcd_delete

tests/test_patroni.py

@@ -9,8 +9,9 @@ import yaml
 
 from mock import Mock, patch
 from patroni.api import RestApiServer
-from patroni.dcs import Cluster, Member
+from patroni.dcs import Cluster, Member, Leader
 from patroni.etcd import Etcd
+from patroni.exceptions import PostgresException
 from patroni import Patroni, main
 from patroni.zookeeper import ZooKeeper
 from six.moves import BaseHTTPServer
@@ -41,6 +42,30 @@ class Mock_BaseServer__is_shut_down:
         pass
 
 
+def get_cluster(initialize, leader):
+    return Cluster(initialize, leader, None, None)
+
+
+def get_cluster_not_initialized_without_leader():
+    return get_cluster(None, None)
+
+
+def get_cluster_initialized_without_leader():
+    return get_cluster(True, None)
+
+
+def get_cluster_not_initialized_with_leader():
+    return get_cluster(False, Leader(0, 0, 0,
+                       Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres',
+                              None, None, 28)))
+
+
+def get_cluster_initialized_with_leader():
+    return get_cluster(True, Leader(0, 0, 0, 
+                       Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres',
+                              None, None, 28)))
+
+
 class TestPatroni(unittest.TestCase):
 
     def __init__(self, method_name='runTest'):
@@ -50,6 +75,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
@@ -126,24 +152,49 @@ 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.ha.dcs.client.read = etcd_read
         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.ha.dcs.get_cluster = get_cluster_not_initialized_without_leader
         self.p.initialize()
 
-        self.p.ha.dcs.race = false
+        self.p.ha.dcs.initialize = false
+        self.p.ha.dcs.get_cluster = get_cluster_initialized_with_leader
         time.sleep = time_sleep
         self.p.ha.dcs.client.read = etcd_read
         self.p.initialize()
 
-        self.p.ha.dcs.current_leader = nop
+        self.p.ha.dcs.get_cluster = get_cluster_initialized_without_leader
-        self.assertRaises(Exception, self.p.initialize)
+        self.assertRaises(SleepException, self.p.initialize)
 
         self.p.postgresql.data_directory_empty = false
         self.p.initialize()
 
+        self.p.ha.dcs.get_cluster = get_cluster_not_initialized_with_leader
+        self.p.postgresql.data_directory_empty = true
+        self.p.initialize()
+
     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.ha.dcs.client.read = etcd_read
+        self.p.ha.dcs.get_cluster = get_cluster_not_initialized_without_leader
+        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(PostgresException, self.p.initialize)
+        self.assertTrue(self.init_cancelled)

tests/test_postgresql.py

@@ -6,6 +6,7 @@ import unittest
 
 from patroni.dcs import Cluster, Leader, Member
 from patroni.postgresql import Postgresql
+from test_ha import true, false
 
 
 def nop(*args, **kwargs):
@@ -217,3 +218,9 @@ class TestPostgresql(unittest.TestCase):
         self.p.start()
         self.p.query = self.mock_query
         self.assertTrue(self.p.stop())
+
+    def test_move_data_directory(self):
+        self.p.is_running = is_running
+        os.rename = nop
+        os.path.isdir = true
+        self.p.move_data_directory()

tests/test_zookeeper.py

@@ -56,10 +56,8 @@ class MockKazooClient:
         func(*args, **kwargs)
 
     def get(self, path, watch=None):
-        if path == '/service/test/no_node':
+        if path == '/no_node':
             raise NoNodeError
-        elif path == '/service/test/other_exception':
-            raise Exception()
         elif '/members/' in path:
             return (
                 'postgres://repuser:rep-pass@localhost:5434/postgres?application_name=http://127.0.0.1:8009/patroni',
@@ -71,8 +69,14 @@ class MockKazooClient:
             if self.leader:
                 return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0))
             return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
+        elif path.endswith('/initialize'):
+            return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
 
     def get_children(self, path, watch=None, include_data=False):
+        if path == '/no_node':
+            raise NoNodeError
+        elif path in ['/service/bla/', '/service/test/']:
+            return ['initialize', 'leader', 'members', 'optime']
         return ['foo', 'bar', 'buzz']
 
     def create(self, path, value="", acl=None, ephemeral=False, sequence=False, makepath=False):
@@ -93,6 +97,8 @@ class MockKazooClient:
                 return
             self.leader = True
             raise Exception
+        elif path.endswith('/initialize'):
+            raise NoNodeError
 
     def set_hosts(self, hosts, randomize_hosts=None):
         pass
@@ -132,7 +138,9 @@ class TestZooKeeper(unittest.TestCase):
 
     def test_get_node(self):
         self.assertIsNone(self.zk.get_node('/no_node'))
-        self.assertIsNone(self.zk.get_node('/other_exception'))
+
+    def test_get_children(self):
+        self.assertListEqual(self.zk.get_children('/no_node'), [])
 
     def test__inner_load_cluster(self):
         self.zk._base_path = self.zk._base_path.replace('test', 'bla')
@@ -146,8 +154,11 @@ class TestZooKeeper(unittest.TestCase):
         self.zk.touch_member('foo')
         self.zk.delete_leader()
 
-    def test_race(self):
+    def test_initialize(self):
-        self.assertFalse(self.zk.race('/initialize'))
+        self.assertFalse(self.zk.initialize())
+
+    def test_cancel_initialization(self):
+        self.zk.cancel_initialization()
 
     def test_touch_member(self):
         self.zk.touch_member('new')