Merge lp:~david-schwarz/lava-dispatcher/multi-target into lp:lava-dispatcher

[David Schwarz]

The changes on this branch give lava-dispatcher the capability to manage and coordinate tests on multiple target machines concurrently. This allows the possibility of tests requiring coordination between multiple machines or multiple groups of machines.

Note that the Python module IPy must be installed on the dispatcher host.

Unmerged revisions

80. By David Schwarz on 2011-07-22

actions: Add scp and verify_file_present actions and example

79. By David Schwarz on 2011-07-22

multi-target: Run tests involving multiple target machines

Each target machines' action sequence runs in a separate thread.
At the end of the run, results are pooled from all threads and
sent as a single bundle to the dashboard server.

Note that package python-ipy must be installed on the
system running the dispatcher.

=== added file 'doc/multi-target_test.json'
--- doc/multi-target_test.json	1970-01-01 00:00:00 +0000
+++ doc/multi-target_test.json	2011-07-22 22:31:22 +0000
@@ -0,0 +1,46 @@
+{
+  "job_name": "multi-target_test",
+  "client_groups": [
+    {
+      "name": "group1",
+      "target_hosts": ["192.168.10.1-192.168.10.10"]
+    },
+    {
+      "name": "group2",
+      "target_hosts": ["192.168.85.11",
+                       "192.168.87.28",
+                       "host1.linaro.org",
+                       "host2",
+                       "host16.hyphenated-domain.com"]
+    }
+  ],
+  "timeout": 18000,
+  "actions": [
+    {
+      "command": "ls",
+      "target_groups": ["group1"]
+    },
+    {
+      "command": "sync_to_label",
+      "target_groups": ["group1","group2"],
+      "parameters":
+        {
+          "label": "ls_done",
+          "block": true
+        }
+    },
+    {
+      "command": "date",
+      "target_groups": ["group1", "group2"]
+    },
+    {
+      "command": "submit_results",
+      "parameters":
+        {
+          "server": "http://validation.linaro.org/launch-control",
+          "stream": "/anonymous/ls-test/",
+          "skip_remote": true
+        }
+    }
+  ]
+}

=== added file 'doc/multi-target_test_2.json'
--- doc/multi-target_test_2.json	1970-01-01 00:00:00 +0000
+++ doc/multi-target_test_2.json	2011-07-22 22:31:22 +0000
@@ -0,0 +1,81 @@
+{
+  "job_name": "multi-target_test_2",
+  "client_groups": [
+    {
+      "name": "group1",
+      "target_hosts": ["panda01"]
+    },
+    {
+      "name": "group2",
+      "target_hosts": ["panda02"]
+    }
+  ],
+  "timeout": 18000,
+  "actions": [
+    {
+      "command": "sync_to_label",
+      "target_groups": ["group1","group2"],
+      "parameters":
+        {
+          "label": "all_clients_up",
+          "block": true
+        }
+    },
+    {
+      "target_groups": ["group1"],
+      "command": "scp",
+      "parameters":
+        {
+          "username": "tester",
+          "host": "10.1.1.1",
+          "filename": "~/exists.txt",
+          "target_path": ""
+        }
+    },
+    {
+      "target_groups": ["group1"],
+      "command": "scp",
+      "parameters":
+        {
+          "username": "tester",
+          "host": "10.1.1.1",
+          "filename": "~/doesnotexist.txt",
+          "target_path": ""
+        }
+    },
+    {
+      "command": "sync_to_label",
+      "target_groups": ["group1","group2"],
+      "parameters":
+        {
+          "label": "wait_for_scp",
+          "block": true
+        }
+    },
+    {
+      "command": "verify_file_present",
+      "target_groups": ["group2"],
+      "parameters":
+        {
+          "file": "~/exists.txt"
+        }
+    },
+    {
+      "command": "verify_file_present",
+      "target_groups": ["group2"],
+      "parameters":
+        {
+          "file": "~/doesnotexist.txt"
+        }
+    },
+    {
+      "command": "submit_results",
+      "parameters":
+        {
+          "server": "http://validation.linaro.org/launch-control",
+          "stream": "/anonymous/panda_multi_scp/",
+          "skip_remote": true
+        }
+    }
+  ]
+}

=== modified file 'lava-dispatch'
--- lava-dispatch	2011-06-27 04:55:08 +0000
+++ lava-dispatch	2011-07-22 22:31:22 +0000
@@ -20,6 +20,7 @@
 # with this program; if not, see <http://www.gnu.org/licenses>.
 
 import sys
+import threading
 
 from lava_dispatcher import LavaTestJob
 
@@ -28,6 +29,8 @@
     print >> sys.stderr, "Usage:\n    lava-dispatch <json job file>"
     sys.exit(status)
 
+threading.stack_size(256000);
+
 if len(sys.argv) != 2:
     usage(1)
 

=== modified file 'lava_dispatcher/__init__.py'
--- lava_dispatcher/__init__.py	2011-07-21 16:55:47 +0000
+++ lava_dispatcher/__init__.py	2011-07-22 22:31:22 +0000
@@ -25,51 +25,185 @@
 from uuid import uuid1
 import base64
 import pexpect
+import re
+from operator import add
 
 from lava_dispatcher.actions import get_all_cmds
 from lava_dispatcher.client import LavaClient, CriticalError, GeneralError
 from lava_dispatcher.android_client import LavaAndroidClient
+<<<<<<< TREE
 
 __version__ = "0.1.0"
 
 class LavaTestJob(object):
+=======
+from threading import Thread
+from lava_dispatcher.utils import ip_addr_range
+from lava_dispatcher.sync import SyncMaster
+
+class LavaTestJob(SyncMaster):
+
+>>>>>>> MERGE-SOURCE
     def __init__(self, job_json):
+        super(LavaTestJob, self).__init__()
         self.job_status = 'pass'
         self.load_job_data(job_json)
-        self.context = LavaContext(self.target, self.image_type)
+        self.client_group_list = dict()
+        self._init_client_threads()
+        self.lava_commands = get_all_cmds()        
 
     def load_job_data(self, job_json):
         self.job_data = json.loads(job_json)
 
+    def _init_client_threads(self):
+        if not self.client_groups:
+            thread = ClientThread(self.target, self.image_type,
+                                  self.target_type, self)
+            self.client_group_list['default'] = thread
+        else:
+            for group in self.client_groups:
+                group_name = group['name']
+                self.client_group_list[group_name] = []
+                for hostname in self._gen_host(group):
+                    image_type = group.get('image_type')
+                    target_type = group.get('target_type')
+                    thread = ClientThread(hostname, image_type, target_type, 
+                                          self)
+                    self.client_group_list[group_name].append(thread)
+
+    def _gen_host(self, group):
+        for host_str in group['target_hosts']:
+            if re.match('^[\d\.]+-[\d\.]+$', host_str):
+                range = host_str.split('-')
+                assert len(range) == 2
+                for host in ip_addr_range(*range):
+                    yield host
+            else:
+   				yield host_str
+
+    def _iter_threads(self, thread_group_list):
+        for thread_group in thread_group_list.itervalues():
+            for thread in thread_group:
+                yield thread
+
     @property
     def target(self):
-        return self.job_data['target']
+        return self.job_data.get('target')
 
     @property
     def image_type(self):
-        if self.job_data.has_key('image_type'):
-            return self.job_data['image_type']
+		return self.job_data.get('image_type')
+
+    @property
+    def target_type(self):
+ 		return self.job_data.get('target_type')
+
+    @property
+    def client_groups(self):
+		return self.job_data.get('client_groups')
 
     def run(self):
-        lava_commands = get_all_cmds()
-
         if self.job_data['actions'][-1]['command'] == 'submit_results':
             submit_results = self.job_data['actions'].pop(-1)
         else:
             submit_results = None
 
+        for cmd in self.job_data['actions']:
+            target_groups = cmd.get('target_groups')
+            if not target_groups:
+                target_groups = {'default' : self.job_data['target']}
+            params = cmd.get('parameters', {})
+            metadata = cmd.get('metadata', {})
+
+            if cmd['command'] == 'sync_to_label':
+                self.init_sync_label(target_groups, params)
+
+            for group_name in target_groups:
+                for client_thread in self.client_group_list[group_name]:
+                    client_thread.add_action(self.lava_commands[cmd['command']],
+                                             cmd['command'],
+                                             params, 
+                                             metadata)
+        thread_iter = self._iter_threads(self.client_group_list)
+        for client_thread in thread_iter:
+            client_thread.start()
+        thread_iter = self._iter_threads(self.client_group_list)            
+        for client_thread in thread_iter:
+            client_thread.join()
+
+        if submit_results:
+            results_context = LavaContext(self)
+            thread_iter = self._iter_threads(self.client_group_list)
+            for client_thread in thread_iter:
+                results_context.subcontext_list.append(client_thread.context)
+                    
+            params = submit_results.get('parameters', {})
+            action = self.lava_commands[submit_results['command']](
+                results_context)
+            action.run(**params)
+
+    def init_sync_label(self, host_groups, params):
+        label = params.get('label')
+        if not label:
+            print 'Job Error in sync_to_label action: No label'
+            raise
+        
+        groups = map(self.client_group_list.get, host_groups)
+        size = reduce(add, map(len, groups))
+        self.add_sync_label(label, size)
+
+
+class LavaContext(object):
+    def __init__(self, sync_master):
+        self.test_data = LavaTestData()
+        self.sync_master = sync_master
+        self.subcontext_list = []
+
+    @property
+    def client(self):
+        return self._client
+        
+
+class ActionThunk(object):
+    def __init__(self, action, params, metadata, action_name):
+        self.action = action
+        self.params = params
+        self.metadata = metadata
+        self.action_name = action_name
+        
+    def run(self):
+        self.action.context.test_data.add_metadata(self.metadata)
+        self.action.run(**self.params) 
+       
+        
+class ClientThread(Thread):
+    def __init__(self, hostname, image_type, target_type, sync_master):
+        super(ClientThread, self).__init__()
+        self.action_list = []
+        self.context = LavaContext(sync_master)
+        if image_type == "android":
+            self.context.client_class = LavaAndroidClient
+            self.context.image_type = image_type
+            self.context.hostname = hostname
+        else:
+            # conmux / serial
+            self.context.client_class = LavaClient
+            self.context.hostname = hostname
+        
+    def add_action(self, action_class, action_name, params, metadata):
+        action = action_class(self.context)
+        self.action_list.append(ActionThunk(action, params, metadata, action_name))
+        
+    def run(self):
+        self.context._client = self.context.client_class(self.context.hostname)
+        
         try:
-            for cmd in self.job_data['actions']:
-                params = cmd.get('parameters', {})
-                metadata = cmd.get('metadata', {})
-                metadata['target.hostname'] = self.target
-                self.context.test_data.add_metadata(metadata)
-                action = lava_commands[cmd['command']](self.context)
+            for action in self.action_list:
                 try:
                     status = 'fail'
-                    action.run(**params)
+                    action.run()
                 except CriticalError, err:
-                    raise err
+                    raise
                 except (pexpect.TIMEOUT, GeneralError), err:
                     pass
                 except Exception, err:
@@ -77,45 +211,28 @@
                 else:
                     status = 'pass'
                 finally:
+                    err_msg = ""
+                    command = action.action_name
                     if status == 'fail':
-                        err_msg = "Lava failed at action " + cmd['command'] \
-                            + " with error: " + str(err) + "\n"
-                        if cmd['command'] == 'lava_test_run':
-                            err_msg = err_msg + "Lava failed with test: " \
-                                + test_name
+                        err_msg = "Lava failed at action %s with error: %s\n" %\
+                                  (command, str(err))
+                        if command == 'lava_test_run':
+                            err_msg += "Lava failed on test: %s" %\
+                                       action.params.get('test_name')
                         exc_type, exc_value, exc_traceback = sys.exc_info()
-                        err_msg = err_msg + repr(traceback.format_tb(exc_traceback))
+                        err_msg += repr(traceback.format_tb(exc_traceback))
                         print >> sys.stderr, err_msg
-                    else:
-                        err_msg = ""
-                    self.context.test_data.add_result(cmd['command'], 
+                    self.context.test_data.add_result(action.action_name, 
                         status, err_msg)
         except:
-            #Capture all user-defined and non-user-defined critical errors
+            # Capture all user-defined and non-user-defined critical errors
+            # Do not raise--allow thread to exit gracefully
+            print "Exception in thread for %s. "\
+                "Exiting thread." % self.context.hostname
             self.context.test_data.job_status='fail'
-            raise
-        finally:
-            if submit_results:
-                params = submit_results.get('parameters', {})
-                action = lava_commands[submit_results['command']](
-                    self.context)
-                action.run(**params)
-
-
-class LavaContext(object):
-    def __init__(self, target, image_type):
-        if image_type != "android":
-            self._client = LavaClient(target)
-        else:
-            self._client = LavaAndroidClient(target)
-        self.test_data = LavaTestData()
-
-    @property
-    def client(self):
-        return self._client
-
 
 class LavaTestData(object):
+
     def __init__(self, test_id='lava'):
         self.job_status = 'pass'
         self.metadata = {}

=== modified file 'lava_dispatcher/actions/launch_control.py'
--- lava_dispatcher/actions/launch_control.py	2011-06-27 04:55:08 +0000
+++ lava_dispatcher/actions/launch_control.py	2011-07-22 22:31:22 +0000
@@ -32,8 +32,8 @@
 class cmd_submit_results_on_host(BaseAction):
     def run(self, server, stream):
         xmlrpc_url = "%s/xml-rpc/" % server
-        srv = xmlrpclib.ServerProxy(xmlrpc_url,
-                allow_none=True, use_datetime=True)
+        srv = xmlrpclib.ServerProxy(xmlrpc_url, allow_none=True, 
+                                    use_datetime=True)
 
         client = self.client
         call("cd /tmp/%s/; ls *.bundle > bundle.lst" % LAVA_RESULT_DIR,
@@ -68,16 +68,7 @@
 class cmd_submit_results(BaseAction):
     all_bundles = []
 
-    def run(self, server, stream, result_disk="testrootfs"):
-        """Submit test results to a launch-control server
-        :param server: URL of the launch-control server
-        :param stream: Stream on the launch-control server to save the result to
-        """
-        #Create l-c server connection
-        xmlrpc_url = "%s/xml-rpc/" % server
-        srv = xmlrpclib.ServerProxy(xmlrpc_url,
-                allow_none=True, use_datetime=True)
-
+    def retrieve_remote_results(self, client):
         client = self.client
         try:
             self.in_master_shell()
@@ -133,17 +124,60 @@
 
             self.all_bundles.append(json.loads(content))
 
+    def run(self, server, stream, result_disk="testrootfs", skip_remote=False):
+        """Submit test results to a launch-control server
+        :param server: URL of the launch-control server
+        :param stream: Stream on the launch-control server to save the result to
+        """
+        #Create l-c server connection
+        xmlrpc_url = "%s/xml-rpc/" % server
+        srv = xmlrpclib.ServerProxy(xmlrpc_url,
+                allow_none=True, use_datetime=True)
+
         main_bundle = self.combine_bundles()
-        self.context.test_data.add_seriallog(
-            self.context.client.get_seriallog())
-        main_bundle['test_runs'].append(self.context.test_data.get_test_run())
-        for test_run in main_bundle['test_runs']:
-            attributes = test_run.get('attributes',{})
-            attributes.update(self.context.test_data.get_metadata())
-            test_run['attributes'] = attributes
+        main_test_runs = main_bundle['test_runs']
+        counter=0
+
+        if len(self.context.subcontext_list) > 0:
+            for context in self.context.subcontext_list:
+                if not skip_remote:
+                    self.retrieve_remote_results(context.client)
+                
+                client_bundle = {
+                                     "test_runs": [],
+                                     "format": "Dashboard Bundle Format 1.2"
+                                }
+                context.test_data.add_seriallog(
+                    context.client.get_seriallog())
+                client_bundle['test_runs'].append(context.test_data.get_test_run())
+                for test_run in client_bundle['test_runs']:
+                    attributes = test_run.get('attributes',{})
+                    attributes.update(context.test_data.get_metadata())
+                    test_run['attributes'] = attributes
+                    if len(main_test_runs) == 0:
+                        main_test_runs.append(test_run.copy())
+                        main_test_runs[0]['test_results'] = []
+                        main_test_runs[0]['attachments'] = []
+                    self.merge_thread_run(main_test_runs, 
+                                          test_run, 
+                                          context.client.hostname,
+                                          counter)
+                counter += 1
         json_bundle = json.dumps(main_bundle)
         srv.put(json_bundle, 'lava-dispatcher.bundle', stream)
 
+    def merge_thread_run(self, main_run, new_run, hostname, serial_number):
+        for test_case in new_run['test_results']:
+            new_id = "%s_%s_%d" % (hostname, test_case['test_case_id'], serial_number)
+            new_test_case = test_case
+            new_test_case['test_case_id'] = new_id
+            main_run[0]['test_results'].append(new_test_case)
+            
+        for attachment in new_run['attachments']:
+            new_attachment = attachment
+            new_attachment['pathname'] = "%s_%s_%d" % (hostname, attachment['pathname'], serial_number)
+            main_run[0]['attachments'].append(new_attachment)
+
     def combine_bundles(self):
         if not self.all_bundles:
             return {

=== added file 'lava_dispatcher/actions/shell_commands.py'
--- lava_dispatcher/actions/shell_commands.py	1970-01-01 00:00:00 +0000
+++ lava_dispatcher/actions/shell_commands.py	2011-07-22 22:31:22 +0000
@@ -0,0 +1,42 @@
+# Copyright (C) 2011 Calxeda, Inc.
+#
+# This file is part of LAVA Dispatcher.
+#
+# LAVA Dispatcher is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# LAVA Dispatcher is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, see <http://www.gnu.org/licenses>.
+
+from lava_dispatcher.actions import BaseAction
+from lava_dispatcher.client import OperationFailed
+from lava_dispatcher.config import MASTER_STR
+
+class cmd_scp(BaseAction):
+    
+    def run(self, username, host, filename, target_path):
+        client = self.client
+        cmd = 'scp %s %s@%s:%s' % (filename, username, host, target_path)
+        client.run_shell_command(cmd, MASTER_STR)
+        # TODO XXX: This does not handle authentication issues (password,
+        # new SSH key, changed SSH key, etc.)
+
+
+# TODO XXX: This class should derive from TestAction, as defined in
+# the result-reporting branch, in order to report pass/fail results
+#based on pattern matching
+class cmd_verify_file_present(BaseAction):
+    
+#    fail_patterns = ['No such file or directory']
+    
+    def run(self, file): 
+        client = self.client
+        cmd = 'ls %s' % (file)
+        client.run_shell_command(cmd, MASTER_STR)

=== added file 'lava_dispatcher/actions/sync_to_label.py'
--- lava_dispatcher/actions/sync_to_label.py	1970-01-01 00:00:00 +0000
+++ lava_dispatcher/actions/sync_to_label.py	2011-07-22 22:31:22 +0000
@@ -0,0 +1,22 @@
+# Copyright (C) 2011 Calxeda, Inc.
+#
+# This file is part of LAVA Dispatcher.
+#
+# LAVA Dispatcher is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# LAVA Dispatcher is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, see <http://www.gnu.org/licenses>.
+
+from lava_dispatcher.actions import BaseAction
+
+class cmd_sync_to_label(BaseAction):
+    def run(self, label, block):
+        self.context.sync_master.sync_to(label, block)

=== added file 'lava_dispatcher/sync.py'
--- lava_dispatcher/sync.py	1970-01-01 00:00:00 +0000
+++ lava_dispatcher/sync.py	2011-07-22 22:31:22 +0000
@@ -0,0 +1,47 @@
+# Copyright (C) 2011 Calxeda, Inc.
+#
+# This file is part of LAVA Dispatcher.
+#
+# LAVA Dispatcher is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# LAVA Dispatcher is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, see <http://www.gnu.org/licenses>.
+
+from threading import Event, Lock
+
+class SyncMaster(object):
+    def __init__(self):
+        self.eventlist_lock = Lock()
+        self.eventlist = dict()
+        
+    class EventCount(object):
+        def __init__(self):
+            self.count = 0
+            self.event = Event()
+        
+    def add_sync_label(self, label, count):
+        if not self.eventlist.get(label):
+            self.eventlist[label] = self.EventCount()
+
+        self.eventlist[label].count += count
+
+    def sync_to(self, label, block):
+        self.eventlist_lock.acquire()
+        try:
+            self.eventlist[label].count -= 1
+            if self.eventlist[label].count == 0:
+                self.eventlist[label].event.set()
+        finally:
+            self.eventlist_lock.release()
+        if block:
+            self.eventlist[label].event.wait()
+
+            
\ No newline at end of file

=== modified file 'lava_dispatcher/utils.py'
--- lava_dispatcher/utils.py	2011-06-27 04:55:08 +0000
+++ lava_dispatcher/utils.py	2011-07-22 22:31:22 +0000
@@ -22,6 +22,7 @@
 import shutil
 import urllib2
 import urlparse
+from IPy import IP
 
 from lava_dispatcher.config import LAVA_CACHEDIR
 
@@ -66,3 +67,10 @@
     path = os.path.join(LAVA_CACHEDIR, url_parts.netloc,
         url_parts.path.lstrip(os.sep))
     return path
+
+def ip_addr_range(a, b):
+    int_a = IP(a).int()
+    int_b = IP(b).int()
+
+    return [IP(x) for x in range(int_a, int_b + 1)]
+