X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/b112b002ee3239803d1948e99463144812a2c213..6fe8e52020d421797306e5c6536afbcee761510a:/sdk/python/arvados/keep.py diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py index 351f7f5dda..71e101cf4c 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={}): + headers={}, + insecure=False): self.root = root self._user_agent_pool = user_agent_pool self._result = {'error': None} @@ -304,6 +305,7 @@ class KeepClient(object): self.put_headers = headers self.upload_counter = upload_counter self.download_counter = download_counter + self.insecure = insecure def usable(self): """Is it worth attempting a request?""" @@ -371,6 +373,8 @@ class KeepClient(object): '{}: {}'.format(k,v) for k,v in self.get_headers.items()]) curl.setopt(pycurl.WRITEFUNCTION, response_body.write) curl.setopt(pycurl.HEADERFUNCTION, self._headerfunction) + if self.insecure: + curl.setopt(pycurl.SSL_VERIFYPEER, 0) if method == "HEAD": curl.setopt(pycurl.NOBODY, True) self._setcurltimeouts(curl, timeout) @@ -463,6 +467,8 @@ class KeepClient(object): '{}: {}'.format(k,v) for k,v in self.put_headers.items()]) curl.setopt(pycurl.WRITEFUNCTION, response_body.write) curl.setopt(pycurl.HEADERFUNCTION, self._headerfunction) + if self.insecure: + curl.setopt(pycurl.SSL_VERIFYPEER, 0) self._setcurltimeouts(curl, timeout) try: curl.perform() @@ -541,7 +547,7 @@ class KeepClient(object): self._lastheadername = name self._headers[name] = value # Returning None implies all bytes were written - + class KeepWriterQueue(queue.Queue): def __init__(self, copies): @@ -552,19 +558,19 @@ class KeepClient(object): self.successful_copies_lock = threading.Lock() self.pending_tries = copies self.pending_tries_notification = threading.Condition() - + def write_success(self, response, replicas_nr): with self.successful_copies_lock: self.successful_copies += replicas_nr self.response = response with self.pending_tries_notification: self.pending_tries_notification.notify_all() - + def write_fail(self, ks): with self.pending_tries_notification: self.pending_tries += 1 self.pending_tries_notification.notify() - + def pending_copies(self): with self.successful_copies_lock: return self.wanted_copies - self.successful_copies @@ -613,25 +619,25 @@ class KeepClient(object): for _ in range(num_threads): w = KeepClient.KeepWriterThread(self.queue, data, data_hash, timeout) self.workers.append(w) - + def add_task(self, ks, service_root): self.queue.put((ks, service_root)) self.total_task_nr += 1 - + def done(self): return self.queue.successful_copies - + def join(self): # Start workers for worker in self.workers: worker.start() # Wait for finished work self.queue.join() - + def response(self): return self.queue.response - - + + class KeepWriterThread(threading.Thread): TaskFailed = RuntimeError() @@ -762,6 +768,11 @@ class KeepClient(object): if local_store is None: local_store = os.environ.get('KEEP_LOCAL_STORE') + if api_client is None: + self.insecure = config.flag_is_true('ARVADOS_API_HOST_INSECURE') + else: + self.insecure = api_client.insecure + self.block_cache = block_cache if block_cache else KeepBlockCache() self.timeout = timeout self.proxy_timeout = proxy_timeout @@ -934,7 +945,8 @@ class KeepClient(object): root, self._user_agent_pool, upload_counter=self.upload_counter, download_counter=self.download_counter, - headers=headers) + headers=headers, + insecure=self.insecure) return local_roots @staticmethod @@ -996,84 +1008,91 @@ class KeepClient(object): self.get_counter.add(1) - locator = KeepLocator(loc_s) - if method == "GET": - slot, first = self.block_cache.reserve_cache(locator.md5sum) - if not first: - self.hits_counter.add(1) - v = slot.get() - return v - - 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 the locator has hints specifying a prefix (indicating a - # remote keepproxy) or the UUID of a local gateway service, - # read data from the indicated service(s) instead of the usual - # list of local disk services. - hint_roots = ['http://keep.{}.arvadosapi.com/'.format(hint[2:]) - for hint in locator.hints if hint.startswith('K@') and len(hint) == 7] - hint_roots.extend([self._gateway_services[hint[2:]]['_service_root'] - for hint in locator.hints if ( - hint.startswith('K@') and - len(hint) == 29 and - self._gateway_services.get(hint[2:]) - )]) - # Map root URLs to their KeepService objects. - roots_map = { - root: self.KeepService(root, self._user_agent_pool, - upload_counter=self.upload_counter, - download_counter=self.download_counter, - headers=headers) - for root in hint_roots - } - - # See #3147 for a discussion of the loop implementation. Highlights: - # * Refresh the list of Keep services after each failure, in case - # it's being updated. - # * Retry until we succeed, we're out of retries, or every available - # service has returned permanent failure. - sorted_roots = [] - roots_map = {} + slot = None blob = None - loop = retry.RetryLoop(num_retries, self._check_loop_result, - backoff_start=2) - for tries_left in loop: - try: - sorted_roots = self.map_new_services( - roots_map, locator, - force_rebuild=(tries_left < num_retries), - need_writable=False, - headers=headers) - except Exception as error: - loop.save_result(error) - continue + try: + locator = KeepLocator(loc_s) + if method == "GET": + slot, first = self.block_cache.reserve_cache(locator.md5sum) + if not first: + self.hits_counter.add(1) + blob = slot.get() + if blob is None: + raise arvados.errors.KeepReadError( + "failed to read {}".format(loc_s)) + return blob + + 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 the locator has hints specifying a prefix (indicating a + # remote keepproxy) or the UUID of a local gateway service, + # read data from the indicated service(s) instead of the usual + # list of local disk services. + hint_roots = ['http://keep.{}.arvadosapi.com/'.format(hint[2:]) + for hint in locator.hints if hint.startswith('K@') and len(hint) == 7] + hint_roots.extend([self._gateway_services[hint[2:]]['_service_root'] + for hint in locator.hints if ( + hint.startswith('K@') and + len(hint) == 29 and + self._gateway_services.get(hint[2:]) + )]) + # Map root URLs to their KeepService objects. + roots_map = { + root: self.KeepService(root, self._user_agent_pool, + upload_counter=self.upload_counter, + download_counter=self.download_counter, + headers=headers, + insecure=self.insecure) + for root in hint_roots + } + + # See #3147 for a discussion of the loop implementation. Highlights: + # * Refresh the list of Keep services after each failure, in case + # it's being updated. + # * Retry until we succeed, we're out of retries, or every available + # service has returned permanent failure. + sorted_roots = [] + roots_map = {} + loop = retry.RetryLoop(num_retries, self._check_loop_result, + backoff_start=2) + for tries_left in loop: + try: + sorted_roots = self.map_new_services( + roots_map, locator, + force_rebuild=(tries_left < num_retries), + need_writable=False, + headers=headers) + except Exception as error: + loop.save_result(error) + continue - # Query KeepService objects that haven't returned - # permanent failure, in our specified shuffle order. - services_to_try = [roots_map[root] - for root in sorted_roots - if roots_map[root].usable()] - for keep_service in services_to_try: - blob = keep_service.get(locator, method=method, timeout=self.current_timeout(num_retries-tries_left)) - if blob is not None: - break - loop.save_result((blob, len(services_to_try))) - - # Always cache the result, then return it if we succeeded. - if method == "GET": - slot.set(blob) - self.block_cache.cap_cache() - if loop.success(): - if method == "HEAD": - return True - else: - return blob + # Query KeepService objects that haven't returned + # permanent failure, in our specified shuffle order. + services_to_try = [roots_map[root] + for root in sorted_roots + if roots_map[root].usable()] + for keep_service in services_to_try: + blob = keep_service.get(locator, method=method, timeout=self.current_timeout(num_retries-tries_left)) + if blob is not None: + break + loop.save_result((blob, len(services_to_try))) + + # Always cache the result, then return it if we succeeded. + if loop.success(): + if method == "HEAD": + return True + else: + return blob + finally: + if slot is not None: + slot.set(blob) + self.block_cache.cap_cache() # Q: Including 403 is necessary for the Keep tests to continue # passing, but maybe they should expect KeepReadError instead? @@ -1144,7 +1163,7 @@ class KeepClient(object): loop.save_result(error) continue - writer_pool = KeepClient.KeepWriterThreadPool(data=data, + writer_pool = KeepClient.KeepWriterThreadPool(data=data, data_hash=data_hash, copies=copies - done, max_service_replicas=self.max_replicas_per_service,