X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/548f183c23606b9febcadbbfd658aa921c3baaf5..0bd1c28bed9a0756c61037947d5a9dccd5066f00:/sdk/python/arvados/keep.py?ds=sidebyside diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py index c50699796b..b5d5d72503 100644 --- a/sdk/python/arvados/keep.py +++ b/sdk/python/arvados/keep.py @@ -18,12 +18,91 @@ import fcntl import time import threading import timer +import datetime +import ssl +_logger = logging.getLogger('arvados.keep') global_client_object = None from api import * import config import arvados.errors +import arvados.util + +class KeepLocator(object): + EPOCH_DATETIME = datetime.datetime.utcfromtimestamp(0) + + def __init__(self, locator_str): + self.size = None + self.loc_hint = None + self._perm_sig = None + self._perm_expiry = None + pieces = iter(locator_str.split('+')) + self.md5sum = next(pieces) + for hint in pieces: + if hint.startswith('A'): + self.parse_permission_hint(hint) + elif hint.startswith('K'): + self.loc_hint = hint # FIXME + elif hint.isdigit(): + self.size = int(hint) + else: + raise ValueError("unrecognized hint data {}".format(hint)) + + def __str__(self): + return '+'.join( + str(s) for s in [self.md5sum, self.size, self.loc_hint, + self.permission_hint()] + if s is not None) + + def _make_hex_prop(name, length): + # Build and return a new property with the given name that + # must be a hex string of the given length. + data_name = '_{}'.format(name) + def getter(self): + return getattr(self, data_name) + def setter(self, hex_str): + if not arvados.util.is_hex(hex_str, length): + raise ValueError("{} must be a {}-digit hex string: {}". + format(name, length, hex_str)) + setattr(self, data_name, hex_str) + return property(getter, setter) + + md5sum = _make_hex_prop('md5sum', 32) + perm_sig = _make_hex_prop('perm_sig', 40) + + @property + def perm_expiry(self): + return self._perm_expiry + + @perm_expiry.setter + def perm_expiry(self, value): + if not arvados.util.is_hex(value, 1, 8): + raise ValueError( + "permission timestamp must be a hex Unix timestamp: {}". + format(value)) + self._perm_expiry = datetime.datetime.utcfromtimestamp(int(value, 16)) + + def permission_hint(self): + data = [self.perm_sig, self.perm_expiry] + if None in data: + return None + data[1] = int((data[1] - self.EPOCH_DATETIME).total_seconds()) + return "A{}@{:08x}".format(*data) + + def parse_permission_hint(self, s): + try: + self.perm_sig, self.perm_expiry = s[1:].split('@', 1) + except IndexError: + raise ValueError("bad permission hint {}".format(s)) + + def permission_expired(self, as_of_dt=None): + if self.perm_expiry is None: + return False + elif as_of_dt is None: + as_of_dt = datetime.datetime.now() + return self.perm_expiry <= as_of_dt + class Keep: @staticmethod @@ -55,6 +134,7 @@ class KeepClient(object): def __init__(self, todo): self._todo = todo self._done = 0 + self._response = None self._todo_lock = threading.Semaphore(todo) self._done_lock = threading.Lock() @@ -73,12 +153,23 @@ class KeepClient(object): with self._done_lock: return (self._done < self._todo) - def increment_done(self): + def save_response(self, response_body, replicas_stored): + """ + Records a response body (a locator, possibly signed) returned by + the Keep server. It is not necessary to save more than + one response, since we presume that any locator returned + in response to a successful request is valid. + """ + with self._done_lock: + self._done += replicas_stored + self._response = response_body + + def response(self): """ - Report that the current thread was successful. + Returns the body from the response to a PUT request. """ with self._done_lock: - self._done += 1 + return self._response def done(self): """ @@ -89,13 +180,17 @@ class KeepClient(object): class KeepWriterThread(threading.Thread): """ - Write a blob of data to the given Keep server. Call - increment_done() of the given ThreadLimiter if the write - succeeds. + Write a blob of data to the given Keep server. On success, call + save_response() of the given ThreadLimiter to save the returned + locator. """ def __init__(self, **kwargs): super(KeepClient.KeepWriterThread, self).__init__() self.args = kwargs + self._success = False + + def success(self): + return self._success def run(self): with self.args['thread_limiter'] as limiter: @@ -103,58 +198,66 @@ class KeepClient(object): # My turn arrived, but the job has been done without # me. return - logging.debug("KeepWriterThread %s proceeding %s %s" % - (str(threading.current_thread()), - self.args['data_hash'], - self.args['service_root'])) - h = httplib2.Http() - url = self.args['service_root'] + self.args['data_hash'] - api_token = config.get('ARVADOS_API_TOKEN') - headers = {'Authorization': "OAuth2 %s" % api_token} - - if self.args['using_proxy']: - # We're using a proxy, so tell the proxy how many copies we - # want it to store - headers['X-Keep-Desired-Replication'] = str(self.args['want_copies']) - - try: + self.run_with_limiter(limiter) + + def run_with_limiter(self, limiter): + _logger.debug("KeepWriterThread %s proceeding %s %s", + str(threading.current_thread()), + self.args['data_hash'], + self.args['service_root']) + h = httplib2.Http(timeout=self.args.get('timeout', None)) + url = self.args['service_root'] + self.args['data_hash'] + api_token = config.get('ARVADOS_API_TOKEN') + headers = {'Authorization': "OAuth2 %s" % api_token} + + if self.args['using_proxy']: + # We're using a proxy, so tell the proxy how many copies we + # want it to store + headers['X-Keep-Desired-Replication'] = str(self.args['want_copies']) + + try: + _logger.debug("Uploading to {}".format(url)) + resp, content = h.request(url.encode('utf-8'), 'PUT', + headers=headers, + body=self.args['data']) + if (resp['status'] == '401' and + re.match(r'Timestamp verification failed', content)): + body = KeepClient.sign_for_old_server( + self.args['data_hash'], + self.args['data']) + h = httplib2.Http(timeout=self.args.get('timeout', None)) resp, content = h.request(url.encode('utf-8'), 'PUT', headers=headers, - body=self.args['data']) - if (resp['status'] == '401' and - re.match(r'Timestamp verification failed', content)): - body = KeepClient.sign_for_old_server( - self.args['data_hash'], - self.args['data']) - h = httplib2.Http() - resp, content = h.request(url.encode('utf-8'), 'PUT', - headers=headers, - body=body) - if re.match(r'^2\d\d$', resp['status']): - logging.debug("KeepWriterThread %s succeeded %s %s" % - (str(threading.current_thread()), - self.args['data_hash'], - self.args['service_root'])) - - if 'x-keep-replicas-stored' in resp: - # Tick the 'done' counter for the number of replica - # reported stored by the server, for the case that - # we're talking to a proxy or other backend that - # stores to multiple copies for us. - replicas = int(resp['x-keep-replicas-stored']) - while replicas > 0: - limiter.increment_done() - replicas -= 1 - else: - limiter.increment_done() - return - logging.warning("Request fail: PUT %s => %s %s" % - (url, resp['status'], content)) - except (httplib2.HttpLib2Error, httplib.HTTPException) as e: - logging.warning("Request fail: PUT %s => %s: %s" % - (url, type(e), str(e))) - - def __init__(self): + body=body) + if re.match(r'^2\d\d$', resp['status']): + self._success = True + _logger.debug("KeepWriterThread %s succeeded %s %s", + str(threading.current_thread()), + self.args['data_hash'], + self.args['service_root']) + replicas_stored = 1 + if 'x-keep-replicas-stored' in resp: + # Tick the 'done' counter for the number of replica + # reported stored by the server, for the case that + # we're talking to a proxy or other backend that + # stores to multiple copies for us. + try: + replicas_stored = int(resp['x-keep-replicas-stored']) + except ValueError: + pass + limiter.save_response(content.strip(), replicas_stored) + else: + _logger.warning("Request fail: PUT %s => %s %s", + url, resp['status'], content) + except (httplib2.HttpLib2Error, + httplib.HTTPException, + ssl.SSLError) as e: + # When using https, timeouts look like ssl.SSLError from here. + # "SSLError: The write operation timed out" + _logger.warning("Request fail: PUT %s => %s: %s", + url, type(e), str(e)) + + def __init__(self, **kwargs): self.lock = threading.Lock() self.service_roots = None self._cache_lock = threading.Lock() @@ -162,6 +265,7 @@ class KeepClient(object): # default 256 megabyte cache self.cache_max = 256 * 1024 * 1024 self.using_proxy = False + self.timeout = kwargs.get('timeout', 60) def shuffled_service_roots(self, hash): if self.service_roots == None: @@ -169,42 +273,78 @@ class KeepClient(object): # Override normal keep disk lookup with an explict proxy # configuration. - keep_proxy_env = self.config.get("ARVADOS_KEEP_PROXY") - if keep_proxy_env != None: + keep_proxy_env = config.get("ARVADOS_KEEP_PROXY") + if keep_proxy_env != None and len(keep_proxy_env) > 0: + if keep_proxy_env[-1:] != '/': keep_proxy_env += "/" self.service_roots = [keep_proxy_env] self.using_proxy = True else: try: - keep_disks = arvados.api().keep_disks().list().execute()['items'] + try: + keep_services = arvados.api().keep_services().accessible().execute()['items'] + except Exception: + keep_services = arvados.api().keep_disks().list().execute()['items'] + + if len(keep_services) == 0: + raise arvados.errors.NoKeepServersError() + + if 'service_type' in keep_services[0] and keep_services[0]['service_type'] == 'proxy': + self.using_proxy = True + roots = (("http%s://%s:%d/" % ('s' if f['service_ssl_flag'] else '', f['service_host'], f['service_port'])) - for f in keep_disks) + for f in keep_services) self.service_roots = sorted(set(roots)) - if len(keep_disks) == 1 and re.match(r'[a-z0-9]{5}-penuu-keepproxy......', keep_disks[0]['uuid']): - # Proxies have a special UUID pattern. - self.using_proxy = True - logging.debug(str(self.service_roots)) + _logger.debug(str(self.service_roots)) finally: self.lock.release() + # Build an ordering with which to query the Keep servers based on the + # contents of the hash. + # "hash" is a hex-encoded number at least 8 digits + # (32 bits) long + + # seed used to calculate the next keep server from 'pool' + # to be added to 'pseq' seed = hash + + # Keep servers still to be added to the ordering pool = self.service_roots[:] + + # output probe sequence pseq = [] + + # iterate while there are servers left to be assigned while len(pool) > 0: if len(seed) < 8: - if len(pseq) < len(hash) / 4: # first time around + # ran out of digits in the seed + if len(pseq) < len(hash) / 4: + # the number of servers added to the probe sequence is less + # than the number of 4-digit slices in 'hash' so refill the + # seed with the last 4 digits and then append the contents + # of 'hash'. seed = hash[-4:] + hash else: + # refill the seed with the contents of 'hash' seed += hash + + # Take the next 8 digits (32 bytes) and interpret as an integer, + # then modulus with the size of the remaining pool to get the next + # selected server. probe = int(seed[0:8], 16) % len(pool) + + # Append the selected server to the probe sequence and remove it + # from the pool. pseq += [pool[probe]] pool = pool[:probe] + pool[probe+1:] + + # Remove the digits just used from the seed seed = seed[8:] - logging.debug(str(pseq)) + _logger.debug(str(pseq)) return pseq class CacheSlot(object): @@ -262,8 +402,6 @@ class KeepClient(object): self._cache_lock.release() def get(self, locator): - #logging.debug("Keep.get %s" % (locator)) - if re.search(r',', locator): return ''.join(self.get(x) for x in locator.split(',')) if 'KEEP_LOCAL_STORE' in os.environ: @@ -271,7 +409,6 @@ class KeepClient(object): expect_hash = re.sub(r'\+.*', '', locator) slot, first = self.reserve_cache(expect_hash) - #logging.debug("%s %s %s" % (slot, first, expect_hash)) if not first: v = slot.get() @@ -279,7 +416,7 @@ class KeepClient(object): try: for service_root in self.shuffled_service_roots(expect_hash): - url = service_root + expect_hash + url = service_root + locator api_token = config.get('ARVADOS_API_TOKEN') headers = {'Authorization': "OAuth2 %s" % api_token, 'Accept': 'application/octet-stream'} @@ -291,7 +428,7 @@ class KeepClient(object): for location_hint in re.finditer(r'\+K@([a-z0-9]+)', locator): instance = location_hint.group(1) - url = 'http://keep.' + instance + '.arvadosapi.com/' + expect_hash + url = 'http://keep.' + instance + '.arvadosapi.com/' + locator blob = self.get_url(url, {}, expect_hash) if blob: slot.set(blob) @@ -309,23 +446,23 @@ class KeepClient(object): def get_url(self, url, headers, expect_hash): h = httplib2.Http() try: - logging.info("Request: GET %s" % (url)) + _logger.info("Request: GET %s", url) with timer.Timer() as t: resp, content = h.request(url.encode('utf-8'), 'GET', headers=headers) - logging.info("Received %s bytes in %s msec (%s MiB/sec)" % (len(content), - t.msecs, - (len(content)/(1024*1024))/t.secs)) + _logger.info("Received %s bytes in %s msec (%s MiB/sec)", + len(content), t.msecs, + (len(content)/(1024*1024))/t.secs) if re.match(r'^2\d\d$', resp['status']): m = hashlib.new('md5') m.update(content) md5 = m.hexdigest() if md5 == expect_hash: return content - logging.warning("Checksum fail: md5(%s) = %s" % (url, md5)) + _logger.warning("Checksum fail: md5(%s) = %s", url, md5) except Exception as e: - logging.info("Request fail: GET %s => %s: %s" % - (url, type(e), str(e))) + _logger.info("Request fail: GET %s => %s: %s", + url, type(e), str(e)) return None def put(self, data, **kwargs): @@ -341,19 +478,37 @@ class KeepClient(object): threads = [] thread_limiter = KeepClient.ThreadLimiter(want_copies) for service_root in self.shuffled_service_roots(data_hash): - t = KeepClient.KeepWriterThread(data=data, - data_hash=data_hash, - service_root=service_root, - thread_limiter=thread_limiter, - using_proxy=self.using_proxy, - want_copies=(want_copies if self.using_proxy else 1)) + t = KeepClient.KeepWriterThread( + data=data, + data_hash=data_hash, + service_root=service_root, + thread_limiter=thread_limiter, + timeout=self.timeout, + using_proxy=self.using_proxy, + want_copies=(want_copies if self.using_proxy else 1)) t.start() threads += [t] for t in threads: t.join() + if thread_limiter.done() < want_copies: + # Retry the threads (i.e., services) that failed the first + # time around. + threads_retry = [] + for t in threads: + if not t.success(): + _logger.warning("Retrying: PUT %s %s", + t.args['service_root'], + t.args['data_hash']) + retry_with_args = t.args.copy() + t_retry = KeepClient.KeepWriterThread(**retry_with_args) + t_retry.start() + threads_retry += [t_retry] + for t in threads_retry: + t.join() have_copies = thread_limiter.done() + # If we're done, return the response from Keep if have_copies >= want_copies: - return (data_hash + '+' + str(len(data))) + return thread_limiter.response() raise arvados.errors.KeepWriteError( "Write fail for %s: wanted %d but wrote %d" % (data_hash, want_copies, have_copies))