X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/088268fc00cb466c4b4576a2dbbaff720cb3e3af..42fff42165a0fa1602758a078746f8697f265f83:/sdk/python/arvados/keep.py diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py index ee1850ff34..bd0e5dc1e6 100644 --- a/sdk/python/arvados/keep.py +++ b/sdk/python/arvados/keep.py @@ -292,7 +292,8 @@ class KeepClient(object): def __init__(self, root, user_agent_pool=queue.LifoQueue(), upload_counter=None, download_counter=None, - headers={},insecure=False): + headers={}, + insecure=False): self.root = root self._user_agent_pool = user_agent_pool self._result = {'error': None} @@ -374,9 +375,11 @@ class KeepClient(object): curl.setopt(pycurl.HEADERFUNCTION, self._headerfunction) if self.insecure: curl.setopt(pycurl.SSL_VERIFYPEER, 0) + else: + curl.setopt(pycurl.CAINFO, arvados.util.ca_certs_path()) if method == "HEAD": curl.setopt(pycurl.NOBODY, True) - self._setcurltimeouts(curl, timeout) + self._setcurltimeouts(curl, timeout, method=="HEAD") try: curl.perform() @@ -420,6 +423,10 @@ class KeepClient(object): _logger.info("HEAD %s: %s bytes", self._result['status_code'], self._result.get('content-length')) + if self._result['headers'].get('x-keep-locator'): + # This is a response to a remote block copy request, return + # the local copy block locator. + return self._result['headers'].get('x-keep-locator') return True _logger.info("GET %s: %s bytes in %s msec (%.3f MiB/sec)", @@ -468,6 +475,8 @@ class KeepClient(object): curl.setopt(pycurl.HEADERFUNCTION, self._headerfunction) if self.insecure: curl.setopt(pycurl.SSL_VERIFYPEER, 0) + else: + curl.setopt(pycurl.CAINFO, arvados.util.ca_certs_path()) self._setcurltimeouts(curl, timeout) try: curl.perform() @@ -511,7 +520,7 @@ class KeepClient(object): self.upload_counter.add(len(body)) return True - def _setcurltimeouts(self, curl, timeouts): + def _setcurltimeouts(self, curl, timeouts, ignore_bandwidth=False): if not timeouts: return elif isinstance(timeouts, tuple): @@ -524,8 +533,9 @@ class KeepClient(object): conn_t, xfer_t = (timeouts, timeouts) bandwidth_bps = KeepClient.DEFAULT_TIMEOUT[2] curl.setopt(pycurl.CONNECTTIMEOUT_MS, int(conn_t*1000)) - curl.setopt(pycurl.LOW_SPEED_TIME, int(math.ceil(xfer_t))) - curl.setopt(pycurl.LOW_SPEED_LIMIT, int(math.ceil(bandwidth_bps))) + if not ignore_bandwidth: + curl.setopt(pycurl.LOW_SPEED_TIME, int(math.ceil(xfer_t))) + curl.setopt(pycurl.LOW_SPEED_LIMIT, int(math.ceil(bandwidth_bps))) def _headerfunction(self, header_line): if isinstance(header_line, bytes): @@ -638,7 +648,7 @@ class KeepClient(object): class KeepWriterThread(threading.Thread): - TaskFailed = RuntimeError() + class TaskFailed(RuntimeError): pass def __init__(self, queue, data, data_hash, timeout=None): super(KeepClient.KeepWriterThread, self).__init__() @@ -657,7 +667,7 @@ class KeepClient(object): try: locator, copies = self.do_task(service, service_root) except Exception as e: - if e is not self.TaskFailed: + if not isinstance(e, self.TaskFailed): _logger.exception("Exception in KeepWriterThread") self.queue.write_fail(service) else: @@ -677,7 +687,7 @@ class KeepClient(object): self.data_hash, result['status_code'], result['body']) - raise self.TaskFailed + raise self.TaskFailed() _logger.debug("KeepWriterThread %s succeeded %s+%i %s", str(threading.current_thread()), @@ -767,14 +777,10 @@ class KeepClient(object): if local_store is None: local_store = os.environ.get('KEEP_LOCAL_STORE') - if config.flag_is_true('ARVADOS_API_HOST_INSECURE'): - self.insecure = True + if api_client is None: + self.insecure = config.flag_is_true('ARVADOS_API_HOST_INSECURE') else: - self.insecure = False - - if api_client is not None: - if not self.insecure and api_client.insecure: - self.insecure = True + self.insecure = api_client.insecure self.block_cache = block_cache if block_cache else KeepBlockCache() self.timeout = timeout @@ -789,6 +795,7 @@ class KeepClient(object): if local_store: self.local_store = local_store + self.head = self.local_store_head self.get = self.local_store_get self.put = self.local_store_put else: @@ -948,7 +955,8 @@ class KeepClient(object): root, self._user_agent_pool, upload_counter=self.upload_counter, download_counter=self.download_counter, - headers=headers,insecure=self.insecure) + headers=headers, + insecure=self.insecure) return local_roots @staticmethod @@ -977,6 +985,11 @@ class KeepClient(object): else: return None + def refresh_signature(self, loc): + """Ask Keep to get the remote block and return its local signature""" + now = datetime.datetime.utcnow().isoformat("T") + 'Z' + return self.head(loc, headers={'X-Keep-Signature': 'local, {}'.format(now)}) + @retry.retry_method def head(self, loc_s, **kwargs): return self._get_or_head(loc_s, method="HEAD", **kwargs) @@ -985,7 +998,7 @@ class KeepClient(object): def get(self, loc_s, **kwargs): return self._get_or_head(loc_s, method="GET", **kwargs) - def _get_or_head(self, loc_s, method="GET", num_retries=None, request_id=None): + def _get_or_head(self, loc_s, method="GET", num_retries=None, request_id=None, headers=None): """Get data from Keep. This method fetches one or more blocks of data from Keep. It @@ -1026,11 +1039,11 @@ class KeepClient(object): self.misses_counter.add(1) - headers = { - 'X-Request-Id': (request_id or - (hasattr(self, 'api_client') and self.api_client.request_id) or - arvados.util.new_request_id()), - } + if headers is None: + headers = {} + headers['X-Request-Id'] = (request_id or + (hasattr(self, 'api_client') and self.api_client.request_id) or + arvados.util.new_request_id()) # If the locator has hints specifying a prefix (indicating a # remote keepproxy) or the UUID of a local gateway service, @@ -1049,7 +1062,8 @@ class KeepClient(object): root: self.KeepService(root, self._user_agent_pool, upload_counter=self.upload_counter, download_counter=self.download_counter, - headers=headers,insecure=self.insecure) + headers=headers, + insecure=self.insecure) for root in hint_roots } @@ -1086,10 +1100,7 @@ class KeepClient(object): # Always cache the result, then return it if we succeeded. if loop.success(): - if method == "HEAD": - return True - else: - return blob + return blob finally: if slot is not None: slot.set(blob) @@ -1110,7 +1121,7 @@ class KeepClient(object): "{} not found".format(loc_s), service_errors) else: raise arvados.errors.KeepReadError( - "failed to read {}".format(loc_s), service_errors, label="service") + "failed to read {} after {}".format(loc_s, loop.attempts_str()), service_errors, label="service") @retry.retry_method def put(self, data, copies=2, num_retries=None, request_id=None): @@ -1189,8 +1200,8 @@ class KeepClient(object): for key in sorted_roots if roots_map[key].last_result()['error']) raise arvados.errors.KeepWriteError( - "failed to write {} (wanted {} copies but wrote {})".format( - data_hash, copies, writer_pool.done()), service_errors, label="service") + "failed to write {} after {} (wanted {} copies but wrote {})".format( + data_hash, loop.attempts_str(), copies, writer_pool.done()), service_errors, label="service") def local_store_put(self, data, copies=1, num_retries=None): """A stub for put(). @@ -1224,5 +1235,17 @@ class KeepClient(object): with open(os.path.join(self.local_store, locator.md5sum), 'rb') as f: return f.read() + def local_store_head(self, loc_s, num_retries=None): + """Companion to local_store_put().""" + try: + locator = KeepLocator(loc_s) + except ValueError: + raise arvados.errors.NotFoundError( + "Invalid data locator: '%s'" % loc_s) + if locator.md5sum == config.EMPTY_BLOCK_LOCATOR.split('+')[0]: + return True + if os.path.exists(os.path.join(self.local_store, locator.md5sum)): + return True + def is_cached(self, locator): return self.block_cache.reserve_cache(expect_hash)