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