diff options
| -rwxr-xr-x | tools/merge_backup.py | 2 | ||||
| -rwxr-xr-x | tools/merge_dist.py | 52 | ||||
| -rw-r--r-- | tools/mergetools.py | 26 |
3 files changed, 50 insertions, 30 deletions
diff --git a/tools/merge_backup.py b/tools/merge_backup.py index abe9f36..05679a1 100755 --- a/tools/merge_backup.py +++ b/tools/merge_backup.py @@ -124,7 +124,7 @@ def merge_backup(args, config, localconfig, secondaries): own_key, paths, hashes_and_entries, session) if sendentryresult["result"] != "ok": - print >>sys.stderr, "sendentry_merge:", sendentryresult + print >>sys.stderr, "sendentries_merge:", sendentryresult sys.exit(1) fetched_entries += len(missingentry_hashes) print >>sys.stderr, fetched_entries, diff --git a/tools/merge_dist.py b/tools/merge_dist.py index 2af1d6c..9d66cfd 100755 --- a/tools/merge_dist.py +++ b/tools/merge_dist.py @@ -6,12 +6,14 @@ import sys import json +import base64 +import requests from time import sleep from base64 import b64encode, b64decode from certtools import timing_point, \ create_ssl_context from mergetools import get_curpos, get_logorder, chunks, get_missingentries, \ - sendsth, sendlog, sendentry, parse_args, perm + sendsth, sendlog, sendentries, parse_args, perm def merge_dist(args, localconfig, frontendnodes, timestamp): paths = localconfig["paths"] @@ -84,25 +86,35 @@ def merge_dist(args, localconfig, frontendnodes, timestamp): paths) timing_point(timing, "get missing") - print >>sys.stderr, "missing entries:", len(missingentries) - sys.stderr.flush() - sent_entries = 0 - print >>sys.stderr, "send missing entries", - sys.stderr.flush() - for missingentry in missingentries: - ehash = b64decode(missingentry) - sendentryresult = sendentry(nodename, nodeaddress, own_key, paths, - chainsdb.get(ehash), ehash) - if sendentryresult["result"] != "ok": - print >>sys.stderr, "sendentry:", sendentryresult - sys.exit(1) - sent_entries += 1 - if sent_entries % 1000 == 0: - print >>sys.stderr, sent_entries, - sys.stderr.flush() - print >>sys.stderr - sys.stderr.flush() - timing_point(timing, "send missing") + + + while missingentries: + print >>sys.stderr, "missing entries:", len(missingentries) + sys.stderr.flush() + + sent_entries = 0 + print >>sys.stderr, "sending missing entries", + sys.stderr.flush() + with requests.sessions.Session() as session: + for missingentry_chunk in chunks(missingentries, 100): + missingentry_hashes = [base64.b64decode(missingentry) for missingentry in missingentry_chunk] + hashes_and_entries = [(hash, chainsdb.get(hash)) for hash in missingentry_hashes] + sendentryresult = sendentries(nodename, nodeaddress, + own_key, paths, + hashes_and_entries, session) + if sendentryresult["result"] != "ok": + print >>sys.stderr, "sendentries:", sendentryresult + sys.exit(1) + sent_entries += len(missingentry_hashes) + print >>sys.stderr, sent_entries, + sys.stderr.flush() + print >>sys.stderr + sys.stderr.flush() + timing_point(timing, "send missing") + + missingentries = get_missingentries(nodename, nodeaddress, + own_key, paths) + timing_point(timing, "get missing") print >>sys.stderr, "sending sth to node", nodename sys.stderr.flush() diff --git a/tools/mergetools.py b/tools/mergetools.py index 80fbf0b..bea09e9 100644 --- a/tools/mergetools.py +++ b/tools/mergetools.py @@ -258,6 +258,10 @@ def backup_sendlog(node, baseurl, own_key, paths, submission): print >>sys.stderr, "ERROR: backup_sendlog", e.response sys.stderr.flush() return None + except requests.packages.urllib3.exceptions.NewConnectionError, e: + print >>sys.stderr, "ERROR: backup_sendlog new connection error" + sys.stderr.flush() + return None except ValueError, e: print >>sys.stderr, "==== FAILED REQUEST ====" print >>sys.stderr, submission @@ -267,16 +271,17 @@ def backup_sendlog(node, baseurl, own_key, paths, submission): sys.stderr.flush() raise e -def sendentry(node, baseurl, own_key, paths, entry, ehash): +def sendentries(node, baseurl, own_key, paths, entries, session=None): try: + json_entries = [{"entry":base64.b64encode(entry), "treeleafhash":base64.b64encode(hash)} for hash, entry in entries] result = http_request( baseurl + "plop/v1/frontend/sendentry", - json.dumps({"entry":base64.b64encode(entry), - "treeleafhash":base64.b64encode(ehash)}), - key=own_key, verifynode=node, publickeydir=paths["publickeys"]) + json.dumps(json_entries), + key=own_key, verifynode=node, publickeydir=paths["publickeys"], + session=session) return json.loads(result) except requests.exceptions.HTTPError, e: - print >>sys.stderr, "ERROR: sendentry", e.reponse + print >>sys.stderr, "ERROR: sendentries", e.response sys.exit(1) except ValueError, e: print >>sys.stderr, "==== FAILED REQUEST ====" @@ -286,9 +291,9 @@ def sendentry(node, baseurl, own_key, paths, entry, ehash): print >>sys.stderr, "========================" sys.stderr.flush() raise e - -def sendentry_merge(node, baseurl, own_key, paths, entry, ehash): - return sendentries_merge(node, baseurl, own_key, paths, [(ehash, entry)]) + except requests.exceptions.ConnectionError, e: + print >>sys.stderr, "ERROR: sendentries", baseurl, e.request, e.response + sys.exit(1) def sendentries_merge(node, baseurl, own_key, paths, entries, session=None): try: @@ -300,7 +305,7 @@ def sendentries_merge(node, baseurl, own_key, paths, entries, session=None): session=session) return json.loads(result) except requests.exceptions.HTTPError, e: - print >>sys.stderr, "ERROR: sendentry_merge", e.response + print >>sys.stderr, "ERROR: sendentries_merge", e.response sys.exit(1) except ValueError, e: print >>sys.stderr, "==== FAILED REQUEST ====" @@ -310,6 +315,9 @@ def sendentries_merge(node, baseurl, own_key, paths, entries, session=None): print >>sys.stderr, "========================" sys.stderr.flush() raise e + except requests.exceptions.ConnectionError, e: + print >>sys.stderr, "ERROR: sendentries_merge", baseurl, e.request, e.response + sys.exit(1) def sendsth(node, baseurl, own_key, paths, submission): try: |