Merge lp:~mirsad-vojnikovic-deactivatedaccount/lava-dispatcher/scheduler-daemon into lp:lava-dispatcher

[Mirsad Vojnikovic]

Here is a daemon proposal, very modest actually. It takes one job definition from TestJob table and executes it in a dispatcher class I created for testing purposes. This dispatcher class will not do anything special, save print the action names and wait 10 seconds before going on. It is rather easy to modify to do the real job. I think this is a good starter for the daemon job which is in front of us.

Next step is to put more jobs as threads, while updating status of device and test job.

[Mirsad Vojnikovic (mirsad-vojnikovic-deactivatedaccount)]

Sorry, forgot to mention how to run the daemon:

 * 'schedulerd.py start' - start
 * 'schedulerd.py stop' - stop
 * 'schedulerd.py debug' - debug mode, printing messages to terminal

[Spring Zhang (qzhang)]

110 + message = "pidfile %s does not exist. Daemon not running?\n"
111 + sys.stderr.write(message % self.pidfile)
It's better to place message in one line, or it seems the message is split.Like:
message = "pidfile %s does not exist. Daemon not running?\n" %self.pidfile

It seems your alignment is by 8 spaces/<tab>, I remember it is 4 spaces. Can Paul confirm?

155 + self.connection = sqlite3.connect('./scheduler/database.db')
Can it be a constant variable for './scheduler/database.db'? Not hardcode in function

71 + file(self.pidfile,'w+').write("%s\n" % pid)
The file object needs to close?

150 + def __init__(self):
151 + self.connection = None
152 + self.cursor = None
153 +
154 + def connect_db(self):
155 + self.connection = sqlite3.connect('./scheduler/database.db')
156 + self.cursor = self.connection.cursor()
Can the two methods merge to __init__()?

207 + db_helper = DbHelper()
And can the database name call from caller by a parameter, not specify in DbHelper class?

188 + #params = cmd.get('parameters', {})
189 + #step = lava_commands[cmd['command']](target)
190 + #step.run(**params)
191 +
192 +
Please delete unused statement.

You can use the dispatcher.py from lp:lava, just merge from lp:lava in your branch.

[Mirsad Vojnikovic (mirsad-vojnikovic-deactivatedaccount)]

Spring, thanks for your comments, but I have decided to drop this proposal since it lacks some important stuff. I will try to reuse your comments for the next proposal instead.

[lp:~mirsad-vojnikovic-deactivatedaccount/lava-dispatcher/scheduler-daemon]

22. By Paul Larson on 2011-04-01

add a dispatcher.run_job() to actually do the work, so that it can be
called from the outside by things like the scheduler

23. By Paul Larson on 2011-04-07

Use virtualenv to install the python dependencies and reorganize
entrypoints a bit.

24. By Paul Larson on 2011-04-07

enable login with openid

25. By Paul Larson on 2011-04-08

Hi, this branch enables the admin interface for the scheduler
application. Creating a superuser for yourself locally when using OpenID
is a bit tedious, but I've included a helper and instructions how to do
it in the README.

26. By Paul Larson on 2011-04-08

update top level structure of lava

27. By Paul Larson on 2011-04-08

Merge initial support for submitting results to launch control

28. By Michael Hudson-Doyle on 2011-04-11

small README updates

29. By Michael Hudson-Doyle on 2011-04-11

fix ROOT_URLCONF

30. By Michael Hudson-Doyle on 2011-04-12

some more clues for doc/QUICKSTART

31. By Paul Larson on 2011-04-12

allow graphical tests to run properly by exporting the display on the
client

32. By Michael Hudson-Doyle on 2011-04-12

mention some packages that need to be installed

33. By Paul Larson on 2011-04-12

Fix make to work under lucid too

34. By Michael Hudson-Doyle on 2011-04-15

borrow style and templates from launch-control

merges ~mirsad-vojnikovic/lava/scheduler-l-c-style with a tweak from me

35. By Michael Hudson-Doyle on 2011-04-15

this seems to be needed

36. By Paul Larson on 2011-04-15

results need to be searched for in /tmp

37. By Mirsad <mirsad@ubuntu1004desktop> on 2011-04-18

Updated daemon for latest comments. Added tags field to Device table. Changed to GPL2+

Unmerged revisions

37. By Mirsad <mirsad@ubuntu1004desktop> on 2011-04-18

Updated daemon for latest comments. Added tags field to Device table. Changed to GPL2+

=== added file 'lava-schedulerd'
--- lava-schedulerd	1970-01-01 00:00:00 +0000
+++ lava-schedulerd	2011-04-05 21:11:05 +0000
@@ -0,0 +1,100 @@
+#!/usr/bin/env python
+
+import sys, time, json
+from scheduler.daemon.daemon import Daemon
+from scheduler.daemon.helpers import DbHelper, TestResult
+from scheduler.daemon.async_job import AsyncJob
+from dispatcher import run_job
+
+class SchedulerDaemon(Daemon):
+    def run(self):
+        db_helper = DbHelper()
+        
+        while True:
+            #Pick up next job
+            db_helper.connect_db()
+            next_job = db_helper.get_next_job()
+
+            if next_job != None:
+                #Prepare TestResult object
+                test_result = TestResult(
+                    job_id = next_job['id'],
+                    device_id = next_job['target_id'],
+                    result = 0
+                )
+
+                #Initiate async job
+                job = AsyncJob(
+                    target = worker, 
+                    callback = callback, 
+                    timeout = next_job['timeout'],
+                    timeout_callback = timeout_callback,
+                    test_result = test_result,
+                    args = (next_job['definition'],)
+                )
+
+                #Set device to RUNNING state
+                db_helper.set_device_state(next_job['target_id'], 2)
+                #Set job to RUNNING state
+                db_helper.set_job_state(next_job['id'], 1)
+
+                #Start job
+                job.start()
+
+            db_helper.disconnect_db()
+            time.sleep(1)
+
+def timeout_callback():
+    """
+    Callback to handle job timeout
+    """
+    #What to do here?
+    #Update job, device status
+    pass
+
+def callback(test_result=None):
+    """
+    Callback to handle normal job ending
+    """
+    #Connect to database
+    db_helper = DbHelper()
+    db_helper.connect_db()
+
+    #Set device to IDLE state
+    db_helper.set_device_state(test_result.device_id, 1)
+    #Set job to COMPLETE state
+    db_helper.set_job_state(test_result.job_id, 2)
+
+    #Disconnect from database
+    db_helper.disconnect_db()
+
+def worker(next_job, q, test_result):
+    """
+    Worker function running jobs
+    """
+    #Missing result from dispatcher here
+    run_job(json.loads(next_job))
+
+    test_result.result = 1
+    q.put(test_result)
+
+
+if __name__ == "__main__":
+    daemon = SchedulerDaemon('/tmp/scheduler-daemon.pid')
+    if len(sys.argv) == 2:
+        if 'start' == sys.argv[1]:
+            daemon.start()
+        elif 'stop' == sys.argv[1]:
+            daemon.stop()
+        elif 'restart' == sys.argv[1]:
+            daemon.restart()
+        elif 'debug' == sys.argv[1]:
+            #Print to terminal, debug purpose only!
+            daemon.run()
+        else:
+            print "Unknown command"
+            sys.exit(2)
+        sys.exit(0)
+    else:
+        print "usage: %s start|stop|restart" % sys.argv[0]
+        sys.exit(2)

=== added directory 'scheduler/daemon'
=== added file 'scheduler/daemon/__init__.py'
=== added file 'scheduler/daemon/async_job.py'
--- scheduler/daemon/async_job.py	1970-01-01 00:00:00 +0000
+++ scheduler/daemon/async_job.py	2011-04-05 21:11:05 +0000
@@ -0,0 +1,33 @@
+from multiprocessing import Queue, Process
+
+class AsyncJob(object):
+    def __init__(self, target=None, callback=None, timeout=None,
+        timeout_callback=None, test_result=None, args=(), kwargs={}):
+       
+        self.target = target
+        self.callback = callback
+        self.timeout = timeout
+        self.timeout_callback = timeout_callback
+        self.test_result = test_result
+        self.args = args
+        self.kwargs = kwargs
+
+    def start(self):
+        self.proc = Process(target=self._proxy)
+        self.proc.start()
+
+    def _proxy(self):
+        # Get result from worker via Queue
+        q = Queue()
+        
+        p = Process(target=self.target, args=(self.args[0], q, self.test_result))
+        p.start()
+        p.join(timeout=self.timeout)
+        
+        if p.is_alive():
+            p.terminate()
+            if self.timeout_callback:
+                self.timeout_callback()
+        else:
+            if self.callback:
+                self.callback(q.get())

=== added file 'scheduler/daemon/daemon.py'
--- scheduler/daemon/daemon.py	1970-01-01 00:00:00 +0000
+++ scheduler/daemon/daemon.py	2011-04-05 21:11:05 +0000
@@ -0,0 +1,133 @@
+#!/usr/bin/env python
+"""
+Python daemon from:
+http://www.jejik.com/articles/2007/02/a_simple_unix_linux_daemon_in_python/
+"""
+
+import sys, os, time, atexit
+from signal import SIGTERM 
+
+class Daemon:
+    """
+    A generic daemon class.
+    
+    Usage: subclass the Daemon class and override the run() method
+    """
+    def __init__(self, pidfile, stdin='/dev/null', stdout='/dev/null', stderr='/dev/null'):
+        self.stdin = stdin
+        self.stdout = stdout
+        self.stderr = stderr
+        self.pidfile = pidfile
+    
+    def daemonize(self):
+        """
+        do the UNIX double-fork magic, see Stevens' "Advanced 
+        Programming in the UNIX Environment" for details (ISBN 0201563177)
+        http://www.erlenstar.demon.co.uk/unix/faq_2.html#SEC16
+        """
+        try: 
+            pid = os.fork() 
+            if pid > 0:
+                # exit first parent
+                sys.exit(0) 
+        except OSError, e: 
+            sys.stderr.write("fork #1 failed: %d (%s)\n" % (e.errno, e.strerror))
+            sys.exit(1)
+    
+        # decouple from parent environment
+        os.chdir("/") 
+        os.setsid() 
+        os.umask(0) 
+    
+        # do second fork
+        try: 
+            pid = os.fork() 
+            if pid > 0:
+                # exit from second parent
+                sys.exit(0) 
+        except OSError, e: 
+            sys.stderr.write("fork #2 failed: %d (%s)\n" % (e.errno, e.strerror))
+            sys.exit(1) 
+    
+        # redirect standard file descriptors
+        sys.stdout.flush()
+        sys.stderr.flush()
+        si = file(self.stdin, 'r')
+        so = file(self.stdout, 'a+')
+        se = file(self.stderr, 'a+', 0)
+        os.dup2(si.fileno(), sys.stdin.fileno())
+        os.dup2(so.fileno(), sys.stdout.fileno())
+        os.dup2(se.fileno(), sys.stderr.fileno())
+    
+        # write pidfile
+        atexit.register(self.delpid)
+        pid = str(os.getpid())
+        file(self.pidfile,'w+').write("%s\n" % pid)
+    
+    def delpid(self):
+        os.remove(self.pidfile)
+
+    def start(self):
+        """
+        Start the daemon
+        """
+        # Check for a pidfile to see if the daemon already runs
+        try:
+            pf = file(self.pidfile,'r')
+            pid = int(pf.read().strip())
+            pf.close()
+        except IOError:
+            pid = None
+    
+        if pid:
+            message = "pidfile %s already exist. Daemon already running?\n"
+            sys.stderr.write(message % self.pidfile)
+            sys.exit(1)
+        
+        # Start the daemon
+        self.daemonize()
+        self.run()
+
+    def stop(self):
+        """
+        Stop the daemon
+        """
+        # Get the pid from the pidfile
+        try:
+            pf = file(self.pidfile,'r')
+            pid = int(pf.read().strip())
+            pf.close()
+        except IOError:
+            pid = None
+    
+        if not pid:
+            message = "pidfile %s does not exist. Daemon not running?\n"
+            sys.stderr.write(message % self.pidfile)
+            return # not an error in a restart
+
+        # Try killing the daemon process	
+        try:
+            while 1:
+                os.kill(pid, SIGTERM)
+                time.sleep(0.1)
+        except OSError, err:
+            err = str(err)
+            if err.find("No such process") > 0:
+                if os.path.exists(self.pidfile):
+                    os.remove(self.pidfile)
+            else:
+                print str(err)
+                sys.exit(1)
+
+    def restart(self):
+        """
+        Restart the daemon
+        """
+        self.stop()
+        self.start()
+
+    def run(self):
+        """
+        You should override this method when you subclass Daemon. It will be called after the process has been
+        daemonized by start() or restart().
+        """

=== added file 'scheduler/daemon/helpers.py'
--- scheduler/daemon/helpers.py	1970-01-01 00:00:00 +0000
+++ scheduler/daemon/helpers.py	2011-04-05 21:11:05 +0000
@@ -0,0 +1,45 @@
+import sys
+import sqlite3
+
+class TestResult(object):
+    def __init__(self, job_id=0, device_id=0, result=0):
+        self.job_id = job_id
+        self.device_id = device_id
+        self.result = result
+
+    def __str__(self):
+        return 'Job ID: %s, Device ID: %s' % (self.job_id, self.device_id)
+
+class DbHelper():
+
+    def __init__(self):
+        self.connection = None
+        self.cursor = None
+
+    def connect_db(self):
+        self.connection = sqlite3.connect('./scheduler/database.db')
+        self.connection.row_factory = sqlite3.Row
+        self.cursor = self.connection.cursor()
+
+    def disconnect_db(self):
+        self.connection.close()
+
+    def get_next_job(self):
+       # sql = """select * from scheduler_app_testjob where status = 0 and submit_time = 
+       #              (select min(submit_time) from scheduler_app_testjob where status = 0)"""
+        sql = """select * from scheduler_app_testjob where status = 0 and submit_time = 
+            (select min(submit_time) from scheduler_app_testjob where exists 
+            (select * from scheduler_app_device where id=scheduler_app_testjob.target_id 
+            and scheduler_app_device.status=1) and status = 0)"""
+        self.cursor.execute(sql)
+        return self.cursor.fetchone()
+
+    def set_job_state(self, job_id, status):
+        sql = "update scheduler_app_testjob set status = ? where id = ?"
+        self.cursor.execute(sql, (status, job_id))
+        self.connection.commit()
+
+    def set_device_state(self, device_id, status):
+        sql = "update scheduler_app_device set status = ? where id = ?"
+        self.cursor.execute(sql, (status, device_id))
+        self.connection.commit()