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