feat: validate test plan before running

The test plan provided is now validated during the first step
to ensure it is valid before proceeding.

Closes #21

ardere/aws.py

@@ -187,13 +187,12 @@ def create_service(self, step):
             self.s3_ready_file,
             step.get("run_delay", 0)
         )
-        service_cmd = step["additional_command_args"]
+        service_cmd = step["command"]
         cmd = ['sh', '-c', '{} && {}'.format(wfc_cmd, service_cmd)]
 
         # Prep the env vars
         env_vars = [{"name": wfc_var, "value": shell_script}]
-        for env_var in step.get("environment_data", []):
-            name, value = env_var.split("=", 1)
+        for name, value in step.get("env", {}).items():
             env_vars.append({"name": name, "value": value})
 
         # ECS wants a family name for task definitions, no spaces, 255 chars
@@ -208,8 +207,6 @@ def create_service(self, step):
             "name": step["name"],
             "image": step["container_name"],
             "cpu": cpu_units,
-            # use host network mode for optimal performance
-            "networkMode": "host",
 
             # using only memoryReservation sets no hard limit
             "memoryReservation": 256,
@@ -228,18 +225,17 @@ def create_service(self, step):
         }
 
         if "port_mapping" in step:
-            ports = []
-            for port_map in step["port_mapping"].split(","):
-                ports.append({
-                    "containerPort": int(port_map)
-                })
+            ports = [{"containerPort": port} for port in step["port_mapping"]]
             container_def["portMappings"] = ports
 
         task_response = self._ecs_client.register_task_definition(
             family=family_name,
             containerDefinitions=[
                 container_def
             ],
+            # use host network mode for optimal performance
+            networkMode="host",
+
             placementConstraints=[
                 # Ensure the service is confined to the right instance type
                 {

ardere/exceptions.py

@@ -4,3 +4,7 @@ class ServicesStartingException(Exception):
 
 class ShutdownPlanException(Exception):
     """Exception to indicate the Plan should be Shutdown"""
+
+
+class ValidationException(Exception):
+    """Exception to indicate validation error parsing input"""

ardere/step_functions.py

@@ -6,18 +6,61 @@
 import boto3
 import botocore
 import toml
+from marshmallow import (
+    Schema,
+    decorators,
+    fields,
+    validate,
+    ValidationError,
+)
 from typing import Any, Dict, List  # noqa
 
-from ardere.aws import ECSManager
+from ardere.aws import (
+    ECSManager,
+    ec2_vcpu_by_type,
+)
 from ardere.exceptions import (
     ServicesStartingException,
-    ShutdownPlanException
+    ShutdownPlanException,
+    ValidationException,
 )
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
 
+class StepValidator(Schema):
+    name = fields.String(required=True)
+    instance_count = fields.Int(required=True)
+    instance_type = fields.String(
+        required=True,
+        validate=validate.OneOf(ec2_vcpu_by_type.keys())
+    )
+    run_max_time = fields.Int(required=True)
+    run_delay = fields.Int()
+    container_name = fields.String(required=True)
+    command = fields.String(required=True)
+    port_mapping = fields.List(fields.Int())
+    env = fields.Dict()
+
+
+class PlanValidator(Schema):
+    ecs_name = fields.String(required=True)
+    name = fields.String(required=True)
+
+    steps = fields.Nested(StepValidator, many=True)
+
+    @decorators.validates("ecs_name")
+    def validate_ecs_name(self, value):
+        """Verify a cluster exists for this name"""
+        client = self.context["boto"].client('ecs')
+        response = client.describe_clusters(
+            clusters=[value]
+        )
+        if not response.get("clusters"):
+            raise ValidationError("No cluster with the provided name.")
+
+
 class AsynchronousPlanRunner(object):
     """Asynchronous Test Plan Runner
 
@@ -61,13 +104,24 @@ def _load_toml(self, event):
         """Loads TOML if necessary"""
         return toml.loads(event["toml"]) if "toml" in event else event
 
+    def _validate_plan(self):
+        """Validates that the loaded plan is correct"""
+        schema = PlanValidator()
+        schema.context["boto"] = self.boto
+        _, errors = schema.load(self.event)
+        if errors:
+            raise ValidationException("Failed to validate: {}".format(errors))
+
     def populate_missing_instances(self):
         """Populate any missing EC2 instances needed for the test plan in the
         cluster
 
         Step 1
 
         """
+        # First, validate the test plan, done only as part of step 1
+        self._validate_plan()
+
         needed = self._build_instance_map()
         logger.info("Plan instances needed: {}".format(needed))
         current_instances = self.ecs.query_active_instances()

requirements.txt

@@ -1,3 +1,4 @@
 futures==3.0.5
 typing==3.5.3.0
-toml==0.9.2
+toml==0.9.2
+marshmallow==2.13.4

serverless.yml

@@ -32,6 +32,7 @@ provider:
          - "ecs:CreateCluster"
          - "ecs:ListServices"
          - "ecs:ListContainerInstances"
+         - "ecs:DescribeClusters"
          - "ecs:DescribeServices"
          - "ecs:DescribeTaskDefinition"
          - "ecs:DescribeContainerInstances"

tests/fixtures.py

@@ -9,12 +9,12 @@
       "instance_count": 1,
       "instance_type": "t2.medium",
       "run_max_time": 140,
-      "environment_data": [
-          "SOME_VAR=great-value"
-      ],
-      "port_mapping": "8000,4000",
+      "env": {
+          "SOME_VAR": "great-value"
+      },
+      "port_mapping": [8000, 4000],
       "container_name": "bbangert/ap-loadtester:latest",
-      "additional_command_args": "./apenv/bin/aplt_testplan wss://autopush.stage.mozaws.net 'aplt.scenarios:notification_forever,1000,1,0'"
+      "command": "./apenv/bin/aplt_testplan wss://autopush.stage.mozaws.net 'aplt.scenarios:notification_forever,1000,1,0'"
     }
   ]
 }
@@ -31,7 +31,7 @@
     instance_count = 8
     instance_type = "m3.medium"
     container_name = "bbangert/ap-loadtester:latest"
-    additional_command_args = "./apenv/bin/aplt_testplan wss://autopush.stage.mozaws.net 'aplt.scenarios:connect_and_idle_forever,10000,5,0'"
+    command = "./apenv/bin/aplt_testplan wss://autopush.stage.mozaws.net 'aplt.scenarios:connect_and_idle_forever,10000,5,0'"
     run_max_time = 300
     volume_mapping = "/var/log:/var/log/$RUN_ID:rw"
     docker_series = "push_tests"
@@ -42,7 +42,7 @@
     run_delay = 330
     instance_type = "m3.medium"
     container_name = "bbangert/ap-loadtester:latest"
-    additional_command_args = "./apenv/bin/aplt_testplan wss://autopush.stage.mozaws.net 'aplt.scenarios:connect_and_idle_forever,10000,5,0'"
+    command = "./apenv/bin/aplt_testplan wss://autopush.stage.mozaws.net 'aplt.scenarios:connect_and_idle_forever,10000,5,0'"
     run_max_time = 300
     volume_mapping = "/var/log:/var/log/$RUN_ID:rw"
     docker_series = "push_tests"

tests/test_aws.py

@@ -104,7 +104,6 @@ def test_create_service(self):
         _, kwargs = ecs._ecs_client.register_task_definition.call_args
         container_def = kwargs["containerDefinitions"][0]
         ok_("portMappings" in container_def)
-        eq_(container_def["networkMode"], "host")
 
     def test_create_services(self):
         ecs = self._make_FUT()

tests/test_step_functions.py

@@ -50,6 +50,14 @@ def test_populate_missing_instances(self):
         self.mock_ecs.query_active_instances.assert_called()
         self.mock_ecs.request_instances.assert_called()
 
+    def test_populate_missing_instances_fail(self):
+        from ardere.exceptions import ValidationException
+        mock_client = mock.Mock()
+        self.mock_boto.client.return_value = mock_client
+        mock_client.describe_clusters.return_value = {"clusters": []}
+        assert_raises(ValidationException,
+                      self.runner.populate_missing_instances)
+
     def test_create_ecs_services(self):
         self.runner.create_ecs_services()
         self.mock_ecs.create_services.assert_called_with(self.plan["steps"])
@@ -133,3 +141,28 @@ def test_cleanup_cluster_error(self):
         )
         self.runner.cleanup_cluster()
         mock_s3.Object.assert_called()
+
+
+class TestValidation(unittest.TestCase):
+    def _make_FUT(self):
+        from ardere.step_functions import PlanValidator
+        return PlanValidator()
+
+    def test_validate_success(self):
+        schema = self._make_FUT()
+        schema.context["boto"] = mock.Mock()
+        plan = json.loads(fixtures.sample_basic_test_plan)
+        data, errors = schema.load(plan)
+        eq_(errors, {})
+        eq_(len(data["steps"]), len(plan["steps"]))
+
+    def test_validate_fail(self):
+        schema = self._make_FUT()
+        schema.context["boto"] = mock_boto = mock.Mock()
+        mock_client = mock.Mock()
+        mock_boto.client.return_value = mock_client
+        mock_client.describe_clusters.return_value = {"clusters": []}
+        plan = json.loads(fixtures.sample_basic_test_plan)
+        data, errors = schema.load(plan)
+        eq_(len(data["steps"]), len(plan["steps"]))
+        eq_(len(errors), 1)