+++ /dev/null
-# vi: encoding=utf-8 ts=8 sts=4 sw=4 et
-
-import glob
-import re
-import string
-import os
-import time
-import shutil
-import sys
-import traceback
-import urllib2
-
-from config import config, init_conf
-import mailer
-import path
-import log
-import loop
-import status
-import lock
-
-retries_times = [5 * 60, 5 * 60, 10 * 60, 10 * 60, 30 * 60, 60 * 60]
-
-def read_name_val(file):
- f = open(file)
- r = {'_file': file[:-5], '_desc': file}
- rx = re.compile(r"^([^:]+)\s*:(.*)$")
- for l in f.xreadlines():
- if l == "END\n":
- f.close()
- return r
- m = rx.search(l)
- if m:
- r[m.group(1)] = string.strip(m.group(2))
- else:
- break
- f.close()
- return None
-
-def scp_file(src, target):
- global problems
- f = os.popen("scp -v -B %s %s 2>&1 < /dev/null" % (src, target))
- p = f.read()
- ret = f.close()
- if ret:
- problems[src] = p
- return ret
-
-def copy_file(src, target):
- try:
- shutil.copyfile(src, target)
- return 0
- except:
- global problems
- exctype, value = sys.exc_info()[:2]
- problems[src] = "cannot copy file: %s" % traceback.format_exception_only(exctype, value)
- return 1
-
-def rsync_file(src, target, host):
- global problems
-
- p = open(path.rsync_password_file, "r")
- password = ""
- for l in p.xreadlines():
- l = string.split(l)
- if len(l) >= 2 and l[0] == host:
- password = l[1]
- p.close()
-
- # NOTE: directing STDIN to /dev/null, does not make rsync to skip asking
- # password, it opens /dev/tty and still asks if password is needed and
- # missing, therefore we always set RSYNC_PASSWORD env var
- os.environ["RSYNC_PASSWORD"] = password
- rsync = "rsync --verbose --archive --timeout=360 --contimeout=360"
- f = os.popen("%s %s %s 2>&1" % (rsync, src, target))
- p = f.read()
- ret = f.close()
- if ret:
- problems[src] = p
- del os.environ["RSYNC_PASSWORD"];
- return ret
-
-def rsync_ssh_file(src, target):
- global problems
- rsync = "rsync --verbose --archive --timeout=360 -e ssh"
- f = os.popen("%s %s %s 2>&1 < /dev/null" % (rsync, src, target))
- p = f.read()
- ret = f.close()
- if ret:
- problems[src] = p
- return ret
-
-def post_file(src, url):
- global problems
- try:
- f = open(src, 'r')
- data = f.read()
- f.close()
- req = urllib2.Request(url, data)
- req.add_header('X-Filename', os.path.basename(src))
- f = urllib2.urlopen(req)
- f.close()
- except Exception, e:
- problems[src] = e
- return e
- return 0
-
-def send_file(src, target):
- global problems
- try:
- log.notice("sending %s to %s (size %d bytes)" % (src, target, os.stat(src).st_size))
- m = re.match('rsync://([^/]+)/.*', target)
- if m:
- return not rsync_file(src, target, host = m.group(1))
- if target != "" and target[0] == '/':
- return not copy_file(src, target)
- m = re.match('scp://([^@:]+@[^/:]+)(:|)(.*)', target)
- if m:
- return not scp_file(src, m.group(1) + ":" + m.group(3))
- m = re.match('ssh\+rsync://([^@:]+@[^/:]+)(:|)(.*)', target)
- if m:
- return not rsync_ssh_file(src, m.group(1) + ":" + m.group(3))
- m = re.match('http://.*', target)
- if m:
- return not post_file(src, target)
- log.alert("unsupported protocol: %s" % target)
- except OSError, e:
- problems[src] = e
- log.error("send_file(%s, %s): %s" % (src, target, e))
- return False
- return True
-
-def maybe_flush_queue(dir):
- retry_delay = 0
- try:
- f = open(dir + "/retry-at")
- last_retry = int(string.strip(f.readline()))
- retry_delay = int(string.strip(f.readline()))
- f.close()
- if last_retry + retry_delay > time.time():
- return
- os.unlink(dir + "/retry-at")
- except:
- pass
-
- status.push("flushing %s" % dir)
-
- if flush_queue(dir):
- f = open(dir + "/retry-at", "w")
- if retry_delay in retries_times:
- idx = retries_times.index(retry_delay)
- if idx < len(retries_times) - 1: idx += 1
- else:
- idx = 0
- f.write("%d\n%d\n" % (time.time(), retries_times[idx]))
- f.close()
-
- status.pop()
-
-def flush_queue(dir):
- q = []
- os.chdir(dir)
- for f in glob.glob(dir + "/*.desc"):
- d = read_name_val(f)
- if d != None: q.append(d)
- def mycmp(x, y):
- rc = cmp(x['Time'], y['Time'])
- if rc == 0 and x.has_key('Type') and y.has_key('Type'):
- return cmp(x['Type'], y['Type'])
- else:
- return rc
- q.sort(mycmp)
-
- error = None
- # copy of q
- remaining = q[:]
- for d in q:
- if not send_file(d['_file'], d['Target']):
- error = d
- continue
- if os.access(d['_file'] + ".info", os.F_OK):
- if not send_file(d['_file'] + ".info", d['Target'] + ".info"):
- error = d
- continue
- os.unlink(d['_file'] + ".info")
- os.unlink(d['_file'])
- os.unlink(d['_desc'])
- remaining.remove(d)
-
- if error != None:
- emails = {}
- emails[config.admin_email] = 1
- pr = ""
- for src, msg in problems.iteritems():
- pr = pr + "[src: %s]\n\n%s\n" % (src, msg)
- for d in remaining:
- if d.has_key('Requester'):
- emails[d['Requester']] = 1
- e = emails.keys()
- m = mailer.Message()
- m.set_headers(to = string.join(e, ", "),
- subject = "[%s] builder queue problem" % config.builder)
- m.write("there were problems sending files from queue %s:\n" % dir)
- m.write("problems:\n")
- m.write("%s\n" % pr)
- m.send()
- log.error("error sending files from %s:\n%s\n" % (dir, pr))
- return 1
-
- return 0
-
-problems = {}
-
-def main():
- if lock.lock("sending-files", non_block = 1) == None:
- return
- init_conf()
- maybe_flush_queue(path.notify_queue_dir)
- maybe_flush_queue(path.buildlogs_queue_dir)
- maybe_flush_queue(path.ftp_queue_dir)
-
-if __name__ == '__main__':
- loop.run_loop(main)
TLD / TLD.git / blobdiff