bin/jobrunner.py

   1 #!/usr/bin/env python
   2
   3 import sys
   4 import logging
   5 import subprocess
   6 from ais.db import *
   7
   8 def main():
   9     from optparse import OptionParser
  10
  11     parser = OptionParser()
  12     parser.add_option('-d', '--debug',
  13         action='store_true', dest='debug', default=False,
  14         help="debug mode")
  15     options, args = parser.parse_args()
  16
  17     if args:
  18         print >> sys.stderr, "That program doesn't take any argument"
  19         sys.exit(1)
  20
  21     if options.debug:
  22         loglevel = logging.DEBUG
  23         sql_setdebug(True)
  24     else:
  25         loglevel = logging.INFO
  26     logging.basicConfig(level=loglevel, format='%(asctime)s %(levelname)s %(message)s')
  27
  28     sqlexec(u'SELECT id, pid FROM job WHERE start_time IS NOT NULL AND finish_time IS NULL')
  29     row = get_common_cursor().fetchone()
  30     if row is not None:
  31         logging.debug('Job %s is running: pid=%s', row[0], row[1])
  32         sys.exit(0)
  33
  34     sqlexec(u'SELECT id, command, extension, user_id FROM job WHERE start_time IS NULL ORDER BY queue_time LIMIT 1')
  35     row = get_common_cursor().fetchone()
  36     if row is None:
  37         logging.debug('Queue is empty.')
  38         sys.exit(0)
  39
  40     jobid, command, extension, user_id = row
  41     logging.info('Starting job %s: %s', jobid, command)
  42
  43     sqlexec(u'UPDATE job SET start_time=now() WHERE id=%(jobid)s', {'jobid': jobid})
  44     dbcommit()
  45
  46     output = file('/var/lib/ais/jobs/'+unicode(jobid)+'.'+extension, 'wb')
  47     p = subprocess.Popen(command, stdout=output, shell=True)
  48     logging.debug('System process id = %s', p.pid)
  49     sqlexec(u'UPDATE job SET pid=' + unicode(p.pid) + ' WHERE id=%(jobid)s', {'jobid': jobid})
  50     dbcommit()
  51
  52     returncode = p.wait()
  53     sqlexec(u'UPDATE job SET pid=NULL, finish_time=now(), result=' + unicode(returncode) + ' WHERE id=%(jobid)s', {'jobid': jobid})
  54     dbcommit()
  55     logging.info('Job complete: result=%s', returncode)
  56
  57     sqlexec(u"INSERT INTO user_message (user_id, user_message_category_id, txt) VALUES(%(user_id)s, 'info', %(msg)s)", {'user_id':user_id, 'msg':('Your <a href="/job/%(jobid)s">job %(jobid)s</a> is complete.' % {'jobid': jobid}) })
  58     dbcommit()
  59
  60 if __name__ == '__main__':
  61     main()