diff options
Diffstat (limited to 'tools')
| -rwxr-xr-x | tools/merge.py | 123 |
1 files changed, 79 insertions, 44 deletions
diff --git a/tools/merge.py b/tools/merge.py index 8766491..9904b84 100755 --- a/tools/merge.py +++ b/tools/merge.py @@ -84,10 +84,10 @@ def get_new_entries(node, baseurl): parsed_result = json.loads(result) if parsed_result.get(u"result") == u"ok": return [base64.b64decode(entry) for entry in parsed_result[u"entries"]] - print "ERROR: fetchnewentries", parsed_result + print >>sys.stderr, "ERROR: fetchnewentries", parsed_result sys.exit(1) except urllib2.HTTPError, e: - print "ERROR: fetchnewentries", e.read() + print >>sys.stderr, "ERROR: fetchnewentries", e.read() sys.exit(1) def get_entries(node, baseurl, hashes): @@ -100,10 +100,10 @@ def get_entries(node, baseurl, hashes): assert len(entries) == len(hashes) assert set(entries.keys()) == set(hashes) return entries - print "ERROR: getentry", parsed_result + print >>sys.stderr, "ERROR: getentry", parsed_result sys.exit(1) except urllib2.HTTPError, e: - print "ERROR: getentry", e.read() + print >>sys.stderr, "ERROR: getentry", e.read() sys.exit(1) def get_curpos(node, baseurl): @@ -112,10 +112,10 @@ def get_curpos(node, baseurl): parsed_result = json.loads(result) if parsed_result.get(u"result") == u"ok": return parsed_result[u"position"] - print "ERROR: currentposition", parsed_result + print >>sys.stderr, "ERROR: currentposition", parsed_result sys.exit(1) except urllib2.HTTPError, e: - print "ERROR: currentposition", e.read() + print >>sys.stderr, "ERROR: currentposition", e.read() sys.exit(1) def sendlog(node, baseurl, submission): @@ -124,14 +124,16 @@ def sendlog(node, baseurl, submission): json.dumps(submission), key=own_key, verifynode=node, publickeydir=paths["publickeys"]) return json.loads(result) except urllib2.HTTPError, e: - print "ERROR: sendlog", e.read() + print >>sys.stderr, "ERROR: sendlog", e.read() + sys.stderr.flush() return None except ValueError, e: - print "==== FAILED REQUEST ====" - print submission - print "======= RESPONSE =======" - print result - print "========================" + print >>sys.stderr, "==== FAILED REQUEST ====" + print >>sys.stderr, submission + print >>sys.stderr, "======= RESPONSE =======" + print >>sys.stderr, result + print >>sys.stderr, "========================" + sys.stderr.flush() raise e def sendentry(node, baseurl, entry, hash): @@ -141,14 +143,15 @@ def sendentry(node, baseurl, entry, hash): verifynode=node, publickeydir=paths["publickeys"]) return json.loads(result) except urllib2.HTTPError, e: - print "ERROR: sendentry", e.read() + print >>sys.stderr, "ERROR: sendentry", e.read() sys.exit(1) except ValueError, e: - print "==== FAILED REQUEST ====" - print hash - print "======= RESPONSE =======" - print result - print "========================" + print >>sys.stderr, "==== FAILED REQUEST ====" + print >>sys.stderr, hash + print >>sys.stderr, "======= RESPONSE =======" + print >>sys.stderr, result + print >>sys.stderr, "========================" + sys.stderr.flush() raise e def sendsth(node, baseurl, submission): @@ -157,14 +160,15 @@ def sendsth(node, baseurl, submission): json.dumps(submission), key=own_key, verifynode=node, publickeydir=paths["publickeys"]) return json.loads(result) except urllib2.HTTPError, e: - print "ERROR: sendsth", e.read() + print >>sys.stderr, "ERROR: sendsth", e.read() sys.exit(1) except ValueError, e: - print "==== FAILED REQUEST ====" - print submission - print "======= RESPONSE =======" - print result - print "========================" + print >>sys.stderr, "==== FAILED REQUEST ====" + print >>sys.stderr, submission + print >>sys.stderr, "======= RESPONSE =======" + print >>sys.stderr, result + print >>sys.stderr, "========================" + sys.stderr.flush() raise e def get_missingentries(node, baseurl): @@ -173,10 +177,10 @@ def get_missingentries(node, baseurl): parsed_result = json.loads(result) if parsed_result.get(u"result") == u"ok": return parsed_result[u"entries"] - print "ERROR: missingentries", parsed_result + print >>sys.stderr, "ERROR: missingentries", parsed_result sys.exit(1) except urllib2.HTTPError, e: - print "ERROR: missingentries", e.read() + print >>sys.stderr, "ERROR: missingentries", e.read() sys.exit(1) def chunks(l, n): @@ -195,7 +199,8 @@ new_entries = set() entries_to_fetch = {} for storagenode in storagenodes: - print "getting new entries from", storagenode["name"] + print >>sys.stderr, "getting new entries from", storagenode["name"] + sys.stderr.flush() new_entries_per_node[storagenode["name"]] = set(get_new_entries(storagenode["name"], "https://%s/" % storagenode["address"])) new_entries.update(new_entries_per_node[storagenode["name"]]) entries_to_fetch[storagenode["name"]] = [] @@ -206,7 +211,8 @@ timing_point(timing, "get new entries") new_entries -= certsinlog -print "adding", len(new_entries), "entries" +print >>sys.stderr, "adding", len(new_entries), "entries" +sys.stderr.flush() if args.nomerge: sys.exit(0) @@ -222,7 +228,8 @@ verifycert = subprocess.Popen([paths["verifycert_bin"], paths["known_roots"]], added_entries = 0 for storagenode in storagenodes: - print "getting", len(entries_to_fetch[storagenode["name"]]), "entries from", storagenode["name"] + print >>sys.stderr, "getting %d entries from %s:" % (len(entries_to_fetch[storagenode["name"]]), storagenode["name"]), + sys.stderr.flush() for chunk in chunks(entries_to_fetch[storagenode["name"]], 100): entries = get_entries(storagenode["name"], "https://%s/" % storagenode["address"], chunk) for hash in chunk: @@ -233,9 +240,14 @@ for storagenode in storagenodes: logorder.append(hash) certsinlog.add(hash) added_entries += 1 + print >>sys.stderr, added_entries, + sys.stderr.flush() + print >>sys.stderr + sys.stderr.flush() fsync_logorder() timing_point(timing, "add entries") -print "added", added_entries, "entries" +print >>sys.stderr, "added", added_entries, "entries" +sys.stderr.flush() verifycert.communicate(struct.pack("I", 0)) @@ -252,12 +264,14 @@ for secondary in secondaries: localdir = localdir + "/" print >>sys.stderr, "copying database to secondary:", remotehost + sys.stderr.flush() rsyncstatus = subprocess.call(["rsync", "-r", "--append", "--rsh=ssh", localdir, remotedir]) if rsyncstatus: print >>sys.stderr, "rsync failed:", rsyncstatus sys.exit(1) print >>sys.stderr, "verifying database at secondary:", remotehost + sys.stderr.flush() verifysecondary = subprocess.Popen(["ssh", remotehost, secondary["verifycommand"], secondary["mergedir"]], stdout=subprocess.PIPE) @@ -275,7 +289,8 @@ for signingnode in signingnodes: root_hash, "https://%s/" % signingnode["address"], key=own_key) break except urllib2.URLError, e: - print e + print >>sys.stderr, e + sys.stderr.flush() if tree_head_signature == None: print >>sys.stderr, "Could not contact any signing nodes" sys.exit(1) @@ -289,19 +304,25 @@ check_sth_signature(ctbaseurl, sth, publickey=logpublickey) timing_point(timing, "build sth") if args.timing: - print timing["deltatimes"] + print >>sys.stderr, timing["deltatimes"] + sys.stderr.flush() -print "root hash", base64.b16encode(root_hash) +print base64.b16encode(root_hash) +sys.stdout.flush() for frontendnode in frontendnodes: nodeaddress = "https://%s/" % frontendnode["address"] nodename = frontendnode["name"] timing = timing_point() - print "distributing for node", nodename + print >>sys.stderr, "distributing for node", nodename + sys.stderr.flush() curpos = get_curpos(nodename, nodeaddress) timing_point(timing, "get curpos") - print "current position", curpos + print >>sys.stderr, "current position", curpos + sys.stderr.flush() entries = [base64.b64encode(entry) for entry in logorder[curpos:]] + print >>sys.stderr, "sending log:", + sys.stderr.flush() for chunk in chunks(entries, 1000): for trynumber in range(5, 0, -1): sendlogresult = sendlog(nodename, nodeaddress, {"start": curpos, "hashes": chunk}) @@ -309,31 +330,45 @@ for frontendnode in frontendnodes: if trynumber == 1: sys.exit(1) select.select([], [], [], 10.0) - print "tries left:", trynumber + print >>sys.stderr, "tries left:", trynumber + sys.stderr.flush() continue break if sendlogresult["result"] != "ok": - print "sendlog:", sendlogresult + print >>sys.stderr, "sendlog:", sendlogresult sys.exit(1) curpos += len(chunk) - print curpos, - sys.stdout.flush() + print >>sys.stderr, curpos, + sys.stderr.flush() + print >>sys.stderr timing_point(timing, "sendlog") - print "log sent" + print >>sys.stderr, "log sent" + sys.stderr.flush() missingentries = get_missingentries(nodename, nodeaddress) timing_point(timing, "get missing") - print "missing entries:", len(missingentries) + print >>sys.stderr, "missing entries:", len(missingentries) + sys.stderr.flush() + fetched_entries = 0 + print >>sys.stderr, "fetching missing entries", + sys.stderr.flush() for missingentry in missingentries: hash = base64.b64decode(missingentry) sendentryresult = sendentry(nodename, nodeaddress, read_chain(chainsdir, hash), hash) if sendentryresult["result"] != "ok": - print "send sth:", sendentryresult + print >>sys.stderr, "send sth:", sendentryresult sys.exit(1) + fetched_entries += 1 + if added_entries % 1000 == 0: + print >>sys.stderr, fetched_entries, + sys.stderr.flush() + print >>sys.stderr + sys.stderr.flush() timing_point(timing, "send missing") sendsthresult = sendsth(nodename, nodeaddress, sth) if sendsthresult["result"] != "ok": - print "send sth:", sendsthresult + print >>sys.stderr, "send sth:", sendsthresult sys.exit(1) timing_point(timing, "send sth") if args.timing: - print timing["deltatimes"] + print >>sys.stderr, timing["deltatimes"] + sys.stderr.flush() |