Merge lp:~stylesen/lava-scheduler/multinode into lp:lava-scheduler/multinode

[Senthil Kumaran S]

Initial bits with a job based scheduler for multi-node support.

We do not accept multi-node jobs yet.

[Neil Williams (codehelp)]

Approved

review: Approve

[lp:~stylesen/lava-scheduler/multinode]

248. By Neil Williams on 2013-06-24

Merge Senthil's initial MultiNode changes for lava-scheduler.

=== modified file 'lava_scheduler_app/api.py'
--- lava_scheduler_app/api.py	2013-05-02 07:35:44 +0000
+++ lava_scheduler_app/api.py	2013-06-20 11:53:35 +0000
@@ -36,7 +36,7 @@
             raise xmlrpclib.Fault(404, "Specified device not found.")
         except DeviceType.DoesNotExist:
             raise xmlrpclib.Fault(404, "Specified device type not found.")
-        return job.id
+        return job
 
     def resubmit_job(self, job_id):
         try:
 
=== modified file 'lava_scheduler_app/management/commands/scheduler.py'
--- lava_scheduler_app/management/commands/scheduler.py	2012-03-27 05:41:09 +0000
+++ lava_scheduler_app/management/commands/scheduler.py	2013-06-20 11:53:35 +0000
@@ -42,7 +42,7 @@
 
         from twisted.internet import reactor
 
-        from lava_scheduler_daemon.service import BoardSet
+        from lava_scheduler_daemon.service import BoardSet, JobQueue
         from lava_scheduler_daemon.dbjobsource import DatabaseJobSource
 
         daemon_options = self._configure(options)
@@ -57,7 +57,7 @@
                 'fake-dispatcher')
         else:
             dispatcher = options['dispatcher']
-        service = BoardSet(
+        service = JobQueue(
             source, dispatcher, reactor, daemon_options=daemon_options)
         reactor.callWhenRunning(service.startService)
         reactor.run()
 
=== added file 'lava_scheduler_app/migrations/0030_auto__add_field_testjob_sub_id.py'
--- lava_scheduler_app/migrations/0030_auto__add_field_testjob_sub_id.py	1970-01-01 00:00:00 +0000
+++ lava_scheduler_app/migrations/0030_auto__add_field_testjob_sub_id.py	2013-06-20 11:53:35 +0000
@@ -0,0 +1,160 @@
+# -*- coding: utf-8 -*-
+import datetime
+from south.db import db
+from south.v2 import SchemaMigration
+from django.db import models
+
+
+class Migration(SchemaMigration):
+
+    def forwards(self, orm):
+        # Adding field 'TestJob.sub_id'
+        db.add_column('lava_scheduler_app_testjob', 'sub_id',
+                      self.gf('django.db.models.fields.CharField')(default='', max_length=200, blank=True),
+                      keep_default=False)
+
+
+    def backwards(self, orm):
+        # Deleting field 'TestJob.sub_id'
+        db.delete_column('lava_scheduler_app_testjob', 'sub_id')
+
+
+    models = {
+        'auth.group': {
+            'Meta': {'object_name': 'Group'},
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
+            'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
+        },
+        'auth.permission': {
+            'Meta': {'ordering': "('content_type__app_label', 'content_type__model', 'codename')", 'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
+            'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
+        },
+        'auth.user': {
+            'Meta': {'object_name': 'User'},
+            'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
+            'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
+            'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
+            'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'symmetrical': 'False', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
+            'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
+            'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
+            'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
+            'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'}),
+            'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
+        },
+        'contenttypes.contenttype': {
+            'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
+            'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
+        },
+        'dashboard_app.bundle': {
+            'Meta': {'ordering': "['-uploaded_on']", 'object_name': 'Bundle'},
+            '_gz_content': ('django.db.models.fields.files.FileField', [], {'max_length': '100', 'null': 'True', 'db_column': "'gz_content'"}),
+            '_raw_content': ('django.db.models.fields.files.FileField', [], {'max_length': '100', 'null': 'True', 'db_column': "'content'"}),
+            'bundle_stream': ('django.db.models.fields.related.ForeignKey', [], {'related_name': "'bundles'", 'to': "orm['dashboard_app.BundleStream']"}),
+            'content_filename': ('django.db.models.fields.CharField', [], {'max_length': '256'}),
+            'content_sha1': ('django.db.models.fields.CharField', [], {'max_length': '40', 'unique': 'True', 'null': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_deserialized': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'uploaded_by': ('django.db.models.fields.related.ForeignKey', [], {'blank': 'True', 'related_name': "'uploaded_bundles'", 'null': 'True', 'to': "orm['auth.User']"}),
+            'uploaded_on': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.utcnow'})
+        },
+        'dashboard_app.bundlestream': {
+            'Meta': {'object_name': 'BundleStream'},
+            'group': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.Group']", 'null': 'True', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_anonymous': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'is_public': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '64', 'blank': 'True'}),
+            'pathname': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '128'}),
+            'slug': ('django.db.models.fields.CharField', [], {'max_length': '64', 'blank': 'True'}),
+            'user': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']", 'null': 'True', 'blank': 'True'})
+        },
+        'lava_scheduler_app.device': {
+            'Meta': {'object_name': 'Device'},
+            'current_job': ('django.db.models.fields.related.ForeignKey', [], {'related_name': "'+'", 'null': 'True', 'on_delete': 'models.SET_NULL', 'to': "orm['lava_scheduler_app.TestJob']", 'blank': 'True', 'unique': 'True'}),
+            'device_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['lava_scheduler_app.DeviceType']"}),
+            'device_version': ('django.db.models.fields.CharField', [], {'default': 'None', 'max_length': '200', 'null': 'True', 'blank': 'True'}),
+            'health_status': ('django.db.models.fields.IntegerField', [], {'default': '0'}),
+            'hostname': ('django.db.models.fields.CharField', [], {'max_length': '200', 'primary_key': 'True'}),
+            'last_health_report_job': ('django.db.models.fields.related.ForeignKey', [], {'related_name': "'+'", 'null': 'True', 'on_delete': 'models.SET_NULL', 'to': "orm['lava_scheduler_app.TestJob']", 'blank': 'True', 'unique': 'True'}),
+            'status': ('django.db.models.fields.IntegerField', [], {'default': '1'}),
+            'tags': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['lava_scheduler_app.Tag']", 'symmetrical': 'False', 'blank': 'True'})
+        },
+        'lava_scheduler_app.devicestatetransition': {
+            'Meta': {'object_name': 'DeviceStateTransition'},
+            'created_by': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']", 'null': 'True', 'on_delete': 'models.SET_NULL', 'blank': 'True'}),
+            'created_on': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'blank': 'True'}),
+            'device': ('django.db.models.fields.related.ForeignKey', [], {'related_name': "'transitions'", 'to': "orm['lava_scheduler_app.Device']"}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'job': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['lava_scheduler_app.TestJob']", 'null': 'True', 'on_delete': 'models.SET_NULL', 'blank': 'True'}),
+            'message': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
+            'new_state': ('django.db.models.fields.IntegerField', [], {}),
+            'old_state': ('django.db.models.fields.IntegerField', [], {})
+        },
+        'lava_scheduler_app.devicetype': {
+            'Meta': {'object_name': 'DeviceType'},
+            'display': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
+            'health_check_job': ('django.db.models.fields.TextField', [], {'default': 'None', 'null': 'True', 'blank': 'True'}),
+            'name': ('django.db.models.fields.SlugField', [], {'max_length': '50', 'primary_key': 'True'})
+        },
+        'lava_scheduler_app.jobfailuretag': {
+            'Meta': {'object_name': 'JobFailureTag'},
+            'description': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '256'})
+        },
+        'lava_scheduler_app.tag': {
+            'Meta': {'object_name': 'Tag'},
+            'description': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.SlugField', [], {'unique': 'True', 'max_length': '50'})
+        },
+        'lava_scheduler_app.testjob': {
+            'Meta': {'object_name': 'TestJob'},
+            '_results_bundle': ('django.db.models.fields.related.OneToOneField', [], {'null': 'True', 'db_column': "'results_bundle_id'", 'on_delete': 'models.SET_NULL', 'to': "orm['dashboard_app.Bundle']", 'blank': 'True', 'unique': 'True'}),
+            '_results_link': ('django.db.models.fields.CharField', [], {'default': 'None', 'max_length': '400', 'null': 'True', 'db_column': "'results_link'", 'blank': 'True'}),
+            'actual_device': ('django.db.models.fields.related.ForeignKey', [], {'default': 'None', 'related_name': "'+'", 'null': 'True', 'blank': 'True', 'to': "orm['lava_scheduler_app.Device']"}),
+            'definition': ('django.db.models.fields.TextField', [], {}),
+            'description': ('django.db.models.fields.CharField', [], {'default': 'None', 'max_length': '200', 'null': 'True', 'blank': 'True'}),
+            'end_time': ('django.db.models.fields.DateTimeField', [], {'null': 'True', 'blank': 'True'}),
+            'failure_comment': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
+            'failure_tags': ('django.db.models.fields.related.ManyToManyField', [], {'symmetrical': 'False', 'related_name': "'failure_tags'", 'blank': 'True', 'to': "orm['lava_scheduler_app.JobFailureTag']"}),
+            'group': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.Group']", 'null': 'True', 'blank': 'True'}),
+            'health_check': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_public': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'log_file': ('django.db.models.fields.files.FileField', [], {'default': 'None', 'max_length': '100', 'null': 'True', 'blank': 'True'}),
+            'priority': ('django.db.models.fields.IntegerField', [], {'default': '50'}),
+            'requested_device': ('django.db.models.fields.related.ForeignKey', [], {'default': 'None', 'related_name': "'+'", 'null': 'True', 'blank': 'True', 'to': "orm['lava_scheduler_app.Device']"}),
+            'requested_device_type': ('django.db.models.fields.related.ForeignKey', [], {'default': 'None', 'related_name': "'+'", 'null': 'True', 'blank': 'True', 'to': "orm['lava_scheduler_app.DeviceType']"}),
+            'start_time': ('django.db.models.fields.DateTimeField', [], {'null': 'True', 'blank': 'True'}),
+            'status': ('django.db.models.fields.IntegerField', [], {'default': '0'}),
+            'sub_id': ('django.db.models.fields.CharField', [], {'max_length': '200', 'blank': 'True'}),
+            'submit_time': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'blank': 'True'}),
+            'submit_token': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['linaro_django_xmlrpc.AuthToken']", 'null': 'True', 'on_delete': 'models.SET_NULL', 'blank': 'True'}),
+            'submitter': ('django.db.models.fields.related.ForeignKey', [], {'related_name': "'+'", 'to': "orm['auth.User']"}),
+            'tags': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['lava_scheduler_app.Tag']", 'symmetrical': 'False', 'blank': 'True'}),
+            'user': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']", 'null': 'True', 'blank': 'True'})
+        },
+        'linaro_django_xmlrpc.authtoken': {
+            'Meta': {'object_name': 'AuthToken'},
+            'created_on': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'}),
+            'description': ('django.db.models.fields.TextField', [], {'default': "''", 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'last_used_on': ('django.db.models.fields.DateTimeField', [], {'null': 'True'}),
+            'secret': ('django.db.models.fields.CharField', [], {'default': "'ynu6yihw337isktsrzd0ocr83r0huox1b5y4qs1c0ktat7s3089d2xgiz3ll0n68fr6q026mep0t5xwg1coxnl2aoknolgowx2779uluan8quez0row0jnk5j2qsxlle'", 'unique': 'True', 'max_length': '128'}),
+            'user': ('django.db.models.fields.related.ForeignKey', [], {'related_name': "'auth_tokens'", 'to': "orm['auth.User']"})
+        }
+    }
+
+    complete_apps = ['lava_scheduler_app']
 \ No newline at end of file
 
=== modified file 'lava_scheduler_app/models.py'
--- lava_scheduler_app/models.py	2013-02-10 21:26:06 +0000
+++ lava_scheduler_app/models.py	2013-06-20 11:53:35 +0000
@@ -2,6 +2,7 @@
 import os
 import simplejson
 import urlparse
+import copy
 
 from django.conf import settings
 from django.contrib.auth.models import User
@@ -249,6 +250,12 @@
 
     id = models.AutoField(primary_key=True)
 
+    sub_id = models.CharField(
+        verbose_name = _(u"Sub ID"),
+        blank = True,
+        max_length = 200
+        )
+
     submitter = models.ForeignKey(
         User,
         verbose_name = _(u"Submitter"),
@@ -398,9 +405,13 @@
         elif 'device_type' in job_data:
             target = None
             device_type = DeviceType.objects.get(name=job_data['device_type'])
+        elif 'device_group' in job_data:
+            target = None
+            device_type = None 
         else:
             raise JSONDataError(
-                "Neither 'target' nor 'device_type' found in job data.")
+                "No 'target' or 'device_type' or 'device_group' are found "
+                "in job data.")
 
         priorities = dict([(j.upper(), i) for i, j in cls.PRIORITY_CHOICES])
         priority = cls.MEDIUM
@@ -463,15 +474,14 @@
                 tags.append(Tag.objects.get(name=tag_name))
             except Tag.DoesNotExist:
                 raise JSONDataError("tag %r does not exist" % tag_name)
+
         job = TestJob(
             definition=json_data, submitter=submitter,
             requested_device=target, requested_device_type=device_type,
             description=job_name, health_check=health_check, user=user,
             group=group, is_public=is_public, priority=priority)
         job.save()
-        for tag in tags:
-            job.tags.add(tag)
-        return job
+        return job.id
 
     def _can_admin(self, user):
         """ used to check for things like if the user can cancel or annotate
 
=== modified file 'lava_scheduler_app/views.py'
--- lava_scheduler_app/views.py	2013-05-02 09:00:04 +0000
+++ lava_scheduler_app/views.py	2013-06-20 11:53:35 +0000
@@ -74,10 +74,16 @@
 
 
 def pklink(record):
+    job_id = record.pk
+    try:
+        if record.sub_id:
+            job_id = record.sub_id
+    except:
+        pass
     return mark_safe(
         '<a href="%s">%s</a>' % (
             record.get_absolute_url(),
-            escape(record.pk)))
+            escape(job_id)))
 
 
 class IDLinkColumn(Column):
@@ -100,13 +106,13 @@
 
 
 def all_jobs_with_device_sort():
-    return TestJob.objects.select_related(
+    jobs = TestJob.objects.select_related(
         "actual_device", "requested_device", "requested_device_type",
         "submitter", "user", "group").extra(
         select={
             'device_sort': 'coalesce(actual_device_id, requested_device_id, requested_device_type_id)'
             }).all()
-
+    return jobs.order_by('submit_time')
 
 
 class JobTable(DataTablesTable):
@@ -126,7 +132,7 @@
         else:
             return ''
 
-    id = RestrictedIDLinkColumn()
+    sub_id = RestrictedIDLinkColumn()
     status = Column()
     priority = Column()
     device = Column(accessor='device_sort')
@@ -137,7 +143,7 @@
     duration = Column()
 
     datatable_opts = {
-        'aaSorting': [[0, 'desc']],
+        'aaSorting': [[6, 'desc']],
         }
     searchable_columns=['description']
 
 
=== modified file 'lava_scheduler_daemon/dbjobsource.py'
--- lava_scheduler_daemon/dbjobsource.py	2013-06-13 14:53:25 +0000
+++ lava_scheduler_daemon/dbjobsource.py	2013-06-20 11:53:35 +0000
@@ -103,6 +103,68 @@
     def getBoardList(self):
         return self.deferForDB(self.getBoardList_impl)
 
+    def _fix_device(self, device, job):
+        """Associate an available/idle DEVICE to the given JOB.
+
+        Returns the job with actual_device set to DEVICE.
+
+        If we are unable to grab the DEVICE then we return None.
+        """
+        DeviceStateTransition.objects.create(
+            created_by=None, device=device, old_state=device.status,
+            new_state=Device.RUNNING, message=None, job=job).save()
+        device.status = Device.RUNNING
+        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()
+            return None
+        else:
+            job.actual_device = device
+            job.save()
+            transaction.commit()
+        return job
+        
+    def getJobList_impl(self):
+        jobs = TestJob.objects.all().filter(status=TestJob.SUBMITTED)
+        job_list = []
+        devices = None
+
+        for job in jobs:
+            if job.actual_device:
+                job_list.append(job)
+            elif job.requested_device:
+                self.logger.info("Checking Requested Device")
+                devices = Device.objects.all().filter(
+                    hostname=job.requested_device.hostname,
+                    status=Device.IDLE)
+            elif job.requested_device_type:
+                self.logger.info("Checking Requested Device Type")
+                devices = Device.objects.all().filter(
+                    device_type=job.requested_device_type,
+                    status=Device.IDLE)
+            else:
+                continue
+            if devices:
+                device = devices[0]
+                job = self._fix_device(device, job)
+                if job:
+                    job_list.append(job)
+
+        return job_list
+
+    def getJobList(self):
+        return self.deferForDB(self.getJobList_impl)
+
     def _get_json_data(self, job):
         json_data = simplejson.loads(job.definition)
         json_data['target'] = job.actual_device.hostname
@@ -229,6 +291,20 @@
     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()
+        shutil.rmtree(job.output_dir, ignore_errors=True)
+        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
+
+    def getJobDetails(self, job):
+        return self.deferForDB(self.getJobDetails_impl, job)
+
     def getOutputDirForJobOnBoard_impl(self, board_name):
         device = Device.objects.get(hostname=board_name)
         job = device.current_job
 
=== added file 'lava_scheduler_daemon/job.py'
--- lava_scheduler_daemon/job.py	1970-01-01 00:00:00 +0000
+++ lava_scheduler_daemon/job.py	2013-06-20 11:53:35 +0000
@@ -0,0 +1,79 @@
+# 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.internet import defer
+from lava_scheduler_daemon.board import MonitorJob, catchall_errback
+
+class NewJob(object):
+    job_cls = MonitorJob
+
+    def __init__(self, source, job, dispatcher, reactor, daemon_options,
+                 job_cls=None):
+        self.source = source
+        self.dispatcher = dispatcher
+        self.reactor = reactor
+        self.daemon_options = daemon_options
+        self.job = job
+        self.board_name = job.actual_device.hostname
+        if job_cls is not None:
+            self.job_cls = job_cls
+        self.running_job = None
+        self.logger = logging.getLogger(__name__ + '.NewJob.' + str(job.id))
+
+    def start(self):
+        self.logger.debug("processing job")
+        if self.job is None:
+            self.logger.debug("no job found for processing")
+            return
+        self.source.getJobDetails(self.job).addCallbacks(
+            self._startJob, self._ebStartJob)
+
+    def _startJob(self, job_data):
+        if job_data is None:
+            self.logger.debug("no job found")
+            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)
+        d = self.running_job.run()
+        d.addCallbacks(self._cbJobFinished, self._ebJobFinished)
+
+    def _ebStartJob(self, result):
+        self.logger.error(
+            '%s: %s\n%s', result.type.__name__, result.value,
+            result.getTraceback())
+        return
+
+    def stop(self):
+        self.logger.debug("stopping")
+
+        if self.running_job is not None:
+            self.logger.debug("job running; deferring stop")
+        else:
+            self.logger.debug("stopping immediately")
+            return defer.succeed(None)
+
+    def _ebJobFinished(self, result):
+        self.logger.exception(result.value)
+
+    def _cbJobFinished(self, result):
+        self.running_job = None
 
=== modified file 'lava_scheduler_daemon/service.py'
--- lava_scheduler_daemon/service.py	2012-12-03 05:03:38 +0000
+++ lava_scheduler_daemon/service.py	2013-06-20 11:53:35 +0000
@@ -5,6 +5,7 @@
 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):
@@ -56,3 +57,35 @@
         self.logger.info(
             "waiting for %s boards", len(self.boards) - len(dead_boards))
         return defer.gatherResults(ds)
+
+
+class JobQueue(Service):
+
+    def __init__(self, source, dispatcher, reactor, daemon_options):
+        self.logger = logging.getLogger(__name__ + '.JobQueue')
+        self.source = source
+        self.dispatcher = dispatcher
+        self.reactor = reactor
+        self.daemon_options = daemon_options
+        self._check_job_call = LoopingCall(self._checkJobs)
+        self._check_job_call.clock = reactor
+
+    def _checkJobs(self):
+        self.logger.debug("Refreshing jobs")
+        return self.source.getJobList().addCallback(
+            self._cbCheckJobs).addErrback(catchall_errback(self.logger))
+
+    def _cbCheckJobs(self, job_list):
+        for job in job_list:
+            self.logger.debug("Found job: %d" % job.id)
+            new_job = NewJob(self.source, job, self.dispatcher, self.reactor,
+                             self.daemon_options)
+            self.logger.info("Starting Job: %d " % job.id)
+            new_job.start()
+
+    def startService(self):
+        self._check_job_call.start(20)
+
+    def stopService(self):
+        self._check_job_call.stop()
+        return None