avoid --abort-on-container-exit to allow containers to react to SIGTERM

marten-seemann · marten-seemann · commit 74c84935a0d9 · 2020-10-15T22:11:13.000+07:00
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -34,12 +34,9 @@ services:
     environment:
       - CRON=$CRON
       - ROLE=server
-      - SERVER_PARAMS=$SERVER_PARAMS
       - SSLKEYLOGFILE=/logs/keys.log
       - QLOGDIR=/logs/qlog/
       - TESTCASE=$TESTCASE_SERVER
-    depends_on:
-      - sim
     cap_add: 
       - NET_ADMIN
     ulimits:
@@ -48,6 +45,8 @@ services:
       rightnet:
         ipv4_address: 193.167.100.100
         ipv6_address: fd00:cafe:cafe:100::100
+    extra_hosts:
+      - "sim:193.167.100.2"
 
   client:
     image: $CLIENT
@@ -61,13 +60,10 @@ services:
     environment:
       - CRON=$CRON
       - ROLE=client
-      - CLIENT_PARAMS=$CLIENT_PARAMS
       - SSLKEYLOGFILE=/logs/keys.log
       - QLOGDIR=/logs/qlog/
       - TESTCASE=$TESTCASE_CLIENT
       - REQUESTS=$REQUESTS
-    depends_on:
-      - sim
     cap_add: 
       - NET_ADMIN
     ulimits:
@@ -81,6 +77,7 @@ services:
       - "server6:fd00:cafe:cafe:100::100"
       - "server46:193.167.100.100"
       - "server46:fd00:cafe:cafe:100::100"
+      - "sim:193.167.0.2"
   
   iperf_server:
     image: martenseemann/quic-interop-iperf-endpoint
@@ -89,8 +86,6 @@ services:
     tty: true
     environment:
       - ROLE=server
-    depends_on:
-      - sim
     cap_add: 
       - NET_ADMIN
     networks:
@@ -110,8 +105,6 @@ services:
     tty: true
     environment:
       - ROLE=client
-    depends_on:
-      - sim
     cap_add: 
       - NET_ADMIN
     networks:
diff --git a/docker.py b/docker.py
@@ -0,0 +1,91 @@
+import io
+import logging
+import shutil
+import subprocess
+import threading
+
+
+class DockerRunner:
+    _containers = None
+    _cond = None
+    _timeout = 0  # in seconds
+    _expired = False
+
+    def __init__(self, timeout: int):
+        self._containers = []
+        self._cond = threading.Condition()
+        self._timeout = timeout
+
+    def add_container(self, name: str, env: dict):
+        self._containers.append({"name": name, "env": env})
+
+    def _run_container(self, cmd: str, env: dict, name: str):
+        self._execute(cmd, env, name)
+        with self._cond:
+            logging.debug("%s container returned.", name)
+            self._cond.notify()
+
+    def _execute(self, cmd: str, env: dict = {}, name: str = ""):
+        p = subprocess.Popen(
+            cmd.split(" "),
+            bufsize=1,
+            env=env,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            universal_newlines=True,
+        )
+        for line in p.stdout:
+            ll = ""
+            if name:
+                ll = name + ": "
+            ll = ll + line.rstrip()
+            logging.debug(ll)
+
+    def _run_timer(self):
+        logging.debug("Timer expired. Stopping all containers.")
+        self._expired = True
+        with self._cond:
+            self._cond.notify()
+
+    def run(self) -> (str, bool):  # returns if the timer expired
+        # also log to a string, so we can parse the output later
+        output_string = io.StringIO()
+        output_stream = logging.StreamHandler(output_string)
+        output_stream.setLevel(logging.DEBUG)
+        logging.getLogger().addHandler(output_stream)
+
+        threads = []
+        # Start all containers (in separate threads)
+        docker_compose = shutil.which("docker-compose")
+        for e in self._containers:
+            t = threading.Thread(
+                target=self._run_container,
+                kwargs={
+                    "cmd": docker_compose + " up " + e["name"],
+                    "env": e["env"],
+                    "name": e["name"],
+                },
+            )
+            t.start()
+            threads.append(t)
+        # set a timer
+        timer = threading.Timer(self._timeout, self._run_timer)
+        timer.start()
+
+        # Wait for the first container to exit.
+        # Then stop all other docker containers.
+        with self._cond:
+            self._cond.wait()
+            names = [x["name"] for x in self._containers]
+            self._execute(
+                shutil.which("docker-compose") + " stop -t 5 " + " ".join(names)
+            )
+        # wait for all threads to finish
+        for t in threads:
+            t.join()
+        timer.cancel()
+
+        output = output_string.getvalue()
+        output_string.close()
+        logging.getLogger().removeHandler(output_stream)
+        return output, self._expired
diff --git a/interop.py b/interop.py
@@ -16,6 +16,7 @@
 from termcolor import colored
 
 import testcases
+from docker import DockerRunner
 from result import TestResult
 from testcases import Perspective
 
@@ -98,9 +99,12 @@ def __init__(
                     self.measurement_results[server][client][measurement] = {}
 
     def _is_unsupported(self, lines: List[str]) -> bool:
-        return any("exited with code 127" in str(line) for line in lines) or any(
-            "exit status 127" in str(line) for line in lines
-        )
+        for line in lines:
+            if "sim exited with code 127" in str(line):
+                continue
+            if "exited with code 127" in str(line) or "exit status 127" in str(line):
+                return True
+        return False
 
     def _check_impl_is_compliant(self, name: str) -> bool:
         """ check if an implementation return UNSUPPORTED for unknown test cases """
@@ -121,46 +125,48 @@ def _check_impl_is_compliant(self, name: str) -> bool:
 
         # check that the client is capable of returning UNSUPPORTED
         logging.debug("Checking compliance of %s client", name)
-        cmd = (
-            "CERTS=" + certs_dir.name + " "
-            "TESTCASE_CLIENT=" + random_string(6) + " "
-            "SERVER_LOGS=/dev/null "
-            "CLIENT_LOGS=" + client_log_dir.name + " "
-            "WWW=" + www_dir.name + " "
-            "DOWNLOADS=" + downloads_dir.name + " "
-            'SCENARIO="simple-p2p --delay=15ms --bandwidth=10Mbps --queue=25" '
-            "CLIENT=" + self._implementations[name]["image"] + " "
-            "docker-compose up --timeout 0 --abort-on-container-exit -V sim client"
-        )
-        output = subprocess.run(
-            cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
-        )
-        if not self._is_unsupported(output.stdout.splitlines()):
+        env_sim = {"SCENARIO": "simple-p2p --delay=15ms --bandwidth=10Mbps --queue=25"}
+        env_client = {
+            "CERTS": "./certs",
+            "TESTCASE_CLIENT": random_string(6),
+            "DOWNLOADS": downloads_dir.name,
+            "CLIENT_LOGS": client_log_dir.name,
+            "CLIENT": self._implementations[name]["image"],
+        }
+        env = {}
+        env.update(env_sim)
+        env.update(env_client)
+        r = DockerRunner(15)
+        r.add_container("client", env)
+        r.add_container("sim", env)
+        output, expired = r.run()
+        if expired or not self._is_unsupported(output.splitlines()):
             logging.error("%s client not compliant.", name)
-            logging.debug("%s", output.stdout.decode("utf-8"))
+            logging.debug("%s", output)
             self.compliant[name] = False
             return False
         logging.debug("%s client compliant.", name)
 
         # check that the server is capable of returning UNSUPPORTED
         logging.debug("Checking compliance of %s server", name)
         server_log_dir = tempfile.TemporaryDirectory(dir="/tmp", prefix="logs_server_")
-        cmd = (
-            "CERTS=" + certs_dir.name + " "
-            "TESTCASE_SERVER=" + random_string(6) + " "
-            "SERVER_LOGS=" + server_log_dir.name + " "
-            "CLIENT_LOGS=/dev/null "
-            "WWW=" + www_dir.name + " "
-            "DOWNLOADS=" + downloads_dir.name + " "
-            "SERVER=" + self._implementations[name]["image"] + " "
-            "docker-compose up -V server"
-        )
-        output = subprocess.run(
-            cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
-        )
-        if not self._is_unsupported(output.stdout.splitlines()):
+        env_server = {
+            "CERTS": "./certs",
+            "TESTCASE_SERVER": random_string(6),
+            "SERVER_LOGS": server_log_dir.name,
+            "WWW": www_dir.name,
+            "SERVER": self._implementations[name]["image"],
+        }
+        env = {}
+        env.update(env_sim)
+        env.update(env_server)
+        r = DockerRunner(15)
+        r.add_container("server", env)
+        r.add_container("sim", env)
+        output, expired = r.run()
+        if expired or not self._is_unsupported(output.splitlines()):
             logging.error("%s server not compliant.", name)
-            logging.debug("%s", output.stdout.decode("utf-8"))
+            logging.debug("%s", output)
             self.compliant[name] = False
             return False
         logging.debug("%s server compliant.", name)
@@ -329,74 +335,54 @@ def _run_test(
             server_keylog_file=server_log_dir.name + "/keys.log",
         )
         print(
-            "Server: "
-            + server
-            + ". Client: "
-            + client
-            + ". Running test case: "
-            + str(testcase)
+            "Server: {}. Client: {}. Running test: {}".format(
+                server, client, str(testcase)
+            )
         )
 
         reqs = " ".join([testcase.urlprefix() + p for p in testcase.get_paths()])
         logging.debug("Requests: %s", reqs)
-        params = (
-            "WAITFORSERVER=server:443 "
-            "CERTS=" + testcase.certs_dir() + " "
-            "TESTCASE_SERVER=" + testcase.testname(Perspective.SERVER) + " "
-            "TESTCASE_CLIENT=" + testcase.testname(Perspective.CLIENT) + " "
-            "WWW=" + testcase.www_dir() + " "
-            "DOWNLOADS=" + testcase.download_dir() + " "
-            "SERVER_LOGS=" + server_log_dir.name + " "
-            "CLIENT_LOGS=" + client_log_dir.name + " "
-            'SCENARIO="{}" '
-            "CLIENT=" + self._implementations[client]["image"] + " "
-            "SERVER=" + self._implementations[server]["image"] + " "
-            'REQUESTS="' + reqs + '" '
-        ).format(testcase.scenario())
-        params += " ".join(testcase.additional_envs())
-        containers = "sim client server " + " ".join(testcase.additional_containers())
-        cmd = (
-            params
-            + " docker-compose up --abort-on-container-exit --timeout 1 "
-            + containers
-        )
-        logging.debug("Command: %s", cmd)
+        r = DockerRunner(timeout=testcase.timeout())
+        env_server = {
+            "CERTS": "./certs",
+            "TESTCASE_SERVER": testcase.testname(Perspective.SERVER),
+            "WWW": testcase.www_dir(),
+            "SERVER_LOGS": server_log_dir.name,
+            "SERVER": self._implementations[server]["image"],
+        }
+        env_client = {
+            "CERTS": "./certs",
+            "TESTCASE_CLIENT": testcase.testname(Perspective.CLIENT),
+            "DOWNLOADS": testcase.download_dir(),
+            "CLIENT_LOGS": client_log_dir.name,
+            "CLIENT": self._implementations[client]["image"],
+            "REQUESTS": reqs,
+        }
+        env_sim = {
+            "SCENARIO": testcase.scenario(),
+            "WAITFORSERVER": "server:443",
+        }
+        env = {}
+        env.update(env_server)
+        env.update(env_client)
+        env.update(env_sim)
+        r.add_container(name="server", env=env)
+        r.add_container(name="client", env=env)
+        r.add_container(name="sim", env=env)
+        for c in testcase.additional_containers():
+            r.add_container(name=c, env=testcase.additional_envs())
+        output, expired = r.run()
 
         status = TestResult.FAILED
-        output = ""
-        expired = False
-        try:
-            r = subprocess.run(
-                cmd,
-                shell=True,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.STDOUT,
-                timeout=testcase.timeout(),
-            )
-            output = r.stdout
-        except subprocess.TimeoutExpired as ex:
-            output = ex.stdout
-            expired = True
-
-        logging.debug("%s", output.decode("utf-8"))
-
-        if expired:
-            logging.debug("Test failed: took longer than %ds.", testcase.timeout())
-            r = subprocess.run(
-                "docker-compose stop " + containers,
-                shell=True,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.STDOUT,
-                timeout=60,
-            )
-            logging.debug("%s", r.stdout.decode("utf-8"))
 
         # copy the pcaps from the simulator
         self._copy_logs("sim", sim_log_dir)
         self._copy_logs("client", client_log_dir)
         self._copy_logs("server", server_log_dir)
 
-        if not expired:
+        if expired:
+            logging.debug("Test failed: took longer than %ds.", testcase.timeout())
+        else:
             lines = output.splitlines()
             if self._is_unsupported(lines):
                 status = TestResult.UNSUPPORTED
diff --git a/testcases.py b/testcases.py
@@ -106,11 +106,11 @@ def urlprefix() -> str:
 
     @staticmethod
     def additional_envs() -> List[str]:
-        return [""]
+        return []
 
     @staticmethod
     def additional_containers() -> List[str]:
-        return [""]
+        return []
 
     def www_dir(self):
         if not self._www_dir:
@@ -1406,8 +1406,8 @@ def timeout() -> int:
         return 180
 
     @staticmethod
-    def additional_envs() -> List[str]:
-        return ["IPERF_CONGESTION=cubic"]
+    def additional_envs() -> dict:
+        return {"IPERF_CONGESTION": "cubic"}
 
     @staticmethod
     def additional_containers() -> List[str]:
