PLD_Builder/request_fetcher.py

   1 # vi: encoding=utf-8 ts=8 sts=4 sw=4 et
   2
   3 import string
   4 import signal
   5 import os
   6 import urllib
   7 import urllib2
   8 import StringIO
   9 import sys
  10 import gzip
  11
  12 import path
  13 import log
  14 import status
  15 import lock
  16 import util
  17 import gpg
  18 import request
  19 import loop
  20 import socket
  21 from acl import acl
  22 from bqueue import B_Queue
  23 from config import config, init_conf
  24
  25 last_count = 0
  26
  27 def alarmalarm(signum, frame):
  28     raise IOError, 'TCP connection hung'
  29
  30 def has_new(control_url):
  31     global last_count
  32     cnt_f = open(path.last_req_no_file)
  33     try:
  34         last_count = int(string.strip(cnt_f.readline()))
  35     except ValueError, e:
  36         last_count = 0
  37
  38     cnt_f.close()
  39     f = None
  40     socket.setdefaulttimeout(240)
  41     signal.signal(signal.SIGALRM, alarmalarm)
  42     signal.alarm(300)
  43     try:
  44         headers = { 'Cache-Control': 'no-cache', 'Pragma': 'no-cache' }
  45         req = urllib2.Request(url=control_url + "/max_req_no", headers=headers)
  46         f = urllib2.urlopen(req)
  47         count = int(string.strip(f.readline()))
  48         signal.alarm(0)
  49     except Exception, e:
  50         signal.alarm(0)
  51         log.error("can't fetch %s: %s" % (control_url + "/max_req_no", e))
  52         sys.exit(1)
  53     res = 0
  54     if count != last_count:
  55         res = 1
  56     f.close()
  57     return res
  58
  59 def fetch_queue(control_url):
  60     signal.signal(signal.SIGALRM, alarmalarm)
  61     socket.setdefaulttimeout(240)
  62     signal.alarm(300)
  63     try:
  64         headers = { 'Cache-Control': 'no-cache', 'Pragma': 'no-cache' }
  65         req = urllib2.Request(url=control_url + "/queue.gz", headers=headers)
  66         f = urllib2.urlopen(req)
  67         signal.alarm(0)
  68     except Exception, e:
  69         signal.alarm(0)
  70         log.error("can't fetch %s: %s" % (control_url + "/queue.gz", e))
  71         sys.exit(1)
  72     sio = StringIO.StringIO()
  73     util.sendfile(f, sio)
  74     f.close()
  75     sio.seek(0)
  76     f = gzip.GzipFile(fileobj = sio)
  77     (signers, body) = gpg.verify_sig(f.read())
  78     u = acl.user_by_email(signers)
  79     if u == None:
  80         log.alert("queue.gz not signed with signature of valid user: %s" % signers)
  81         sys.exit(1)
  82     if not u.can_do("sign_queue", "all"):
  83         log.alert("user %s is not allowed to sign my queue" % u.login)
  84         sys.exit(1)
  85     return request.parse_requests(body)
  86
  87 def handle_reqs(builder, reqs):
  88     qpath = path.queue_file + "-" + builder
  89     if not os.access(qpath, os.F_OK):
  90         util.append_to(qpath, "<queue/>\n")
  91     q = B_Queue(qpath)
  92     q.lock(0)
  93     q.read()
  94     for r in reqs:
  95         if r.kind != 'group':
  96             raise Exception, 'handle_reqs: fatal: huh? %s' % r.kind
  97         need_it = 0
  98         for b in r.batches:
  99             if builder in b.builders:
 100                 need_it = 1
 101         if need_it:
 102             log.notice("queued %s (%d) for %s" % (r.id, r.no, builder))
 103             q.add(r)
 104     q.write()
 105     q.unlock()
 106
 107 def main():
 108     lck = lock.lock("request_fetcher", non_block = True)
 109     if lck == None:
 110         sys.exit(1)
 111     init_conf()
 112     acl.try_reload()
 113
 114     status.push("fetching requests")
 115     if has_new(config.control_url):
 116         q = fetch_queue(config.control_url)
 117         max_no = 0
 118         q_new = []
 119         for r in q:
 120             if r.no > max_no:
 121                 max_no = r.no
 122             if r.no > last_count:
 123                 q_new.append(r)
 124         for b in config.binary_builders:
 125             handle_reqs(b, q_new)
 126         f = open(path.last_req_no_file, "w")
 127         f.write("%d\n" % max_no)
 128         f.close()
 129     status.pop()
 130     lck.close()
 131
 132 if __name__ == '__main__':
 133     # http connection is established (and few bytes transferred through it)
 134     # each $secs seconds.
 135     loop.run_loop(main, secs = 10)