X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/3e3389504c6841e160ac086328693001261e016d..f12350f25275fbf4c6c8692122f5eacce89794ee:/sdk/python/arvados/keep.py diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py index 4c2d474010..561d34cd40 100644 --- a/sdk/python/arvados/keep.py +++ b/sdk/python/arvados/keep.py @@ -19,47 +19,44 @@ 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 +import arvados.config as config import arvados.errors +import arvados.util class KeepLocator(object): EPOCH_DATETIME = datetime.datetime.utcfromtimestamp(0) - HEX_RE = re.compile(r'^[0-9a-fA-F]+$') + HINT_RE = re.compile(r'^[A-Z][A-Za-z0-9@_-]+$') def __init__(self, locator_str): - self.size = None - self.loc_hint = None + self.hints = [] self._perm_sig = None self._perm_expiry = None pieces = iter(locator_str.split('+')) self.md5sum = next(pieces) + try: + self.size = int(next(pieces)) + except StopIteration: + self.size = None for hint in pieces: - if hint.startswith('A'): + if self.HINT_RE.match(hint) is None: + raise ValueError("unrecognized hint data {}".format(hint)) + elif 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)) + self.hints.append(hint) def __str__(self): return '+'.join( - str(s) for s in [self.md5sum, self.size, self.loc_hint, - self.permission_hint()] + str(s) for s in [self.md5sum, self.size, + self.permission_hint()] + self.hints if s is not None) - def _is_hex_length(self, s, *size_spec): - if len(size_spec) == 1: - good_len = (len(s) == size_spec[0]) - else: - good_len = (size_spec[0] <= len(s) <= size_spec[1]) - return good_len and self.HEX_RE.match(s) - 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. @@ -67,7 +64,7 @@ class KeepLocator(object): def getter(self): return getattr(self, data_name) def setter(self, hex_str): - if not self._is_hex_length(hex_str, length): + 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) @@ -82,7 +79,7 @@ class KeepLocator(object): @perm_expiry.setter def perm_expiry(self, value): - if not self._is_hex_length(value, 1, 8): + if not arvados.util.is_hex(value, 1, 8): raise ValueError( "permission timestamp must be a hex Unix timestamp: {}". format(value)) @@ -109,12 +106,31 @@ class KeepLocator(object): return self.perm_expiry <= as_of_dt -class Keep: - @staticmethod - def global_client_object(): +class Keep(object): + """Simple interface to a global KeepClient object. + + THIS CLASS IS DEPRECATED. Please instantiate your own KeepClient with your + own API client. The global KeepClient will build an API client from the + current Arvados configuration, which may not match the one you built. + """ + _last_key = None + + @classmethod + def global_client_object(cls): global global_client_object - if global_client_object == None: + # Previously, KeepClient would change its behavior at runtime based + # on these configuration settings. We simulate that behavior here + # by checking the values and returning a new KeepClient if any of + # them have changed. + key = (config.get('ARVADOS_API_HOST'), + config.get('ARVADOS_API_TOKEN'), + config.flag_is_true('ARVADOS_API_HOST_INSECURE'), + config.get('ARVADOS_KEEP_PROXY'), + config.get('ARVADOS_EXTERNAL_CLIENT') == 'true', + os.environ.get('KEEP_LOCAL_STORE')) + if (global_client_object is None) or (cls._last_key != key): global_client_object = KeepClient() + cls._last_key = key return global_client_object @staticmethod @@ -125,8 +141,8 @@ class Keep: def put(data, **kwargs): return Keep.global_client_object().put(data, **kwargs) -class KeepClient(object): +class KeepClient(object): class ThreadLimiter(object): """ Limit the number of threads running at a given time to @@ -183,15 +199,21 @@ class KeepClient(object): with self._done_lock: return self._done + class KeepWriterThread(threading.Thread): """ 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): + def __init__(self, api_token, **kwargs): super(KeepClient.KeepWriterThread, self).__init__() + self._api_token = api_token self.args = kwargs + self._success = False + + def success(self): + return self._success def run(self): with self.args['thread_limiter'] as limiter: @@ -199,101 +221,149 @@ 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: - logging.debug("Uploading to {}".format(url)) + 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'] + headers = {'Authorization': "OAuth2 %s" % (self._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'])) - 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 - return limiter.save_response(content.strip(), replicas_stored) - - 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.debug("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.debug("Request fail: PUT %s => %s: %s", + url, type(e), str(e)) + + + def __init__(self, api_client=None, proxy=None, timeout=60, + api_token=None, local_store=None): + """Initialize a new KeepClient. + + Arguments: + * api_client: The API client to use to find Keep services. If not + provided, KeepClient will build one from available Arvados + configuration. + * proxy: If specified, this KeepClient will send requests to this + Keep proxy. Otherwise, KeepClient will fall back to the setting + of the ARVADOS_KEEP_PROXY configuration setting. If you want to + ensure KeepClient does not use a proxy, pass in an empty string. + * timeout: The timeout for all HTTP requests, in seconds. Default + 60. + * api_token: If you're not using an API client, but only talking + directly to a Keep proxy, this parameter specifies an API token + to authenticate Keep requests. It is an error to specify both + api_client and api_token. If you specify neither, KeepClient + will use one available from the Arvados configuration. + * local_store: If specified, this KeepClient will bypass Keep + services, and save data to the named directory. If unspecified, + KeepClient will fall back to the setting of the $KEEP_LOCAL_STORE + environment variable. If you want to ensure KeepClient does not + use local storage, pass in an empty string. This is primarily + intended to mock a server for testing. + """ self.lock = threading.Lock() - self.service_roots = None - self._cache_lock = threading.Lock() - self._cache = [] - # default 256 megabyte cache - self.cache_max = 256 * 1024 * 1024 - self.using_proxy = False + if proxy is None: + proxy = config.get('ARVADOS_KEEP_PROXY') + if api_token is None: + api_token = config.get('ARVADOS_API_TOKEN') + elif api_client is not None: + raise ValueError( + "can't build KeepClient with both API client and token") + if local_store is None: + local_store = os.environ.get('KEEP_LOCAL_STORE') + + if local_store: + self.local_store = local_store + self.get = self.local_store_get + self.put = self.local_store_put + else: + self.timeout = timeout + self.cache_max = 256 * 1024 * 1024 # Cache is 256MiB + self._cache = [] + self._cache_lock = threading.Lock() + if proxy: + if not proxy.endswith('/'): + proxy += '/' + self.api_token = api_token + self.service_roots = [proxy] + self.using_proxy = True + else: + # It's important to avoid instantiating an API client + # unless we actually need one, for testing's sake. + if api_client is None: + api_client = arvados.api('v1') + self.api_client = api_client + self.api_token = api_client.api_token + self.service_roots = None + self.using_proxy = None def shuffled_service_roots(self, hash): - if self.service_roots == None: - self.lock.acquire() + if self.service_roots is None: + with self.lock: + try: + keep_services = self.api_client.keep_services().accessible() + except Exception: # API server predates Keep services. + keep_services = self.api_client.keep_disks().list() - # Override normal keep disk lookup with an explict proxy - # configuration. - keep_proxy_env = config.get("ARVADOS_KEEP_PROXY") - if keep_proxy_env != None and len(keep_proxy_env) > 0: + keep_services = keep_services.execute().get('items') + if not keep_services: + raise arvados.errors.NoKeepServersError() - if keep_proxy_env[-1:] != '/': - keep_proxy_env += "/" - self.service_roots = [keep_proxy_env] - self.using_proxy = True - else: - try: - 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_services) - self.service_roots = sorted(set(roots)) - logging.debug(str(self.service_roots)) - finally: - self.lock.release() + self.using_proxy = (keep_services[0].get('service_type') == + 'proxy') + + roots = (("http%s://%s:%d/" % + ('s' if f['service_ssl_flag'] else '', + f['service_host'], + f['service_port'])) + for f in keep_services) + self.service_roots = sorted(set(roots)) + _logger.debug(str(self.service_roots)) # Build an ordering with which to query the Keep servers based on the # contents of the hash. @@ -336,7 +406,7 @@ class KeepClient(object): # Remove the digits just used from the seed seed = seed[8:] - logging.debug(str(pseq)) + _logger.debug(str(pseq)) return pseq class CacheSlot(object): @@ -393,17 +463,13 @@ class KeepClient(object): finally: 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: - return KeepClient.local_store_get(locator) - expect_hash = re.sub(r'\+.*', '', locator) + def get(self, loc_s): + if ',' in loc_s: + return ''.join(self.get(x) for x in loc_s.split(',')) + locator = KeepLocator(loc_s) + expect_hash = locator.md5sum slot, first = self.reserve_cache(expect_hash) - #logging.debug("%s %s %s" % (slot, first, expect_hash)) if not first: v = slot.get() @@ -411,9 +477,8 @@ class KeepClient(object): try: for service_root in self.shuffled_service_roots(expect_hash): - url = service_root + locator - api_token = config.get('ARVADOS_API_TOKEN') - headers = {'Authorization': "OAuth2 %s" % api_token, + url = service_root + loc_s + headers = {'Authorization': "OAuth2 %s" % (self.api_token,), 'Accept': 'application/octet-stream'} blob = self.get_url(url, headers, expect_hash) if blob: @@ -421,9 +486,10 @@ class KeepClient(object): self.cap_cache() return blob - for location_hint in re.finditer(r'\+K@([a-z0-9]+)', locator): - instance = location_hint.group(1) - url = 'http://keep.' + instance + '.arvadosapi.com/' + locator + for hint in locator.hints: + if not hint.startswith('K@'): + continue + url = 'http://keep.' + hint[2:] + '.arvadosapi.com/' + loc_s blob = self.get_url(url, {}, expect_hash) if blob: slot.set(blob) @@ -441,48 +507,61 @@ class KeepClient(object): def get_url(self, url, headers, expect_hash): h = httplib2.Http() try: - logging.info("Request: GET %s" % (url)) + _logger.debug("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() + md5 = hashlib.md5(content).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.debug("Request fail: GET %s => %s: %s", + url, type(e), str(e)) return None - def put(self, data, **kwargs): - if 'KEEP_LOCAL_STORE' in os.environ: - return KeepClient.local_store_put(data) - m = hashlib.new('md5') - m.update(data) - data_hash = m.hexdigest() + def put(self, data, copies=2): + data_hash = hashlib.md5(data).hexdigest() have_copies = 0 - want_copies = kwargs.get('copies', 2) + want_copies = copies if not (want_copies > 0): return data_hash 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( + self.api_token, + 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.debug("Retrying: PUT %s %s", + t.args['service_root'], + t.args['data_hash']) + retry_with_args = t.args.copy() + t_retry = KeepClient.KeepWriterThread(self.api_token, + **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: @@ -495,26 +574,22 @@ class KeepClient(object): def sign_for_old_server(data_hash, data): return (("-----BEGIN PGP SIGNED MESSAGE-----\n\n\n%d %s\n-----BEGIN PGP SIGNATURE-----\n\n-----END PGP SIGNATURE-----\n" % (int(time.time()), data_hash)) + data) - - @staticmethod - def local_store_put(data): - m = hashlib.new('md5') - m.update(data) - md5 = m.hexdigest() + def local_store_put(self, data): + md5 = hashlib.md5(data).hexdigest() locator = '%s+%d' % (md5, len(data)) - with open(os.path.join(os.environ['KEEP_LOCAL_STORE'], md5 + '.tmp'), 'w') as f: + with open(os.path.join(self.local_store, md5 + '.tmp'), 'w') as f: f.write(data) - os.rename(os.path.join(os.environ['KEEP_LOCAL_STORE'], md5 + '.tmp'), - os.path.join(os.environ['KEEP_LOCAL_STORE'], md5)) + os.rename(os.path.join(self.local_store, md5 + '.tmp'), + os.path.join(self.local_store, md5)) return locator - @staticmethod - def local_store_get(locator): - r = re.search('^([0-9a-f]{32,})', locator) - if not r: + def local_store_get(self, loc_s): + try: + locator = KeepLocator(loc_s) + except ValueError: raise arvados.errors.NotFoundError( - "Invalid data locator: '%s'" % locator) - if r.group(0) == config.EMPTY_BLOCK_LOCATOR.split('+')[0]: + "Invalid data locator: '%s'" % loc_s) + if locator.md5sum == config.EMPTY_BLOCK_LOCATOR.split('+')[0]: return '' - with open(os.path.join(os.environ['KEEP_LOCAL_STORE'], r.group(0)), 'r') as f: + with open(os.path.join(self.local_store, locator.md5sum), 'r') as f: return f.read()