#!/usr/bin/env python # -*- coding: utf-8 -*- # # Copyright (c) 2014-2015, NORDUnet A/S. # See LICENSE for licensing information. # # Fetch new entries from all storage nodes. # See catlfish/doc/merge.txt for more about the merge process. # import sys import struct import subprocess import requests import signal import logging from time import sleep from multiprocessing import Process, Pipe from random import Random from mergetools import get_logorder, verify_entry, get_new_entries, \ chunks, fsync_logorder, get_entries, add_to_logorder, \ hexencode, hexdecode, parse_args, perm, flock_ex_or_fail, Status, \ terminate_child_procs from certtools import timing_point, write_file, create_ssl_context def merge_fetch_sequenced(args, config, localconfig): paths = localconfig["paths"] storagenodes = config["storagenodes"] mergedb = paths["mergedb"] logorderfile = mergedb + "/logorder" statusfile = mergedb + "/merge_fetch.status" s = Status(statusfile) chainsdb = perm(localconfig.get("dbbackend", "filedb"), mergedb + "/chains") own_key = (localconfig["nodename"], "%s/%s-private.pem" % (paths["privatekeys"], localconfig["nodename"])) timing = timing_point() logorder = get_logorder(logorderfile) timing_point(timing, "get logorder") certsinlog = set(logorder) new_entries_per_node = {} new_entries = set() entries_to_fetch = {} for storagenode in storagenodes: logging.info("getting new entries from %s", storagenode["name"]) new_entries_per_node[storagenode["name"]] = \ set(get_new_entries(storagenode["name"], "https://%s/" % storagenode["address"], own_key, paths)) new_entries.update(new_entries_per_node[storagenode["name"]]) entries_to_fetch[storagenode["name"]] = [] timing_point(timing, "get new entries") new_entries -= certsinlog logging.info("adding %d entries", len(new_entries)) for ehash in new_entries: for storagenode in storagenodes: if ehash in new_entries_per_node[storagenode["name"]]: entries_to_fetch[storagenode["name"]].append(ehash) break verifycert = subprocess.Popen( [paths["verifycert_bin"], paths["known_roots"]], stdin=subprocess.PIPE, stdout=subprocess.PIPE) added_entries = 0 for storagenode in storagenodes: nentries = len(entries_to_fetch[storagenode["name"]]) logging.info("getting %d entries from %s", nentries, storagenode["name"]) with requests.sessions.Session() as session: for chunk in chunks(entries_to_fetch[storagenode["name"]], 100): entries = get_entries(storagenode["name"], "https://%s/" % storagenode["address"], own_key, paths, chunk, session=session) for ehash in chunk: entry = entries[ehash] verify_entry(verifycert, entry, ehash) chainsdb.add(ehash, entry) add_to_logorder(logorderfile, ehash) logorder.append(ehash) certsinlog.add(ehash) added_entries += 1 s.status("PROG getting %d entries from %s: %d" % (nentries, storagenode["name"], added_entries)) chainsdb.commit() fsync_logorder(logorderfile) timing_point(timing, "add entries") logging.info("added %d entries", added_entries) verifycert.communicate(struct.pack("I", 0)) if args.timing: logging.debug("timing: merge_fetch: %s", timing["deltatimes"]) tree_size = len(logorder) if tree_size == 0: return (0, '') else: return (tree_size, logorder[tree_size-1]) def merge_fetch_worker(args, localconfig, storagenode, pipe): paths = localconfig["paths"] own_key = (localconfig["nodename"], "%s/%s-private.pem" % (paths["privatekeys"], localconfig["nodename"])) # NOTE: We should probably verifycert.communicate(struct.pack("I",0)) # to ask the verifycert process to quit nicely. verifycert = subprocess.Popen([paths["verifycert_bin"], paths["known_roots"]], stdin=subprocess.PIPE, stdout=subprocess.PIPE) to_fetch = set() timeout = max(3, args.mergeinterval / 10) while True: if pipe.poll(timeout): msg = pipe.recv() if len(msg) < 2: continue cmd = msg[0] ehash = msg[1] if cmd == 'FETCH': to_fetch.add(ehash) if to_fetch: logging.info("%s: fetching %d entries", storagenode["name"], len(to_fetch)) fetchlist = list(to_fetch) with requests.sessions.Session() as session: entries = get_entries(storagenode["name"], "https://%s/" % storagenode["address"], own_key, paths, fetchlist, session=session) for ehash in fetchlist: entry = entries[ehash] verify_entry(verifycert, entry, ehash) pipe.send(('FETCHED', ehash, entry)) to_fetch.remove(ehash) new_entries = get_new_entries(storagenode["name"], "https://%s/" % storagenode["address"], own_key, paths) for ehash in new_entries: pipe.send(('NEWENTRY', ehash)) def term(signal, arg): terminate_child_procs() sys.exit(1) def newworker(name, args): my_conn, child_conn = Pipe() p = Process(target=merge_fetch_worker, args=tuple(args + [child_conn]), name='merge_fetch_%s' % name) p.daemon = True p.start() logging.debug("%s started, pid %d", name, p.pid) return (name, my_conn, p) def merge_fetch_parallel(args, config, localconfig): paths = localconfig["paths"] storagenodes = config["storagenodes"] mergedb = paths["mergedb"] chainsdb = perm(localconfig.get("dbbackend", "filedb"), mergedb + "/chains") logorderfile = mergedb + "/logorder" currentsizefile = mergedb + "/fetched" rand = Random() signal.signal(signal.SIGTERM, term) procs = {} for storagenode in storagenodes: name = storagenode['name'] procs[name] = newworker(name, [args, localconfig, storagenode]) currentsizefilecontent = "" logorder = get_logorder(logorderfile) # List of hashes in log, hexencoded. entries_in_log = set(logorder) # Set of hashes in log, binary. fetch_set = set() fetch_dict = {} while procs: # Poll worker processes and handle messages. assert(not fetch_set) newentry = [] for name, pipe, p in procs.values(): if not p.is_alive(): logging.warning("%s is gone, restarting", name) procs[name] = newworker(name, [args, localconfig, storagenodes[name]]) continue logging.info("polling %s", name) while pipe.poll(0): msg = pipe.recv() if len(msg) < 2: continue cmd = msg[0] ehash = msg[1] if cmd == 'NEWENTRY': logging.info("NEWENTRY at %s: %s", name, hexencode(ehash)) fetch_set.add(ehash) elif cmd == 'FETCHED': if len(msg) != 3: continue entry = msg[2] logging.info("FETCHED from %s: %s", name, hexencode(ehash)) chainsdb.add(ehash, entry) # Commit later. newentry.append(ehash) # Writing to logorderfile later. logorder.append(hexencode(ehash)) entries_in_log.add(ehash) if ehash in fetch_set: fetch_set.remove(ehash) del fetch_dict[ehash] chainsdb.commit() for ehash in newentry: add_to_logorder(logorderfile, ehash) fsync_logorder(logorderfile) # Ask workers to fetch entries. logging.debug("nof entries to fetch including entries in log: %d", len(fetch_set)) fetch_set -= entries_in_log logging.info("entries to fetch: %d", len(fetch_set)) # Add each entry in fetch_set to fetch_dict, key bing the hash # and value being a list of storage nodes, in randomised # order. for e in fetch_set: if not e in fetch_dict: save = procs.values() l = procs.values() rand.shuffle(l) assert save == procs.values() fetch_dict[e] = l # For each entry to fetch, treat its list of nodes as a # circular list and ask the one in the front to fetch the # entry. while fetch_set: ehash = fetch_set.pop() nodes = fetch_dict[ehash] node = nodes.pop(0) fetch_dict[ehash] = nodes + [node] name, pipe, p = node logging.info("asking %s to fetch %s", name, hexencode(ehash)) pipe.send(('FETCH', ehash)) # Update the 'fetched' file. logsize = len(logorder) if logsize == 0: last_hash = '' else: last_hash = logorder[logsize - 1] newcontent = {"index": logsize - 1, "hash": last_hash} if newcontent != currentsizefilecontent: logging.info("updating 'fetched' file: %d %s", logsize - 1, last_hash) currentsizefilecontent = newcontent write_file(currentsizefile, currentsizefilecontent) # Wait some. sleep(1) return 0 def main(): """ If no `--mergeinterval': Fetch new entries from all storage nodes, in sequence, updating the 'logorder' file and the 'chains' database. Write 'fetched' to reflect how far in 'logorder' we've succesfully fetched and verified. If `--mergeinterval': Start one process per storage node, read their stdout for learning about two things: (i) new entries ready for fetching ("NEWENTRY") and (ii) new entries being succesfully fetched ("FETCHED"). Write to their stdin ("FETCH") when they should fetch another entry. Update 'logorder' and the 'chains' database as we see new FETCHED messages. Write 'fetched' to reflect how far in 'logorder' we've succesfully fetched and verified. Keep doing this forever. NOTE: The point of having 'fetched' is that it can be atomically written while 'logorder' cannot (unless we're fine with rewriting it for each and every update, which we're not). TODO: Deduplicate some code. """ args, config, localconfig = parse_args() paths = localconfig["paths"] mergedb = paths["mergedb"] currentsizefile = mergedb + "/fetched" lockfile = mergedb + "/.merge_fetch.lock" loglevel = getattr(logging, args.loglevel.upper()) if args.mergeinterval is None: logging.basicConfig(level=loglevel) else: logging.basicConfig(filename=args.logdir + "/merge_fetch.log", level=loglevel) if not flock_ex_or_fail(lockfile): logging.critical("unable to take lock %s", lockfile) return 1 create_ssl_context(cafile=paths["https_cacertfile"]) if args.mergeinterval: return merge_fetch_parallel(args, config, localconfig) else: logsize, last_hash = merge_fetch_sequenced(args, config, localconfig) currentsize = {"index": logsize - 1, "hash": hexencode(last_hash)} logging.debug("writing to %s: %s", currentsizefile, currentsize) write_file(currentsizefile, currentsize) return 0 if __name__ == '__main__': sys.exit(main())