Merge lp:~stylesen/lava-scheduler/review-take-1 into lp:lava-scheduler/multinode

[Senthil Kumaran S]

Addressed all review comments from Antonio and removed all legacy code which is no longer required.

[Neil Williams (codehelp)]

Thanks Senthil - tested on multinode.v.l.o & approved.

review: Approve

=== modified file 'lava_scheduler_app/api.py'
--- lava_scheduler_app/api.py	2013-08-16 09:21:16 +0000
+++ lava_scheduler_app/api.py	2013-08-28 13:17:52 +0000
@@ -49,7 +49,7 @@
             job = TestJob.objects.accessible_by_principal(self.user).get(pk=job_id)
         except TestJob.DoesNotExist:
             raise xmlrpclib.Fault(404, "Specified job not found.")
-        if job.target_group:
+        if job.is_multinode:
             return self.submit_job(job.multinode_definition)
         else:
             return self.submit_job(job.definition)
@@ -60,7 +60,7 @@
         job = TestJob.objects.get(pk=job_id)
         if not job.can_cancel(self.user):
             raise xmlrpclib.Fault(403, "Permission denied.")
-        if job.target_group:
+        if job.is_multinode:
             multinode_jobs = TestJob.objects.all().filter(
                 target_group=job.target_group)
             for multinode_job in multinode_jobs:

=== modified file 'lava_scheduler_app/management/commands/schedulermonitor.py'
--- lava_scheduler_app/management/commands/schedulermonitor.py	2012-12-03 05:03:38 +0000
+++ lava_scheduler_app/management/commands/schedulermonitor.py	2013-08-28 13:17:52 +0000
@@ -31,7 +31,7 @@
 
     def handle(self, *args, **options):
         from twisted.internet import reactor
-        from lava_scheduler_daemon.board import Job
+        from lava_scheduler_daemon.job import Job
         daemon_options = self._configure(options)
         source = DatabaseJobSource()
         dispatcher, board_name, json_file = args

=== modified file 'lava_scheduler_app/models.py'
--- lava_scheduler_app/models.py	2013-08-22 05:58:24 +0000
+++ lava_scheduler_app/models.py	2013-08-28 13:17:52 +0000
@@ -640,13 +640,20 @@
 
     @property
     def sub_jobs_list(self):
-        if self.target_group:
+        if self.is_multinode:
             jobs = TestJob.objects.filter(
                 target_group=self.target_group).order_by('id')
             return jobs
         else:
             return None
 
+    @property
+    def is_multinode(self):
+        if self.target_group:
+            return True
+        else:
+            return False
+
 
 class DeviceStateTransition(models.Model):
     created_on = models.DateTimeField(auto_now_add=True)

=== modified file 'lava_scheduler_app/templates/lava_scheduler_app/job_sidebar.html'
--- lava_scheduler_app/templates/lava_scheduler_app/job_sidebar.html	2013-08-22 05:58:24 +0000
+++ lava_scheduler_app/templates/lava_scheduler_app/job_sidebar.html	2013-08-28 13:17:52 +0000
@@ -63,7 +63,7 @@
     <dt>Finished at:</dt>
     <dd>{{ job.end_time|default:"not finished" }}</dd>
 
-    {% if job.target_group %}
+    {% if job.is_multinode %}
     <dt>Sub Jobs:</dt>
     {% for subjob in job.sub_jobs_list %}
     <dd>
@@ -87,7 +87,7 @@
     <li>
         <a href="{% url lava.scheduler.job.definition job.pk %}">Definition</a>
     </li>
-    {% if job.target_group %}
+    {% if job.is_multinode %}
     <li>
         <a href="{% url lava.scheduler.job.multinode_definition job.pk %}"> Multinode Definition</a>
     </li>

=== modified file 'lava_scheduler_app/utils.py'
--- lava_scheduler_app/utils.py	2013-08-23 09:52:50 +0000
+++ lava_scheduler_app/utils.py	2013-08-28 13:17:52 +0000
@@ -45,49 +45,54 @@
     node_json = {}
     all_nodes = {}
     node_actions = {}
-    if "device_group" in json_jobdata:
-        # get all the roles and create node action list for each role.
-        for group in json_jobdata["device_group"]:
-            node_actions[group["role"]] = []
-
-        # Take each action and assign it to proper roles. If roles are not
-        # specified for a specific action, then assign it to all the roles.
-        all_actions = json_jobdata["actions"]
-        for role in node_actions.keys():
-            for action in all_actions:
-                new_action = copy.deepcopy(action)
-                if 'parameters' in new_action \
-                        and 'role' in new_action["parameters"]:
-                    if new_action["parameters"]["role"] == role:
-                        new_action["parameters"].pop('role', None)
-                        node_actions[role].append(new_action)
-                else:
+    
+    # Check if we are operating on multinode job data. Else return the job
+    # data as it is.
+    if "device_group" in json_jobdata and target_group:
+        pass
+    else:
+        return json_jobdata
+
+    # get all the roles and create node action list for each role.
+    for group in json_jobdata["device_group"]:
+        node_actions[group["role"]] = []
+
+    # Take each action and assign it to proper roles. If roles are not
+    # specified for a specific action, then assign it to all the roles.
+    all_actions = json_jobdata["actions"]
+    for role in node_actions.keys():
+        for action in all_actions:
+            new_action = copy.deepcopy(action)
+            if 'parameters' in new_action \
+                    and 'role' in new_action["parameters"]:
+                if new_action["parameters"]["role"] == role:
+                    new_action["parameters"].pop('role', None)
                     node_actions[role].append(new_action)
-
-        group_count = 0
-        for clients in json_jobdata["device_group"]:
-            group_count += int(clients["count"])
-        for clients in json_jobdata["device_group"]:
-            role = str(clients["role"])
-            count = int(clients["count"])
-            node_json[role] = []
-            for c in range(0, count):
-                node_json[role].append({})
-                node_json[role][c]["timeout"] = json_jobdata["timeout"]
-                node_json[role][c]["job_name"] = json_jobdata["job_name"]
-                node_json[role][c]["tags"] = clients["tags"]
-                node_json[role][c]["group_size"] = group_count
-                node_json[role][c]["target_group"] = target_group
-                node_json[role][c]["actions"] = node_actions[role]
-
-                node_json[role][c]["role"] = role
-                # multinode node stage 2
-                node_json[role][c]["logging_level"] = json_jobdata["logging_level"]
-                node_json[role][c]["device_type"] = clients["device_type"]
-
-        return node_json
-
-    return 0
+            else:
+                node_actions[role].append(new_action)
+
+    group_count = 0
+    for clients in json_jobdata["device_group"]:
+        group_count += int(clients["count"])
+    for clients in json_jobdata["device_group"]:
+        role = str(clients["role"])
+        count = int(clients["count"])
+        node_json[role] = []
+        for c in range(0, count):
+            node_json[role].append({})
+            node_json[role][c]["timeout"] = json_jobdata["timeout"]
+            node_json[role][c]["job_name"] = json_jobdata["job_name"]
+            node_json[role][c]["tags"] = clients["tags"]
+            node_json[role][c]["group_size"] = group_count
+            node_json[role][c]["target_group"] = target_group
+            node_json[role][c]["actions"] = node_actions[role]
+
+            node_json[role][c]["role"] = role
+            # multinode node stage 2
+            node_json[role][c]["logging_level"] = json_jobdata["logging_level"]
+            node_json[role][c]["device_type"] = clients["device_type"]
+
+    return node_json
 
 
 def requested_device_count(json_data):
@@ -100,17 +105,13 @@
 
     {'kvm': 1, 'qemu': 3, 'panda': 1}
 
-    If the job is not a multinode job, then return None.
+    If the job is not a multinode job, then return an empty dictionary.
     """
     job_data = simplejson.loads(json_data)
+    requested_devices = {}
     if 'device_group' in job_data:
-        requested_devices = {}
         for device_group in job_data['device_group']:
             device_type = device_group['device_type']
             count = device_group['count']
             requested_devices[device_type] = count
-        return requested_devices
-    else:
-        # TODO: Put logic to check whether we have requested devices attached
-        #       to this lava-server, even if it is a single node job?
-        return None
+    return requested_devices

=== modified file 'lava_scheduler_app/views.py'
--- lava_scheduler_app/views.py	2013-08-27 15:18:24 +0000
+++ lava_scheduler_app/views.py	2013-08-28 13:17:52 +0000
@@ -801,7 +801,7 @@
 def job_cancel(request, pk):
     job = get_restricted_job(request.user, pk)
     if job.can_cancel(request.user):
-        if job.target_group:
+        if job.is_multinode:
             multinode_jobs = TestJob.objects.all().filter(
                 target_group=job.target_group)
             for multinode_job in multinode_jobs:
@@ -826,7 +826,7 @@
     if job.can_resubmit(request.user):
         response_data["is_authorized"] = True
 
-        if job.target_group:
+        if job.is_multinode:
             definition = job.multinode_definition
         else:
             definition = job.definition

=== removed file 'lava_scheduler_daemon/board.py'
--- lava_scheduler_daemon/board.py	2013-08-19 10:44:11 +0000
+++ lava_scheduler_daemon/board.py	1970-01-01 00:00:00 +0000
@@ -1,355 +0,0 @@
-import json
-import os
-import signal
-import tempfile
-import logging
-
-from twisted.internet.error import ProcessDone, ProcessExitedAlready
-from twisted.internet.protocol import ProcessProtocol
-from twisted.internet import defer, task
-
-
-def catchall_errback(logger):
-    def eb(failure):
-        logger.error(
-            '%s: %s\n%s', failure.type.__name__, failure.value,
-            failure.getTraceback())
-    return eb
-
-
-class DispatcherProcessProtocol(ProcessProtocol):
-
-    def __init__(self, deferred, job):
-        self.logger = logging.getLogger(__name__ + '.DispatcherProcessProtocol')
-        self.deferred = deferred
-        self.log_size = 0
-        self.job = job
-
-    def childDataReceived(self, childFD, data):
-        self.log_size += len(data)
-        if self.log_size > self.job.daemon_options['LOG_FILE_SIZE_LIMIT']:
-            if not self.job._killing:
-                self.job.cancel("exceeded log size limit")
-
-    def childConnectionLost(self, childFD):
-        self.logger.info("childConnectionLost for %s: %s",
-                         self.job.board_name, childFD)
-
-    def processExited(self, reason):
-        self.logger.info("processExited for %s: %s",
-                         self.job.board_name, reason.value)
-
-    def processEnded(self, reason):
-        self.logger.info("processEnded for %s: %s",
-                         self.job.board_name, reason.value)
-        self.deferred.callback(reason.value.exitCode)
-
-
-class Job(object):
-
-    def __init__(self, job_data, dispatcher, source, board_name, reactor,
-                 daemon_options):
-        self.job_data = job_data
-        self.dispatcher = dispatcher
-        self.source = source
-        self.board_name = board_name
-        self.logger = logging.getLogger(__name__ + '.Job.' + board_name)
-        self.reactor = reactor
-        self.daemon_options = daemon_options
-        self._json_file = None
-        self._source_lock = defer.DeferredLock()
-        self._checkCancel_call = task.LoopingCall(self._checkCancel)
-        self._signals = ['SIGINT', 'SIGINT', 'SIGTERM', 'SIGTERM', 'SIGKILL']
-        self._time_limit_call = None
-        self._killing = False
-        self._kill_reason = ''
-
-    def _checkCancel(self):
-        if self._killing:
-            self.cancel()
-        else:
-            return self._source_lock.run(
-                self.source.jobCheckForCancellation,
-                self.board_name).addCallback(self._maybeCancel)
-
-    def cancel(self, reason=None):
-        if not self._killing:
-            if reason is None:
-                reason = "killing job for unknown reason"
-            self._kill_reason = reason
-            self.logger.info(reason)
-        self._killing = True
-        if self._signals:
-            signame = self._signals.pop(0)
-        else:
-            self.logger.warning("self._signals is empty!")
-            signame = 'SIGKILL'
-        self.logger.info(
-            'attempting to kill job with signal %s' % signame)
-        try:
-            self._protocol.transport.signalProcess(getattr(signal, signame))
-        except ProcessExitedAlready:
-            pass
-
-    def _maybeCancel(self, cancel):
-        if cancel:
-            self.cancel("killing job by user request")
-        else:
-            logging.debug('not cancelling')
-
-    def _time_limit_exceeded(self):
-        self._time_limit_call = None
-        self.cancel("killing job for exceeding timeout")
-
-    def run(self):
-        d = self.source.getOutputDirForJobOnBoard(self.board_name)
-        return d.addCallback(self._run).addErrback(
-            catchall_errback(self.logger))
-
-    def _run(self, output_dir):
-        d = defer.Deferred()
-        json_data = self.job_data
-        fd, self._json_file = tempfile.mkstemp()
-        with os.fdopen(fd, 'wb') as f:
-            json.dump(json_data, f)
-        self._protocol = DispatcherProcessProtocol(d, self)
-        self.reactor.spawnProcess(
-            self._protocol, self.dispatcher, args=[
-                self.dispatcher, self._json_file, '--output-dir', output_dir],
-            childFDs={0: 0, 1: 'r', 2: 'r'}, env=None)
-        self._checkCancel_call.start(10)
-        timeout = max(
-            json_data['timeout'], self.daemon_options['MIN_JOB_TIMEOUT'])
-        self._time_limit_call = self.reactor.callLater(
-            timeout, self._time_limit_exceeded)
-        d.addBoth(self._exited)
-        return d
-
-    def _exited(self, exit_code):
-        self.logger.info("job finished on %s", self.job_data['target'])
-        if self._json_file is not None:
-            os.unlink(self._json_file)
-        self.logger.info("reporting job completed")
-        if self._time_limit_call is not None:
-            self._time_limit_call.cancel()
-        self._checkCancel_call.stop()
-        return self._source_lock.run(
-            self.source.jobCompleted,
-            self.board_name,
-            exit_code,
-            self._killing).addCallback(
-                lambda r: exit_code)
-
-
-class SchedulerMonitorPP(ProcessProtocol):
-
-    def __init__(self, d, board_name):
-        self.d = d
-        self.board_name = board_name
-        self.logger = logging.getLogger(__name__ + '.SchedulerMonitorPP')
-
-    def childDataReceived(self, childFD, data):
-        self.logger.warning(
-            "scheduler monitor for %s produced output: %r on fd %s",
-            self.board_name, data, childFD)
-
-    def processEnded(self, reason):
-        if not reason.check(ProcessDone):
-            self.logger.error(
-                "scheduler monitor for %s crashed: %s",
-                self.board_name, reason)
-        self.d.callback(None)
-
-
-class MonitorJob(object):
-
-    def __init__(self, job_data, dispatcher, source, board_name, reactor,
-                 daemon_options):
-        self.logger = logging.getLogger(__name__ + '.MonitorJob')
-        self.job_data = job_data
-        self.dispatcher = dispatcher
-        self.source = source
-        self.board_name = board_name
-        self.reactor = reactor
-        self.daemon_options = daemon_options
-        self._json_file = None
-
-    def run(self):
-        d = defer.Deferred()
-        json_data = self.job_data
-        fd, self._json_file = tempfile.mkstemp()
-        with os.fdopen(fd, 'wb') as f:
-            json.dump(json_data, f)
-
-        childFDs = {0: 0, 1: 1, 2: 2}
-        args = [
-            'setsid', 'lava-server', 'manage', 'schedulermonitor',
-            self.dispatcher, str(self.board_name), self._json_file,
-            '-l', self.daemon_options['LOG_LEVEL']]
-        if self.daemon_options['LOG_FILE_PATH']:
-            args.extend(['-f', self.daemon_options['LOG_FILE_PATH']])
-            childFDs = None
-        self.logger.info('executing "%s"', ' '.join(args))
-        self.reactor.spawnProcess(
-            SchedulerMonitorPP(d, self.board_name), 'setsid',
-            childFDs=childFDs, env=None, args=args)
-        d.addBoth(self._exited)
-        return d
-
-    def _exited(self, result):
-        if self._json_file is not None:
-            os.unlink(self._json_file)
-        return result
-
-
-class Board(object):
-    """
-    A board runs jobs.  A board can be in four main states:
-
-     * stopped (S)
-       * the board is not looking for or processing jobs
-     * checking (C)
-       * a call to check for a new job is in progress
-     * waiting (W)
-       * no job was found by the last call to getJobForBoard and so the board
-         is waiting for a while before calling again.
-     * running (R)
-       * a job is running (or a job has completed but the call to jobCompleted
-         on the job source has not)
-
-    In addition, because we can't stop a job instantly nor abort a check for a
-    new job safely (because a if getJobForBoard returns a job, it has already
-    been marked as started), there are variations on the 'checking' and
-    'running' states -- 'checking with stop requested' (C+S) and 'running with
-    stop requested' (R+S).  Even this is a little simplistic as there is the
-    possibility of .start() being called before the process of stopping
-    completes, but we deal with this by deferring any actions taken by
-    .start() until the board is really stopped.
-
-    Events that cause state transitions are:
-
-     * start() is called.  We cheat and pretend that this can only happen in
-       the stopped state by stopping first, and then move into the C state.
-
-     * stop() is called.  If we in the C or R state we move to C+S or R+S
-       resepectively.  If we are in S, C+S or R+S, we stay there.  If we are
-       in W, we just move straight to S.
-
-     * getJobForBoard() returns a job.  We can only be in C or C+S here, and
-       move into R or R+S respectively.
-
-     * getJobForBoard() indicates that there is no job to perform.  Again we
-       can only be in C or C+S and move into W or S respectively.
-
-     * a job completes (i.e. the call to jobCompleted() on the source
-       returns).  We can only be in R or R+S and move to C or S respectively.
-
-     * the timer that being in state W implies expires.  We move into C.
-
-    The cheating around start means that interleaving start and stop calls may
-    not always do what you expect.  So don't mess around in that way please.
-    """
-
-    job_cls = MonitorJob
-
-    def __init__(self, source, board_name, dispatcher, reactor, daemon_options,
-                 job_cls=None):
-        self.source = source
-        self.board_name = board_name
-        self.dispatcher = dispatcher
-        self.reactor = reactor
-        self.daemon_options = daemon_options
-        if job_cls is not None:
-            self.job_cls = job_cls
-        self.running_job = None
-        self._check_call = None
-        self._stopping_deferreds = []
-        self.logger = logging.getLogger(__name__ + '.Board.' + board_name)
-        self.checking = False
-
-    def _state_name(self):
-        if self.running_job:
-            state = "R"
-        elif self._check_call:
-            assert not self._stopping_deferreds
-            state = "W"
-        elif self.checking:
-            state = "C"
-        else:
-            assert not self._stopping_deferreds
-            state = "S"
-        if self._stopping_deferreds:
-            state += "+S"
-        return state
-
-    def start(self):
-        self.logger.debug("start requested")
-        self.stop().addCallback(self._start)
-
-    def _start(self, ignored):
-        self.logger.debug("starting")
-        self._stopping_deferreds = []
-        self._checkForJob()
-
-    def stop(self):
-        self.logger.debug("stopping")
-        if self._check_call is not None:
-            self._check_call.cancel()
-            self._check_call = None
-
-        if self.running_job is not None or self.checking:
-            self.logger.debug("job running; deferring stop")
-            self._stopping_deferreds.append(defer.Deferred())
-            return self._stopping_deferreds[-1]
-        else:
-            self.logger.debug("stopping immediately")
-            return defer.succeed(None)
-
-    def _checkForJob(self):
-        self.logger.debug("checking for job")
-        self._check_call = None
-        self.checking = True
-        self.source.getJobForBoard(self.board_name).addCallbacks(
-            self._maybeStartJob, self._ebCheckForJob)
-
-    def _ebCheckForJob(self, result):
-        self.logger.error(
-            '%s: %s\n%s', result.type.__name__, result.value,
-            result.getTraceback())
-        self._maybeStartJob(None)
-
-    def _finish_stop(self):
-        self.logger.debug(
-            "calling %s deferreds returned from stop()",
-            len(self._stopping_deferreds))
-        for d in self._stopping_deferreds:
-            d.callback(None)
-        self._stopping_deferreds = []
-
-    def _maybeStartJob(self, job_data):
-        self.checking = False
-        if job_data is None:
-            self.logger.debug("no job found")
-            if self._stopping_deferreds:
-                self._finish_stop()
-            else:
-                self._check_call = self.reactor.callLater(
-                    10, self._checkForJob)
-            return
-        self.logger.info("starting job %r", job_data)
-        self.running_job = self.job_cls(
-            job_data, self.dispatcher, self.source, self.board_name,
-            self.reactor, self.daemon_options, None)
-        d = self.running_job.run()
-        d.addCallbacks(self._cbJobFinished, self._ebJobFinished)
-
-    def _ebJobFinished(self, result):
-        self.logger.exception(result.value)
-        self._checkForJob()
-
-    def _cbJobFinished(self, result):
-        self.running_job = None
-        if self._stopping_deferreds:
-            self._finish_stop()
-        else:
-            self._checkForJob()

=== modified file 'lava_scheduler_daemon/dbjobsource.py'
--- lava_scheduler_daemon/dbjobsource.py	2013-08-12 12:45:13 +0000
+++ lava_scheduler_daemon/dbjobsource.py	2013-08-28 13:17:52 +0000
@@ -93,21 +93,6 @@
                 transaction.leave_transaction_management()
         return self.deferToThread(wrapper, *args, **kw)
 
-    def getBoardList_impl(self):
-        self.logger.info("Checking configured devices")
-        configured_boards = [
-            x.hostname for x in dispatcher_config.get_devices()]
-        boards = []
-        for d in configured_boards:
-            self.logger.info("%s is configured" % d.hostname)
-        for d in Device.objects.all():
-            if d.hostname in configured_boards:
-                boards.append({'hostname': d.hostname})
-        return boards
-
-    def getBoardList(self):
-        return self.deferForDB(self.getBoardList_impl)
-
     def _get_health_check_jobs(self):
         """Gets the list of configured boards and checks which are the boards
         that require health check.
@@ -203,7 +188,7 @@
                 continue
             if devices:
                 for d in devices:
-                    self.logger.info("Checking %s" % d.hostname)
+                    self.logger.debug("Checking %s" % d.hostname)
                     if d.hostname in configured_boards:
                        if job:
                            job = self._fix_device(d, job)
@@ -214,7 +199,7 @@
         # target_group are assigned devices.
         final_job_list = copy.deepcopy(job_list)
         for job in job_list:
-            if job.target_group:
+            if job.is_multinode:
                 multinode_jobs = TestJob.objects.all().filter(
                     target_group=job.target_group)
 
@@ -301,65 +286,6 @@
         else:
             return None
 
-    def getJobForBoard_impl(self, board_name):
-        while True:
-            device = Device.objects.get(hostname=board_name)
-            if device.status != Device.IDLE:
-                return None
-            if not device.device_type.health_check_job:
-                run_health_check = False
-            elif device.health_status == Device.HEALTH_UNKNOWN:
-                run_health_check = True
-            elif device.health_status == Device.HEALTH_LOOPING:
-                run_health_check = True
-            elif not device.last_health_report_job:
-                run_health_check = True
-            else:
-                run_health_check = device.last_health_report_job.end_time < datetime.datetime.now() - datetime.timedelta(days=1)
-            if run_health_check:
-                job = self._getHealthCheckJobForBoard(device)
-            else:
-                job = self._getJobFromQueue(device)
-            if job:
-                DeviceStateTransition.objects.create(
-                    created_by=None, device=device, old_state=device.status,
-                    new_state=Device.RUNNING, message=None, job=job).save()
-                job.status = TestJob.RUNNING
-                job.start_time = datetime.datetime.utcnow()
-                job.actual_device = device
-                device.status = Device.RUNNING
-                shutil.rmtree(job.output_dir, ignore_errors=True)
-                device.current_job = job
-                try:
-                    # The unique constraint on current_job may cause this to
-                    # fail in the case of concurrent requests for different
-                    # boards grabbing the same job.  If there are concurrent
-                    # requests for the *same* board they may both return the
-                    # same job -- this is an application level bug though.
-                    device.save()
-                except IntegrityError:
-                    self.logger.info(
-                        "job %s has been assigned to another board -- "
-                        "rolling back", job.id)
-                    transaction.rollback()
-                    continue
-                else:
-                    job.log_file.save(
-                        'job-%s.log' % job.id, ContentFile(''), save=False)
-                    job.submit_token = AuthToken.objects.create(user=job.submitter)
-                    job.save()
-                    json_data = self._get_json_data(job)
-                    transaction.commit()
-                    return json_data
-            else:
-                # _getHealthCheckJobForBoard can offline the board, so commit
-                # in this branch too.
-                transaction.commit()
-                return None
-
-    def getJobForBoard(self, board_name):
-        return self.deferForDB(self.getJobForBoard_impl, board_name)
-
     def getJobDetails_impl(self, job):
         job.status = TestJob.RUNNING
         job.start_time = datetime.datetime.utcnow()

=== modified file 'lava_scheduler_daemon/job.py'
--- lava_scheduler_daemon/job.py	2013-07-17 12:48:53 +0000
+++ lava_scheduler_daemon/job.py	2013-08-28 13:17:52 +0000
@@ -16,13 +16,211 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with LAVA Scheduler.  If not, see <http://www.gnu.org/licenses/>.
 
+import json
+import os
+import signal
+import tempfile
 import logging
 
-from twisted.internet import defer
-from lava_scheduler_daemon.board import MonitorJob
-
-
-class NewJob(object):
+from twisted.internet.error import ProcessDone, ProcessExitedAlready
+from twisted.internet.protocol import ProcessProtocol
+from twisted.internet import defer, task
+
+
+def catchall_errback(logger):
+    def eb(failure):
+        logger.error(
+            '%s: %s\n%s', failure.type.__name__, failure.value,
+            failure.getTraceback())
+    return eb
+
+
+class DispatcherProcessProtocol(ProcessProtocol):
+
+    def __init__(self, deferred, job):
+        self.logger = logging.getLogger(__name__ + '.DispatcherProcessProtocol')
+        self.deferred = deferred
+        self.log_size = 0
+        self.job = job
+
+    def childDataReceived(self, childFD, data):
+        self.log_size += len(data)
+        if self.log_size > self.job.daemon_options['LOG_FILE_SIZE_LIMIT']:
+            if not self.job._killing:
+                self.job.cancel("exceeded log size limit")
+
+    def childConnectionLost(self, childFD):
+        self.logger.info("childConnectionLost for %s: %s",
+                         self.job.board_name, childFD)
+
+    def processExited(self, reason):
+        self.logger.info("processExited for %s: %s",
+                         self.job.board_name, reason.value)
+
+    def processEnded(self, reason):
+        self.logger.info("processEnded for %s: %s",
+                         self.job.board_name, reason.value)
+        self.deferred.callback(reason.value.exitCode)
+
+
+class Job(object):
+
+    def __init__(self, job_data, dispatcher, source, board_name, reactor,
+                 daemon_options):
+        self.job_data = job_data
+        self.dispatcher = dispatcher
+        self.source = source
+        self.board_name = board_name
+        self.logger = logging.getLogger(__name__ + '.Job.' + board_name)
+        self.reactor = reactor
+        self.daemon_options = daemon_options
+        self._json_file = None
+        self._source_lock = defer.DeferredLock()
+        self._checkCancel_call = task.LoopingCall(self._checkCancel)
+        self._signals = ['SIGINT', 'SIGINT', 'SIGTERM', 'SIGTERM', 'SIGKILL']
+        self._time_limit_call = None
+        self._killing = False
+        self._kill_reason = ''
+
+    def _checkCancel(self):
+        if self._killing:
+            self.cancel()
+        else:
+            return self._source_lock.run(
+                self.source.jobCheckForCancellation,
+                self.board_name).addCallback(self._maybeCancel)
+
+    def cancel(self, reason=None):
+        if not self._killing:
+            if reason is None:
+                reason = "killing job for unknown reason"
+            self._kill_reason = reason
+            self.logger.info(reason)
+        self._killing = True
+        if self._signals:
+            signame = self._signals.pop(0)
+        else:
+            self.logger.warning("self._signals is empty!")
+            signame = 'SIGKILL'
+        self.logger.info(
+            'attempting to kill job with signal %s' % signame)
+        try:
+            self._protocol.transport.signalProcess(getattr(signal, signame))
+        except ProcessExitedAlready:
+            pass
+
+    def _maybeCancel(self, cancel):
+        if cancel:
+            self.cancel("killing job by user request")
+        else:
+            logging.debug('not cancelling')
+
+    def _time_limit_exceeded(self):
+        self._time_limit_call = None
+        self.cancel("killing job for exceeding timeout")
+
+    def run(self):
+        d = self.source.getOutputDirForJobOnBoard(self.board_name)
+        return d.addCallback(self._run).addErrback(
+            catchall_errback(self.logger))
+
+    def _run(self, output_dir):
+        d = defer.Deferred()
+        json_data = self.job_data
+        fd, self._json_file = tempfile.mkstemp()
+        with os.fdopen(fd, 'wb') as f:
+            json.dump(json_data, f)
+        self._protocol = DispatcherProcessProtocol(d, self)
+        self.reactor.spawnProcess(
+            self._protocol, self.dispatcher, args=[
+                self.dispatcher, self._json_file, '--output-dir', output_dir],
+            childFDs={0: 0, 1: 'r', 2: 'r'}, env=None)
+        self._checkCancel_call.start(10)
+        timeout = max(
+            json_data['timeout'], self.daemon_options['MIN_JOB_TIMEOUT'])
+        self._time_limit_call = self.reactor.callLater(
+            timeout, self._time_limit_exceeded)
+        d.addBoth(self._exited)
+        return d
+
+    def _exited(self, exit_code):
+        self.logger.info("job finished on %s", self.job_data['target'])
+        if self._json_file is not None:
+            os.unlink(self._json_file)
+        self.logger.info("reporting job completed")
+        if self._time_limit_call is not None:
+            self._time_limit_call.cancel()
+        self._checkCancel_call.stop()
+        return self._source_lock.run(
+            self.source.jobCompleted,
+            self.board_name,
+            exit_code,
+            self._killing).addCallback(
+                lambda r: exit_code)
+
+
+class SchedulerMonitorPP(ProcessProtocol):
+
+    def __init__(self, d, board_name):
+        self.d = d
+        self.board_name = board_name
+        self.logger = logging.getLogger(__name__ + '.SchedulerMonitorPP')
+
+    def childDataReceived(self, childFD, data):
+        self.logger.warning(
+            "scheduler monitor for %s produced output: %r on fd %s",
+            self.board_name, data, childFD)
+
+    def processEnded(self, reason):
+        if not reason.check(ProcessDone):
+            self.logger.error(
+                "scheduler monitor for %s crashed: %s",
+                self.board_name, reason)
+        self.d.callback(None)
+
+
+class MonitorJob(object):
+
+    def __init__(self, job_data, dispatcher, source, board_name, reactor,
+                 daemon_options):
+        self.logger = logging.getLogger(__name__ + '.MonitorJob')
+        self.job_data = job_data
+        self.dispatcher = dispatcher
+        self.source = source
+        self.board_name = board_name
+        self.reactor = reactor
+        self.daemon_options = daemon_options
+        self._json_file = None
+
+    def run(self):
+        d = defer.Deferred()
+        json_data = self.job_data
+        fd, self._json_file = tempfile.mkstemp()
+        with os.fdopen(fd, 'wb') as f:
+            json.dump(json_data, f)
+
+        childFDs = {0: 0, 1: 1, 2: 2}
+        args = [
+            'setsid', 'lava-server', 'manage', 'schedulermonitor',
+            self.dispatcher, str(self.board_name), self._json_file,
+            '-l', self.daemon_options['LOG_LEVEL']]
+        if self.daemon_options['LOG_FILE_PATH']:
+            args.extend(['-f', self.daemon_options['LOG_FILE_PATH']])
+            childFDs = None
+        self.logger.info('executing "%s"', ' '.join(args))
+        self.reactor.spawnProcess(
+            SchedulerMonitorPP(d, self.board_name), 'setsid',
+            childFDs=childFDs, env=None, args=args)
+        d.addBoth(self._exited)
+        return d
+
+    def _exited(self, result):
+        if self._json_file is not None:
+            os.unlink(self._json_file)
+        return result
+
+
+class JobRunner(object):
     job_cls = MonitorJob
 
     def __init__(self, source, job, dispatcher, reactor, daemon_options,
@@ -39,7 +237,7 @@
         if job_cls is not None:
             self.job_cls = job_cls
         self.running_job = None
-        self.logger = logging.getLogger(__name__ + '.NewJob.' + str(job.id))
+        self.logger = logging.getLogger(__name__ + '.JobRunner.' + str(job.id))
 
     def start(self):
         self.logger.debug("processing job")

=== modified file 'lava_scheduler_daemon/service.py'
--- lava_scheduler_daemon/service.py	2013-07-22 12:43:45 +0000
+++ lava_scheduler_daemon/service.py	2013-08-28 13:17:52 +0000
@@ -1,62 +1,28 @@
+# Copyright (C) 2013 Linaro Limited
+#
+# Author: Senthil Kumaran <[e-mail]>
+#
+# This file is part of LAVA Scheduler.
+#
+# LAVA Scheduler is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3 as
+# published by the Free Software Foundation
+#
+# LAVA Scheduler 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 Affero General Public License
+# along with LAVA Scheduler.  If not, see <http://www.gnu.org/licenses/>.
+
 import logging
 
 from twisted.application.service import Service
 from twisted.internet import defer
 from twisted.internet.task import LoopingCall
 
-from lava_scheduler_daemon.board import Board, catchall_errback
-from lava_scheduler_daemon.job import NewJob
-
-
-class BoardSet(Service):
-
-    def __init__(self, source, dispatcher, reactor, daemon_options):
-        self.logger = logging.getLogger(__name__ + '.BoardSet')
-        self.source = source
-        self.boards = {}
-        self.dispatcher = dispatcher
-        self.reactor = reactor
-        self.daemon_options = daemon_options
-        self._update_boards_call = LoopingCall(self._updateBoards)
-        self._update_boards_call.clock = reactor
-
-    def _updateBoards(self):
-        self.logger.debug("Refreshing board list")
-        return self.source.getBoardList().addCallback(
-            self._cbUpdateBoards).addErrback(catchall_errback(self.logger))
-
-    def _cbUpdateBoards(self, board_cfgs):
-        '''board_cfgs is an array of dicts {hostname=name} '''
-        new_boards = {}
-        for board_cfg in board_cfgs:
-            board_name = board_cfg['hostname']
-
-            if board_cfg['hostname'] in self.boards:
-                board = self.boards.pop(board_name)
-                new_boards[board_name] = board
-            else:
-                self.logger.info("Adding board: %s" % board_name)
-                new_boards[board_name] = Board(
-                    self.source, board_name, self.dispatcher, self.reactor,
-                    self.daemon_options)
-                new_boards[board_name].start()
-        for board in self.boards.values():
-            self.logger.info("Removing board: %s" % board.board_name)
-            board.stop()
-        self.boards = new_boards
-
-    def startService(self):
-        self._update_boards_call.start(20)
-
-    def stopService(self):
-        self._update_boards_call.stop()
-        ds = []
-        dead_boards = []
-        for board in self.boards.itervalues():
-            ds.append(board.stop().addCallback(dead_boards.append))
-        self.logger.info(
-            "waiting for %s boards", len(self.boards) - len(dead_boards))
-        return defer.gatherResults(ds)
+from lava_scheduler_daemon.job import JobRunner, catchall_errback
 
 
 class JobQueue(Service):
@@ -77,8 +43,8 @@
 
     def _cbCheckJobs(self, job_list):
         for job in job_list:
-            new_job = NewJob(self.source, job, self.dispatcher, self.reactor,
-                             self.daemon_options)
+            new_job = JobRunner(self.source, job, self.dispatcher,
+                                self.reactor, self.daemon_options)
             self.logger.info("Starting Job: %d " % job.id)
             new_job.start()
 

=== modified file 'lava_scheduler_daemon/tests/test_board.py'
--- lava_scheduler_daemon/tests/test_board.py	2013-07-17 12:48:53 +0000
+++ lava_scheduler_daemon/tests/test_board.py	2013-08-28 13:17:52 +0000
@@ -38,7 +38,7 @@
 
 class TestJob(object):
 
-    def __init__(self, job_data, dispatcher, source, board_name, reactor, options, use_celery):
+    def __init__(self, job_data, dispatcher, source, board_name, reactor, options):
         self.json_data = job_data
         self.dispatcher = dispatcher
         self.reactor = reactor