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