Add netem chaos tests (MaterializeInc#3387)

JLDLaughlin · web-flow · commit 803fbd6c366c · 2020-06-16T17:53:55.000-04:00
Use pumba (a tool for chaos testing Docker containers) to add the following
chaos networking tests: delay, rate, loss, duplication, corruption. Each new type
of networking test is added as a new workflow in the chaos crate.
diff --git a/misc/python/materialize/cli/mzconduct.py b/misc/python/materialize/cli/mzconduct.py
@@ -19,6 +19,7 @@
 import webbrowser
 from datetime import datetime, timezone
 from pathlib import Path
+from threading import Thread
 from typing import (
     Any,
     Callable,
@@ -946,27 +947,110 @@ def run(self, comp: Composition, workflow: Workflow) -> None:
             raise Failed(f"Unable to kill container {container_id}: {e}")
 
 
-@Steps.register("chaos-delay-docker")
-class ChaosDelayDockerStep(WorkflowStep):
-    """Delay the incoming and outgoing network traffic for a Docker service.
+class ChaosNetemStep(WorkflowStep):
+    """Base class for running network chaos tests against a Docker container.
 
-    Params:
-        service: Docker service to delay, will be used to grep for container id
-                 NOTE: service name must be unique to correctly match the container id
-        delay: milliseconds to delay network traffic (default: 100ms)
+    We use pumba (a chaos testing tool for Docker) to run the various netem tests.
+    pumba only supports Docker container names (not container ids), meaning that
+    we have to pass the exact container name to these steps. We should fix this in
+    the future.
     """
 
-    def __init__(self, service: str, delay: int = 100) -> None:
-        self._service = service
-        self._delay = delay
+    def __init__(self, duration: int):
+        self._duration = duration
 
     def run(self, comp: Composition, workflow: Workflow) -> None:
+        if self._duration == -1:
+            # If duration isn't provided, run for 7 days in a non-blocking thread.
+            duration = 10080
+        else:
+            duration = self._duration
+
+        cmd = self.get_cmd(duration).split()
+        netem_thread = Thread(target=self.threaded_netem, args=(cmd,), daemon=True)
+        netem_thread.start()
+
+    def get_cmd(self, duration: int) -> str:
+        pass
+
+    def threaded_netem(self, cmd: List[str]) -> None:
         try:
-            container_id = comp.get_container_id(self._service)
-            cmd = f"docker exec {container_id} tc qdisc add dev eth0 root netem delay {self._delay}ms".split()
             spawn.runv(cmd)
         except subprocess.CalledProcessError as e:
-            raise Failed(f"Unable to delay container {container_id}: {e}")
+            raise Failed(f"Unable to run netem chaos command: {e}")
+
+
+@Steps.register("chaos-delay-docker")
+class ChaosDelayDockerStep(ChaosNetemStep):
+    """Delay the egress network traffic for a Docker service.
+    """
+
+    def __init__(
+        self, container: str, duration: int = -1, delay: int = 250, jitter: int = 250,
+    ) -> None:
+        super().__init__(duration=duration)
+        self._container = container
+        self._delay = delay
+        self._jitter = jitter
+
+    def get_cmd(self, duration: int) -> str:
+        return f"pumba --random netem --duration {duration}m delay --time {self._delay} \
+            --jitter {self._jitter} --distribution normal {self._container}"
+
+
+@Steps.register("chaos-rate-docker")
+class ChaosRateDockerStep(ChaosNetemStep):
+    """Limit the egress network traffic for a Docker service.
+    """
+
+    def __init__(self, container: str, duration: int = -1) -> None:
+        super().__init__(duration=duration)
+        self._container = container
+
+    def get_cmd(self, duration: int) -> str:
+        return f"pumba netem --duration {duration}m rate {self._container}"
+
+
+@Steps.register("chaos-loss-docker")
+class ChaosLossDockerStep(ChaosNetemStep):
+    """Lose a percent of a Docker container's network packets.
+    """
+
+    def __init__(self, container: str, percent: int, duration: int = -1) -> None:
+        super().__init__(duration=duration)
+        self._container = container
+        self._percent = percent
+
+    def get_cmd(self, duration: int) -> str:
+        return f"pumba netem --duration {duration}m loss --percent {self._percent} {self._container}"
+
+
+@Steps.register("chaos-duplicate-docker")
+class ChaosDuplicateDockerStep(ChaosNetemStep):
+    """Duplicate a percent of a Docker container's network packets.
+    """
+
+    def __init__(self, container: str, percent: int, duration: int = -1) -> None:
+        super().__init__(duration=duration)
+        self._container = container
+        self._percent = percent
+
+    def get_cmd(self, duration: int) -> str:
+        return f"pumba netem --duration {duration}m duplicate --percent {self._percent} {self._container}"
+
+
+@Steps.register("chaos-corrupt-docker")
+class ChaosCorruptDockerStep(ChaosNetemStep):
+    """Corrupt a percent of a Docker container's network packets.
+    """
+
+    def __init__(self, container: str, percent: int, duration: int = -1) -> None:
+        super().__init__(duration=duration)
+        self._container = container
+        self._percent = percent
+
+    def get_cmd(self, duration: int) -> str:
+        return f"pumba netem --duration {duration}m corrupt --percent {self._percent} {self._container}"
 
 
 @Steps.register("chaos-confirm")
@@ -991,6 +1075,10 @@ def run(self, comp: Composition, workflow: Workflow) -> None:
             if not comp.docker_container_is_running(container_id):
                 raise Failed(f"chaos-confirm: container {container_id} is not running")
         else:
+            if comp.docker_container_is_running(container_id):
+                raise Failed(
+                    f"chaos-confirm: expected {container_id} to have exited, is running"
+                )
             actual_exit_code = comp.docker_inspect("{{.State.ExitCode}}", container_id)
             if actual_exit_code != f"'{self._exit_code}'":
                 raise Failed(
diff --git a/test/chaos/mzcompose.yml b/test/chaos/mzcompose.yml
@@ -69,35 +69,58 @@ services:
 
 mzconduct:
   workflows:
-    # This test is designed to delay inbound and outbound network
-    # traffic of the Kafka broker.
-    #
-    # Success: Kafka broker traffic is delayed, materialize does not
-    #          crash, chaos container exits successfully.
-    #          Note: to confirm that broker traffic is delayed, you can run the
-    #                following on the Kafka broker's container: `tc qdisc show  dev eth0`
-    # Failure: Kafka broker traffic is not delayed, materialize crashes,
-    #          or chaos container exits unsuccessfully.
+    # This test is designed to delay egress network traffic of the Kafka broker.
     delay-kafka:
       steps:
         - step: workflow
           workflow: start-everything
         - step: chaos-delay-docker
-          service: kafka
-        - step: run
-          service: chaos
-          command: >-
-            --materialized-host materialized
-            --materialized-port 6875
-            --kafka-url kafka:9092
-            --kafka-partitions 5
-            --message-count 1000000
-        - step: chaos-confirm
-          service: materialized
-          running: true
-        - step: chaos-confirm
-          service: chaos_run
-          exit_code: 0
+          container: chaos_kafka_1
+        - step: workflow
+          workflow: run-netem-and-confirm
+
+    # This test is designed to rate limit egress network traffic of the Kafka broker.
+    rate-kafka:
+      steps:
+        - step: workflow
+          workflow: start-everything
+        - step: chaos-rate-docker
+          container: chaos_kafka_1
+        - step: workflow
+          workflow: run-netem-and-confirm
+
+    # This test is designed to test packet loss from the Kafka broker.
+    loss-kafka:
+      steps:
+        - step: workflow
+          workflow: start-everything
+        - step: chaos-loss-docker
+          container: chaos_kafka_1
+          percent: 10
+        - step: workflow
+          workflow: run-netem-and-confirm
+
+    # This test is designed to test packet loss from the Kafka broker.
+    duplicate-kafka:
+      steps:
+        - step: workflow
+          workflow: start-everything
+        - step: chaos-duplicate-docker
+          container: chaos_kafka_1
+          percent: 10
+        - step: workflow
+          workflow: run-netem-and-confirm
+
+    # This test is designed to test packet corruption from the Kafka broker.
+    corrupt-kafka:
+      steps:
+        - step: workflow
+          workflow: start-everything
+        - step: chaos-corrupt-docker
+          container: chaos_kafka_1
+          percent: 10
+        - step: workflow
+          workflow: run-netem-and-confirm
 
     # This test is designed to pause and unpause the running Kafka broker
     # (chaos-kill-container sends a SIGSTOP signal to the container).
@@ -192,3 +215,29 @@ mzconduct:
         - step: wait-for-tcp
           host: materialized
           port: 6875
+
+    # To verify a netem chaos test, we run the chaos container to completion
+    # and check the following conditions:
+    #
+    # - the materialized container did not crash
+    # - the kafka container did not crash
+    # - the chaos container completed successfully
+    run-netem-and-confirm:
+      steps:
+        - step: run
+          service: chaos
+          command: >-
+            --materialized-host materialized
+            --materialized-port 6875
+            --kafka-url kafka:9092
+            --kafka-partitions 100
+            --message-count 100000000
+        - step: chaos-confirm
+          service: materialized
+          running: true
+        - step: chaos-confirm
+          service: kafka
+          running: true
+        - step: chaos-confirm
+          service: chaos_run
+          exit_code: 0
