Merge pull request #292 from TomasTomecek/race-cond-creatdkr-conts

fix race condition when starting dkr conts via binary

Makefile

@@ -16,8 +16,8 @@ install-test-requirements:
 exec-test:
 	cat pytest.ini
 	@# use it like this: `make exec-test TEST_TARGET="tests/unit/"`
-	PYTHONPATH=$(CURDIR) pytest-2 $(TEST_TARGET)
-	PYTHONPATH=$(CURDIR) pytest-3 $(TEST_TARGET)
+	PYTHONPATH=$(CURDIR) pytest-2 $(TEST_TARGET) --verbose --showlocals
+	PYTHONPATH=$(CURDIR) pytest-3 $(TEST_TARGET) --verbose --showlocals
 
 check: test
 

conu/backend/docker/image.py

@@ -277,11 +277,21 @@ def _run_container(self, run_command_instance, callback):
         run_command_instance.options += ["--cidfile=%s" % tmpfile]
         logger.debug("docker command: %s" % run_command_instance)
         response = callback()
-        # and we need to wait now; inotify would be better but is way more complicated and
-        # adds dependency
-        Probe(timeout=10, count=10, pause=0.1, fnc=lambda: os.path.exists(tmpfile)).run()
+
+        def get_cont_id():
+            if not os.path.exists(tmpfile):
+                return False
+            with open(tmpfile, 'r') as fd:
+                content = fd.read()
+            return bool(content)
+
+        Probe(timeout=2, count=10, pause=0.1, fnc=get_cont_id).run()
+
         with open(tmpfile, 'r') as fd:
             container_id = fd.read()
+
+        if not container_id:
+            raise ConuException("We could not get container's ID, it probably was not created")
         return container_id, response
 
     def run_via_binary(self, run_command_instance=None, command=None, volumes=None,

conu/utils/probes.py

@@ -42,9 +42,10 @@ def __init__(self,
         :param timeout:              Number of seconds spent on trying. Set timeout to -1 for infinite run.
         :param pause:                Number of seconds waited between multiple function result checks
         :param count:                Maximum number of tries, defaults to infinite, represented by -1
-        :param expected_exceptions:  When one of expected_exception is raised, probe ignores it and tries to run function again.
-                                         To ignore multiple exceptions use parenthesized tuple.
-        :param expected_retval:      When expected_retval is recieved, probe ends successfully
+        :param expected_exceptions:  When one of expected_exception is raised, probe ignores it and
+                                     tries to run function again. To ignore multiple exceptions use
+                                     parenthesized tuple.
+        :param expected_retval:      When expected_retval is received, probe ends successfully
         :param fnc:                  Function which run is checked by probe
         """
         self.timeout = timeout
@@ -103,11 +104,16 @@ def _wrapper(self, q, start):
         logger.debug("Running \"%s\" with parameters: \"%s\":\t%s/%s"
                      % (func_name, str(self.kwargs), round(time.time() - start), self.timeout))
         try:
-            q.put(self.fnc(**self.kwargs))
-        except self.expected_exceptions:
+            result = self.fnc(**self.kwargs)
+            # let's log only first 50 characters of the response
+            logger.debug("callback result = %s", str(result)[:50])
+            q.put(result)
+        except self.expected_exceptions as ex:
+            logger.debug("expected exception was caught: %s", ex)
             q.put(False)
-        except Exception as e:
-            q.put(e)
+        except Exception as ex:
+            logger.debug("adding exception %s to queue", ex)
+            q.put(ex)
 
     def _run(self):
         start = time.time()
@@ -117,44 +123,47 @@ def _run(self):
         p.start()
         logger.debug("first process started: pid=%s", p.pid)
         tries = 1
-        while (tries <= self.count or self.count == -1) and \
-                (self.timeout == -1 or time.time() - start <= self.timeout):
+        while tries <= self.count or self.count == -1:
+            elapsed = time.time() - start
+            if self.timeout != -1 and elapsed > self.timeout:
+                logger.info("timeout was reached, elapsed: %s", elapsed)
+                break
             if p.is_alive():
                 logger.debug("pausing for %s before next try", self.pause)
                 time.sleep(self.pause)
-            elif not fnc_queue.empty():
+            else:
+                logger.debug("waiting for process to end...")
+                p.join()
+                if fnc_queue.empty():
+                    raise RuntimeError("queue is empty when it shouldn't be")
                 result = fnc_queue.get()
+                logger.debug("result = %s", result)
                 if isinstance(result, Exception):
                     # TODO: use result's traceback
                     if self.queue:
                         self.queue.put(result)
+                        return False
                     else:
                         raise result
                 elif not (result == self.expected_retval):
-                    logger.debug("result = %s", result)
-                    logger.debug("waiting for process to end...")
-                    p.join()
                     logger.debug("process ended, about to start another one")
                     p = Process(target=self._wrapper, args=(fnc_queue, start))
                     p.start()
                     tries += 1
                     logger.debug("attempt no. %s started, pid: %s", tries, p.pid)
                 else:
                     return True
-            else:
-                return True
+        p.terminate()
+        p.join()
+        if -1 < self.count < tries:
+            e = CountExceeded()
         else:
-            p.terminate()
-            p.join()
-            if -1 < self.count < tries:
-                e = CountExceeded
-            else:
-                e = ProbeTimeout("Timeout exceeded.")
-            logger.warning("probe is unsuccessful: %s", e)
-            if self.queue:
-                self.queue.put(e)
-            else:
-                raise e
+            e = ProbeTimeout("Timeout exceeded.")
+        logger.warning("probe is unsuccessful: %s", e)
+        if self.queue:
+            self.queue.put(e)
+        else:
+            raise e
 
 
 class ProbeTimeout(ConuException):

tests/integration/test_backend.py

@@ -25,7 +25,7 @@
 from conu.fixtures import docker_backend
 
 
-# Remove xfail once https://github.com/user-cont/conu/issues/262 is fixed.
+# FIXME: Remove xfail once https://github.com/user-cont/conu/issues/262 is fixed.
 @pytest.mark.xfail(raises=AssertionError)
 def test_cleanup_containers():
     with DockerBackend(logging_level=logging.DEBUG, cleanup=[CleanupPolicy.CONTAINERS]) as backend:

tests/integration/test_docker.py

@@ -128,7 +128,7 @@ def test_copy_from(tmpdir):
 
 
 def test_container_create_failed():
-    with DockerBackend() as backend:
+    with DockerBackend(logging_level=10) as backend:
         image = backend.ImageClass(FEDORA_MINIMAL_REPOSITORY, tag=FEDORA_MINIMAL_REPOSITORY_TAG)
         # should raise an exc, there is no such command: waldo; we need to find waldo first
         with pytest.raises(ConuException):

tests/unit/test_probe.py

@@ -13,12 +13,15 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
+import logging
 import time
 
 from conu import Probe, ProbeTimeout, CountExceeded
+from conu.apidefs.backend import set_logging
 
 import pytest
 
+
 ARGUMENT = "key"
 MESSAGE = "It is both alive and dead"
 
@@ -29,12 +32,16 @@ def snoozer(seconds=1):
 
 
 def value_err_raise():
-    raise ValueError
+    raise ValueError()
 
 
 class TestProbe(object):
+    @classmethod
+    def setup_class(cls):
+        set_logging(level=logging.DEBUG)
+
     def test_in_backgroud(self):
-        probe = Probe(timeout=5, pause=0.5, fnc=snoozer, seconds=2)
+        probe = Probe(timeout=1, pause=0.5, fnc=snoozer, seconds=0.1)
 
         probe.run()
         assert not probe.is_alive()
@@ -49,48 +56,36 @@ def test_exception(self):
 
         # probe and caller in one thread
         # probe should ignore expected_exceptions
-        start = time.time()
-        probe = Probe(timeout=3, pause=0.5, expected_exceptions=ValueError, fnc=value_err_raise)
+        probe = Probe(timeout=1, pause=0.2, expected_exceptions=ValueError, fnc=value_err_raise)
         with pytest.raises(ProbeTimeout):
             probe.run()
-        assert (time.time() - start) > 2, "Timeout not reached with unsuccessful function"
 
         # probe should not ignore exceptions other than expected exceptions
-        start = time.time()
-        probe = Probe(timeout=5, pause=0.5, expected_exceptions=ImportError, fnc=value_err_raise)
+        probe = Probe(timeout=1, pause=0.2, expected_exceptions=ImportError, fnc=value_err_raise)
         with pytest.raises(ValueError):
             probe.run()
-        assert (time.time() - start) < 1, "Timeout exceeded"
 
-        start = time.time()
-        probe = Probe(timeout=5, pause=0.5, fnc=value_err_raise)
+        probe = Probe(timeout=1, pause=0.2, fnc=value_err_raise)
         with pytest.raises(ValueError):
             probe.run()
-        assert (time.time() - start) < 1, "Timeout exceeded"
 
         # run in background
         # probe should ignore expected_exceptions
-        start = time.time()
-        probe = Probe(timeout=3, pause=0.5, expected_exceptions=ValueError, fnc=value_err_raise)
+        probe = Probe(timeout=1, pause=0.2, expected_exceptions=ValueError, fnc=value_err_raise)
         probe.run_in_background()
         with pytest.raises(ProbeTimeout):
             probe.join()
-        assert (time.time() - start) > 2, "Timeout not reached with unsuccessful function"
 
         # probe should not ignore exceptions other than expected exceptions
-        start = time.time()
-        probe = Probe(timeout=5, pause=0.5, expected_exceptions=ImportError, fnc=value_err_raise)
+        probe = Probe(timeout=1, pause=0.2, expected_exceptions=ImportError, fnc=value_err_raise)
         probe.run_in_background()
         with pytest.raises(ValueError):
             probe.join()
-        assert (time.time() - start) < 1, "Timeout exceeded"
 
-        start = time.time()
-        probe = Probe(timeout=5, pause=0.5, fnc=value_err_raise)
+        probe = Probe(timeout=1, pause=0.2, fnc=value_err_raise)
         probe.run_in_background()
         with pytest.raises(ValueError):
             probe.join()
-        assert (time.time() - start) < 1, "Timeout exceeded"
 
     def test_count(self):
         def say_no():
@@ -135,18 +130,18 @@ def test_reach_timeout(self):
         # probe should reach timeout with long-running function calls
         # probe and caller in one thread
         start = time.time()
-        probe = Probe(timeout=3, pause=0.5, fnc=snoozer, seconds=10)
+        probe = Probe(timeout=1, pause=0.2, fnc=snoozer, seconds=10)
         with pytest.raises(ProbeTimeout):
             probe.run()
-        assert (time.time() - start) > 2, "Timeout not reached with unsuccessful function"
+        assert (time.time() - start) < 3, "Time elapsed is way too high"
 
         # in background
         start = time.time()
-        probe = Probe(timeout=3, pause=0.5, fnc=snoozer, seconds=10)
+        probe = Probe(timeout=1, pause=0.2, fnc=snoozer, seconds=10)
         probe.run_in_background()
         with pytest.raises(ProbeTimeout):
             probe.join()
-        assert (time.time() - start) > 2, "Timeout not reached with unsuccessful function"
+        assert (time.time() - start) < 3, "Time elapsed is way too high"
 
     def test_expected_retval(self):
         def truth():
@@ -159,14 +154,14 @@ def lie():
         start = time.time()
         probe = Probe(timeout=5, pause=0.5, fnc=truth, expected_retval=MESSAGE)
         probe.run()
-        assert (time.time() - start) < 1, "Timeout exceeded"
+        assert (time.time() - start) < 1.0, "Timeout exceeded"
 
         # probe should reach timeout when expected_retval is not reached
         start = time.time()
         probe = Probe(timeout=3, pause=0.5, fnc=lie, expected_retval=MESSAGE)
         with pytest.raises(ProbeTimeout):
             probe.run()
-        assert (time.time() - start) > 2, "Timeout not reached with unsuccessful function"
+        assert (time.time() - start) > 2.0, "Timeout not reached with unsuccessful function"
 
     def test_concurrency(self):
         pool = []