Merge pull request #322 from harlowja/better-cleanup

Do cleanup in a more more robust manner

Author: bbangert

kazoo/recipe/lock.py

@@ -547,7 +547,6 @@ def release(self):
     def _inner_release(self):
         if not self.is_acquired:
             return False
-
         try:
             self.client.delete(self.create_path)
         except NoNodeError:  # pragma: nocover

kazoo/testing/common.py

@@ -21,6 +21,7 @@
 
 
 import code
+import logging
 import os
 import os.path
 import shutil
@@ -34,6 +35,9 @@
 from glob import glob
 
 
+log = logging.getLogger(__name__)
+
+
 def debug(sig, frame):
     """Interrupt running process, and provide a python prompt for
     interactive debugging."""
@@ -145,22 +149,26 @@ def run(self):
 log4j.appender.ROLLINGFILE.File=""" + to_java_compatible_path(  # NOQA
                 self.working_path + os.sep + "zookeeper.log\n"))
 
-        self.process = subprocess.Popen(
-            args=["java",
-                  "-cp", self.classpath,
-
-                  # "-Dlog4j.debug",
-                  "-Dreadonlymode.enabled=true",
-                  "-Dzookeeper.log.dir=%s" % log_path,
-                  "-Dzookeeper.root.logger=INFO,CONSOLE",
-                  "-Dlog4j.configuration=file:%s" % log4j_path,
-
-                  # OS X: Prevent java from appearing in menu bar, process dock
-                  # and from activation of the main workspace on run.
-                  "-Djava.awt.headless=true",
-
-                  "org.apache.zookeeper.server.quorum.QuorumPeerMain",
-                  config_path])
+        args = [
+            "java",
+            "-cp", self.classpath,
+
+            # "-Dlog4j.debug",
+            "-Dreadonlymode.enabled=true",
+            "-Dzookeeper.log.dir=%s" % log_path,
+            "-Dzookeeper.root.logger=INFO,CONSOLE",
+            "-Dlog4j.configuration=file:%s" % log4j_path,
+
+            # OS X: Prevent java from appearing in menu bar, process dock
+            # and from activation of the main workspace on run.
+            "-Djava.awt.headless=true",
+
+            "org.apache.zookeeper.server.quorum.QuorumPeerMain",
+            config_path,
+        ]
+        self.process = subprocess.Popen(args=args)
+        log.info("Started zookeeper process %s using args %s",
+                 self.process.pid, args)
         self._running = True
 
     @property
@@ -226,6 +234,11 @@ def stop(self):
             return
         self.process.terminate()
         self.process.wait()
+        if self.process.returncode != 0:
+            log.warn("Zookeeper process %s failed to terminate with"
+                     " non-zero return code (it terminated with %s return"
+                     " code instead)", self.process.pid,
+                     self.process.returncode)
         self._running = False
 
     def destroy(self):

kazoo/testing/harness.py

@@ -8,7 +8,7 @@
 from kazoo import python2atexit as atexit
 
 from kazoo.client import KazooClient
-from kazoo.exceptions import NotEmptyError
+from kazoo.exceptions import KazooException, NotEmptyError
 from kazoo.protocol.states import (
     KazooState
 )
@@ -81,14 +81,13 @@ def servers(self):
         return ",".join([s.address for s in self.cluster])
 
     def _get_nonchroot_client(self):
-        return KazooClient(self.servers)
+        c = KazooClient(self.servers)
+        self._clients.append(c)
+        return c
 
     def _get_client(self, **kwargs):
         c = KazooClient(self.hosts, **kwargs)
-        try:
-            self._clients.append(c)
-        except AttributeError:
-            self._client = [c]
+        self._clients.append(c)
         return c
 
     def lose_connection(self, event_factory):
@@ -105,47 +104,31 @@ def setup_zookeeper(self, **client_options):
         The cluster will only be created on the first invocation and won't be
         fully torn down until exit.
         """
-        if not self.cluster[0].running:
+        do_start = False
+        for s in self.cluster:
+            if not s.running:
+                do_start = True
+        if do_start:
             self.cluster.start()
         namespace = "/kazootests" + uuid.uuid4().hex
         self.hosts = self.servers + namespace
-
         if 'timeout' not in client_options:
             client_options['timeout'] = 0.8
         self.client = self._get_client(**client_options)
         self.client.start()
         self.client.ensure_path("/")
 
     def teardown_zookeeper(self):
-        """Clean up any ZNodes created during the test
-        """
-        if not self.cluster[0].running:
-            self.cluster.start()
-
-        tries = 0
-        if self.client and self.client.connected:
-            while tries < 3:
-                try:
-                    self.client.retry(self.client.delete, '/', recursive=True)
-                    break
-                except NotEmptyError:
-                    pass
-                tries += 1
-            self.client.stop()
-            self.client.close()
-            del self.client
-        else:
-            client = self._get_client()
-            client.start()
-            client.retry(client.delete, '/', recursive=True)
-            client.stop()
-            client.close()
-            del client
-
-        for client in self._clients:
-            client.stop()
-            del client
-        self._clients = None
+        """Reset and cleanup the zookeeper cluster that was started."""
+        while self._clients:
+            c = self._clients.pop()
+            try:
+                c.stop()
+            except KazooException:
+                log.exception("Failed stopping client %s", c)
+            finally:
+                c.close()
+        self.client = None
 
     def __break_connection(self, break_event, expected_state, event_factory):
         """Break ZooKeeper connection using the specified event."""

kazoo/tests/test_lock.py

@@ -37,6 +37,16 @@ def wait(self):
 class KazooLockTests(KazooTestCase):
     thread_count = 20
 
+    def __init__(self, *args, **kw):
+        super(KazooLockTests, self).__init__(*args, **kw)
+        self.threads_made = []
+
+    def tearDown(self):
+        super(KazooLockTests, self).tearDown()
+        while self.threads_made:
+            t = self.threads_made.pop()
+            t.join()
+
     @staticmethod
     def make_condition():
         return threading.Condition()
@@ -45,9 +55,11 @@ def make_condition():
     def make_event():
         return threading.Event()
 
-    @staticmethod
-    def make_thread(*args, **kwargs):
-        return threading.Thread(*args, **kwargs)
+    def make_thread(self, *args, **kwargs):
+        t = threading.Thread(*args, **kwargs)
+        t.daemon = True
+        self.threads_made.append(t)
+        return t
 
     @staticmethod
     def make_wait():
@@ -389,6 +401,16 @@ def _thread(lock, event, timeout):
 
 class TestSemaphore(KazooTestCase):
 
+    def __init__(self, *args, **kw):
+        super(TestSemaphore, self).__init__(*args, **kw)
+        self.threads_made = []
+
+    def tearDown(self):
+        super(TestSemaphore, self).tearDown()
+        while self.threads_made:
+            t = self.threads_made.pop()
+            t.join()
+
     @staticmethod
     def make_condition():
         return threading.Condition()
@@ -397,9 +419,11 @@ def make_condition():
     def make_event():
         return threading.Event()
 
-    @staticmethod
-    def make_thread(*args, **kwargs):
-        return threading.Thread(*args, **kwargs)
+    def make_thread(self, *args, **kwargs):
+        t = threading.Thread(*args, **kwargs)
+        t.daemon = True
+        self.threads_made.append(t)
+        return t
 
     def setUp(self):
         super(TestSemaphore, self).setUp()
@@ -539,8 +563,8 @@ def sema_one():
                 event.set()
                 event2.wait()
 
-        thread = self.make_thread(target=sema_one, args=())
-        thread.start()
+        thread1 = self.make_thread(target=sema_one, args=())
+        thread1.start()
 
         started.wait()
         eq_(lh_semaphore.lease_holders(), ['george'])
@@ -552,8 +576,8 @@ def expire():
             self.expire_session(self.make_event)
             expired.set()
 
-        thread = self.make_thread(target=expire, args=())
-        thread.start()
+        thread2 = self.make_thread(target=expire, args=())
+        thread2.start()
         expire_semaphore.wake_event.wait()
         expired.wait()
 
@@ -563,7 +587,9 @@ def expire():
         event.wait(15)
         eq_(expire_semaphore.lease_holders(), ['fred'])
         event2.set()
-        thread.join()
+
+        for t in (thread1, thread2):
+            t.join()
 
     def test_inconsistent_max_leases(self):
         sem1 = self.client.Semaphore(self.lockpath, max_leases=1)