Base for new dispatcher design

Merge in pipeline support
Add YAML line number retrieval.
Merge CompositeAction into Pipeline

Change-Id: I75cfec232c83340adb77d3ca85cce32da2086846

Author: Neil Williams

ci-run

@@ -4,4 +4,4 @@ set -e
 
 pep8 --ignore E501 .
 
-python setup.py test
+TMPDIR=. python setup.py test

lava/dispatcher/commands.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import sys
+import yaml
 
 from json_schema_validator.errors import ValidationError
 from lava.tool.command import Command
@@ -11,6 +12,7 @@
 import lava_dispatcher.config
 from lava_dispatcher.config import get_config, get_device_config, list_devices
 from lava_dispatcher.job import LavaTestJob, validate_job_data
+import lava_dispatcher.pipeline.parser
 
 
 class SetUserConfigDirAction(argparse.Action):
@@ -48,6 +50,43 @@ def invoke(self):
             print d
 
 
+def run_legacy_job(job_data, oob_file, config, output_dir, validate):
+
+    if os.getuid() != 0:
+        logging.error("lava dispatch has to be run as root")
+        exit(1)
+
+    json_job_data = json.dumps(job_data)
+    job = LavaTestJob(json_job_data, oob_file, config, output_dir)
+
+    #FIXME Return status
+    if validate:
+        try:
+            validate_job_data(job.job_data)
+        except ValidationError as e:
+            print e
+    else:
+        job.run()
+
+
+def get_pipeline_runner(job):
+    def run_pipeline_job(job_data, oob_file, config, output_dir, validate_only):
+        try:
+            # FIXME use job_data   (?)
+            # FIXME use oob_file   (!)
+            # FIXME use config     (!)
+            # FIXME use output_dir (!)
+            job.validate()
+            if not validate_only:
+                job.run()
+        except lava_dispatcher.pipeline.JobError as e:
+            print(e)
+            sys.exit(2)
+
+        # TODO make use of the arguments passed in
+    return run_pipeline_job
+
+
 class dispatch(DispatcherCommand):
     """
     Run test scenarios on virtual and physical hardware
@@ -86,10 +125,6 @@ def invoke(self):
 
         config = None
 
-        if os.getuid() != 0:
-            logging.error("lava dispatch has to be run as root")
-            exit(1)
-
         if self.args.oob_fd:
             oob_file = os.fdopen(self.args.oob_fd, 'w')
         else:
@@ -128,38 +163,36 @@ def invoke(self):
                 setproctitle("%s [job: %s]" % (
                     getproctitle(), self.args.job_id))
 
-        # Load the scenario file
-        with open(self.args.job_file) as stream:
-            jobdata = stream.read()
-            json_jobdata = json.loads(jobdata)
+        # Load the job file
+        job_runner, job_data = self.parse_job_file(self.args.job_file)
 
         # detect multinode and start a NodeDispatcher to work with the LAVA Coordinator.
         if not self.args.validate:
-            if 'target_group' in json_jobdata:
-                node = NodeDispatcher(json_jobdata, oob_file, self.args.output_dir)
+            if 'target_group' in job_data:
+                node = NodeDispatcher(job_data, oob_file, self.args.output_dir)
                 node.run()
                 # the NodeDispatcher has started and closed.
                 exit(0)
         if self.args.target is None:
-            if 'target' not in json_jobdata:
+            if 'target' not in job_data:
                 logging.error("The job file does not specify a target device. "
                               "You must specify one using the --target option.")
                 exit(1)
         else:
-            json_jobdata['target'] = self.args.target
-            jobdata = json.dumps(json_jobdata)
+            job_data['target'] = self.args.target
         if self.args.output_dir and not os.path.isdir(self.args.output_dir):
             os.makedirs(self.args.output_dir)
-        job = LavaTestJob(jobdata, oob_file, config, self.args.output_dir)
 
-        #FIXME Return status
-        if self.args.validate:
-            try:
-                validate_job_data(job.job_data)
-            except ValidationError as e:
-                print e
-        else:
-            job.run()
+        job_runner(job_data, oob_file, config, self.args.output_dir, self.args.validate)
+
+    def parse_job_file(self, filename):
+        if filename.lower().endswith('.yaml') or filename.lower().endswith('.yml'):
+            parser = lava_dispatcher.pipeline.parser.JobParser()
+            job = parser.parse(open(filename))
+            return get_pipeline_runner(job), job.parameters
+
+        # everything else is assumed to be JSON
+        return run_legacy_job, json.load(open(filename))
 
 
 class DeviceCommand(DispatcherCommand):

lava_dispatcher/pipeline/README.rst

@@ -0,0 +1,165 @@
+Requirements
+============
+
+Functional:
+
+- It should be possible to have multiple simultaneous connections to the
+  device and run parallel tests on it. Example: one serial connection
+  and one SSH connection.
+
+- It should be possible to interact not only with "high-level" software
+  such as bootloader and OS, but with bare metal as well.
+
+- Support for different types of images:
+
+  - pre-built image
+  - linaro hwpack + rootfs
+  - kernel + ramdisk/rootfs
+  - tftp
+  - nfsroot
+
+- Support for different bootloaders on same platform. Example: uboot,
+  uefi, and second stage (grub) pipeline
+
+- It should be possible to choose which device to boot from. This
+  impacts both the deployment code and the boot code
+
+- It must be possible to test advanced multi boot test cases with
+  repetition - suspend, kexec, wake test cases. In special it is
+  necessary to test wake, suspend, reboot, kexec etc.
+
+- The dispatcher should be able to provide interactive support to
+  low-level serial.  For some new devices, remote bringup is often
+  necessary because developers can't have a device on their desks. When
+  necessary, interact with the scheduler to put board online/offline.
+
+Non-functional:
+
+- Speed. Avoid as much overhead as possible.
+
+- Security. Should not require to be run as root. If necessary, let's
+  have a separate helper program that can be setuid to do the stuff that
+  actually needs root privileges.
+
+- Simplicity.
+
+  - Having master image and test system on the same device makes several
+    actions harder than they need to be. Master images must be booted
+    from the network so that the actualy storage on the devices are left
+    entirely to the test system. When possibel, deployment to the test
+    system should be done by "just" dd'ing an image to the desired
+    device.
+
+  - Avoid as much as possible running commands on the target. When
+    it is possible to perform some operation in the dispatcher host,
+    let's not perform it on the target.
+
+Design
+======
+
+The proposed design is based around the Pipes and Filters architectural
+pattern, which is reified for instance in the UNIX pipes system. The
+idea is to have every piece of funcionality as self-contained as
+possible, and to be able to compose them in sequence to achieve the
+desired high-level funcionality.
+
+Main concepts in the design
+---------------------------
+
+- *Device* represents the device under test.
+
+- *Connection* is a data connection between the dispatcher host and the
+  device under test. Examples of connections: serial connection, SSH
+  connection, adb shell, etc.
+
+- *Action* an action that has to be performed. A Action can be a
+  shell commands run on the target, an operations run on
+  the dispatcher host, or anything. Actions should be as constrained as
+  possible so that all possible errors can be easily tracked. Where
+  multiple operations are required, use an action which contains
+  an internal pipeline and add the individual commands as actions
+  within that pipeline.
+
+  Actions must be aggregated into a *Pipeline* - the top level object is
+  always a pipeline. Pipelines can repeat actions and actions can include
+  internal pipelines containing more actions. Actions have parameters which
+  are set during the parsing of the YAML submission. Parameter data is
+  static within each action and is used to validate the action before any
+  pipeline is run. Dynamic data is set in the context which is available
+  via the parent pipeline of any action. Actions must be idempotent and
+  must raise a RuntimeError exception if the dynamic data is absent or
+  unusable. Errors in parameter data must raise a JobError exception.
+  Each command will receive a connection as an input parameter and can
+  optionally provide a different connection to the command that
+  comes after it. Usually, the first command in a pipeline will receive
+  *None* as connection, and must provide a connection to the subsequent
+  command.
+
+  See `Connection Management`_ below for other requirements that
+  Actions must observe.
+
+- *Image* represents the test system that needs to be deployed to the
+  target.
+
+  Each command in a pipeline will be given a chance to insert data into
+  the root filesystem of the image, before the pipeline starts to run.
+
+- *Deployment* is a strategy to deploy a given image to a given device.
+  Subclasses of deployment represent the different ways of deploying
+  images to device, which depend on both the type of image and on the
+  capabilities of the device.
+
+- *Job*. A Job aggregates a *Device* representing the target device to
+  be used, an *Image* to be deployed, and *Action* to be executed. The
+  Action can be, and usually *will* be, a composite command composed
+  of several subcommands.
+
+  The chosen deployment strategy will be chosen based on the image and
+  the device.
+
+Connection management
+---------------------
+
+Connections to devices under test are often unreliable and have been a
+major source of problems in automation. This way, in the case where a
+connection failure (disconnection, serial corruption) during the
+execution of a command, that command will be re-tried. Because of this,
+every step performed by a command must be prepared to be idempotent,
+i.e. to do nothing in the case where it has been performed before, and
+more importantly, to not crash if has been performed before.
+
+Exceptions
+----------
+
+LAVA must be clear on what was the likely cause of an incomplete test
+job or a failed test result. Any one failure must trigger only one
+exception. e.g. A JobError which results in a RuntimeError is still
+a bug in the dispatcher code as it should have been caught during
+the validation step.
+
+- *JobError*: An Error arising from the information supplied as part of
+    the TestJob. e.g. HTTP404 on a file to be downloaded as part of the
+    preparation of the TestJob or a download which results in a file
+    which tar or gzip does not recognise. This exception is used when
+    data supplied as the parameters to an Action causes that action
+    to fail. Job errors should always be supported by a unit test.
+
+- *InfrastructureError: Exceptions based on an error raised by a component
+    of the test which is neither the LAVA dispatcher code nor the
+    code being executed on the device under test. This includes
+    errors arising from the device (like the arndale SD controller
+    issue) and errors arising from the hardware to which the device
+    is connected (serial console connection, ethernet switches or
+    internet connection beyond the control of the device under test).
+    Actions are required to include code to check for likely
+    infrastructure errors so that pipelines can retry or fail the
+    test, recording whether a retry fixed the infrastructure error.
+
+- *TestError*: exceptions raised when the device under test did not
+    behave as expected.
+
+- *RuntimeError*: Exceptions arising from dynamic data prepared by
+    LAVA Dispatcher and failures of Actions not already handled by
+    the code. Runtime errors are bugs in lava-dispatcher code. (It is
+    also a bug to use the wrong exception type). Fixes for runtime
+    error bugs should always include a unit test.

lava_dispatcher/pipeline/__init__.py

@@ -0,0 +1,89 @@
+import sys
+from collections import OrderedDict
+from lava_dispatcher.pipeline.action import *
+from lava_dispatcher.pipeline.serial import *
+from lava_dispatcher.pipeline.ssh import *
+from lava_dispatcher.pipeline.shell import *
+from lava_dispatcher.pipeline.run import *
+
+from lava_dispatcher.pipeline.job_actions.boot import *
+from lava_dispatcher.pipeline.job_actions.commands import *
+from lava_dispatcher.pipeline.job_actions.deploy import *
+from lava_dispatcher.pipeline.job_actions.logic import *
+from lava_dispatcher.pipeline.job_actions.submit import *
+from lava_dispatcher.pipeline.job_actions.test import *
+
+
+class Pipeline(object):
+
+    def __init__(self, parent=None):
+        self.children = {}
+        self.actions = []
+        self.summary = "pipeline"
+        self.parent = None
+        self.branch_level = 1  # the level of the last added child
+        if not parent:
+            self.children = {self: self.actions}
+        elif not parent.level:
+            raise RuntimeError("Tried to create a pipeline with an invalid parent.")
+        else:
+            # parent must be an Action
+            if type(parent) != Action:
+                raise RuntimeError("Internal pipelines need an Action as a parent")
+            self.parent = parent
+            self.branch_level = parent.level
+
+    def add_action(self, action):
+
+        if not action or not issubclass(type(action), Action):
+            raise RuntimeError("Only actions can be added to a pipeline: %s" % action)
+        if not action:
+            raise RuntimeError("Unable to add empty action to pipeline")
+        self.actions.append(action)
+        action.level = "%s.%s" % (self.branch_level, len(self.actions))
+        if self.parent:
+            self.children.update({self: self.actions})
+            self.parent.pipeline = self
+        else:
+            action.level = "%s" % (len(self.actions))
+
+    def _describe(self, structure):
+        for action in self.actions:
+            structure[action.level] = {
+                'description': action.description,
+                'summary': action.summary,
+                'content': action.explode()
+            }
+            if not action.pipeline:
+                continue
+            action.pipeline._describe(structure)
+
+    def describe(self):
+        """
+        Describe the current pipeline, recursing through any
+        internal pipelines.
+        :return: JSON string of the structure
+        """
+        structure = OrderedDict()
+        self._describe(structure)
+        return structure
+
+    @property
+    def errors(self):
+        sub_action_errors = [a.errors for a in self.actions]
+        return reduce(lambda a, b: a + b, sub_action_errors)
+
+    def run_actions(self, connection, args=None):
+        for action in self.actions:
+            new_connection = action.run(connection, args)
+            if new_connection:
+                connection = new_connection
+        return connection
+
+    def prepare_actions(self):
+        for action in self.actions:
+            action.prepare()
+
+    def post_process_actions(self):
+        for action in self.actions:
+            action.post_process()