PLD_Builder/request_fetcher.py

   1 import string
   2 import os
   3 import urllib
   4 import StringIO
   5 import sys
   6 import gzip
   7
   8 import path
   9 import log
  10 import status
  11 import lock
  12 import util
  13 import gpg
  14 import request
  15 import loop
  16 from acl import acl
  17 from bqueue import B_Queue
  18 from config import config, init_conf
  19
  20 last_count = 0
  21
  22 def has_new(control_url):
  23   global last_count
  24   cnt_f = open(path.last_req_no_file)
  25   last_count = int(string.strip(cnt_f.readline()))
  26   cnt_f.close()
  27   f = urllib.urlopen(control_url + "/max_req_no")
  28   res = 0
  29   if int(string.strip(f.readline())) != last_count:
  30     res = 1
  31   f.close()
  32   return res
  33
  34 def fetch_queue(control_url, queue_signed_by):
  35   f = urllib.urlopen(control_url + "/queue.gz")
  36   sio = StringIO.StringIO()
  37   util.sendfile(f, sio)
  38   f.close()
  39   sio.seek(0)
  40   f = gzip.GzipFile(fileobj = sio)
  41   (signers, body) = gpg.verify_sig(f)
  42   u = acl.user_by_email(signers)
  43   if u == None:
  44     log.alert("queue.gz not signed with signature of valid user: %s" % signers)
  45     sys.exit(1)
  46   if u.login != queue_signed_by:
  47     log.alert("queue.gz should be signed by %s not by %s" \
  48         % (queue_signed_by, u.login))
  49     sys.exit(1)
  50   body.seek(0)
  51   return request.parse_requests(body)
  52
  53 def handle_reqs(builder, reqs):
  54   qpath = path.queue_file + "-" + builder
  55   if not os.access(qpath, os.F_OK):
  56     util.append_to(qpath, "<queue/>\n")
  57   q = B_Queue(qpath)
  58   q.lock(0)
  59   q.read()
  60   for r in reqs:
  61     if r.kind != 'group':
  62       raise 'handle_reqs: fatal: huh? %s' % r.kind
  63     need_it = 0
  64     for b in r.batches:
  65       if builder in b.builders:
  66         need_it = 1
  67     if need_it:
  68       log.notice("queued %s (%d) for %s" % (r.id, r.no, builder))
  69       q.add(r)
  70   q.write()
  71   q.unlock()
  72
  73 def main():
  74   lck = lock.lock("request_fetcher")
  75   init_conf("")
  76
  77   status.push("fetching requests")
  78   if has_new(config.control_url):
  79     q = fetch_queue(config.control_url, config.queue_signed_by)
  80     max_no = 0
  81     q_new = []
  82     for r in q:
  83       if r.no > max_no:
  84         max_no = r.no
  85       if r.no > last_count:
  86         q_new.append(r)
  87     for b in config.binary_builders:
  88       handle_reqs(b, q_new)
  89     f = open(path.last_req_no_file, "w")
  90     f.write("%d\n" % max_no)
  91     f.close()
  92   status.pop()
  93   lck.close()
  94
  95 if __name__ == '__main__':
  96   # http connection is established (and few bytes transferred through it)
  97   # each $secs seconds.
  98   loop.run_loop(main, secs = 10)