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