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