X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/68dae835a2f8bebd5664211a60fc76d50e5ecc23..1fa1006e5b21cfc30e694890db969c20864af8c6:/sdk/python/arvados/keep.py diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py index 46bd1cb464..d92758062d 100644 --- a/sdk/python/arvados/keep.py +++ b/sdk/python/arvados/keep.py @@ -1,6 +1,4 @@ import gflags -import httplib -import httplib2 import logging import os import pprint @@ -21,6 +19,7 @@ import timer import datetime import ssl import socket +import requests _logger = logging.getLogger('arvados.keep') global_client_object = None @@ -210,6 +209,14 @@ class KeepBlockCache(object): self._cache_lock.release() class KeepClient(object): + + # Default Keep server connection timeout: 3 seconds + # Default Keep server read timeout: 30 seconds + # Default Keep proxy connection timeout: 20 seconds + # Default Keep proxy read timeout: 60 seconds + DEFAULT_TIMEOUT = (3, 30) + DEFAULT_PROXY_TIMEOUT = (20, 60) + class ThreadLimiter(object): """ Limit the number of threads running at a given time to @@ -269,10 +276,10 @@ class KeepClient(object): class KeepService(object): # Make requests to a single Keep service, and track results. - HTTP_ERRORS = (httplib2.HttpLib2Error, httplib.HTTPException, + HTTP_ERRORS = (requests.exceptions.RequestException, socket.error, ssl.SSLError) - def __init__(self, root, **headers): + def __init__(self, root, timeout=None, **headers): self.root = root self.last_result = None self.success_flag = None @@ -288,19 +295,19 @@ class KeepClient(object): def last_status(self): try: - return int(self.last_result[0].status) + return self.last_result.status_code except (AttributeError, IndexError, ValueError): return None - def get(self, http, locator): - # http is an httplib2.Http object. + def get(self, locator, timeout=None): # locator is a KeepLocator object. url = self.root + str(locator) _logger.debug("Request: GET %s", url) try: with timer.Timer() as t: - result = http.request(url.encode('utf-8'), 'GET', - headers=self.get_headers) + result = requests.get(url.encode('utf-8'), + headers=self.get_headers, + timeout=timeout) except self.HTTP_ERRORS as e: _logger.debug("Request fail: GET %s => %s: %s", url, type(e), str(e)) @@ -308,7 +315,7 @@ class KeepClient(object): else: self.last_result = result self.success_flag = retry.check_http_response_success(result) - content = result[1] + content = result.content _logger.info("%s response: %s bytes in %s msec (%.3f MiB/sec)", self.last_status(), len(content), t.msecs, (len(content)/(1024.0*1024))/t.secs) @@ -320,12 +327,14 @@ class KeepClient(object): url, resp_md5) return None - def put(self, http, hash_s, body): + def put(self, hash_s, body, timeout=None): url = self.root + hash_s _logger.debug("Request: PUT %s", url) try: - result = http.request(url.encode('utf-8'), 'PUT', - headers=self.put_headers, body=body) + result = requests.put(url.encode('utf-8'), + data=body, + headers=self.put_headers, + timeout=timeout) except self.HTTP_ERRORS as e: _logger.debug("Request fail: PUT %s => %s: %s", url, type(e), str(e)) @@ -366,12 +375,13 @@ class KeepClient(object): str(threading.current_thread()), self.args['data_hash'], self.args['service_root']) - h = httplib2.Http(timeout=self.args.get('timeout', None)) self._success = bool(self.service.put( - h, self.args['data_hash'], self.args['data'])) + self.args['data_hash'], + self.args['data'], + timeout=self.args.get('timeout', None))) status = self.service.last_status() if self._success: - resp, body = self.service.last_result + result = self.service.last_result _logger.debug("KeepWriterThread %s succeeded %s %s", str(threading.current_thread()), self.args['data_hash'], @@ -381,17 +391,18 @@ class KeepClient(object): # 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']) + replicas_stored = int(result.headers['x-keep-replicas-stored']) except (KeyError, ValueError): replicas_stored = 1 - limiter.save_response(body.strip(), replicas_stored) + limiter.save_response(result.text.strip(), replicas_stored) elif status is not None: _logger.debug("Request fail: PUT %s => %s %s", self.args['data_hash'], status, - self.service.last_result[1]) + self.service.last_result.text) - def __init__(self, api_client=None, proxy=None, timeout=300, + def __init__(self, api_client=None, proxy=None, + timeout=DEFAULT_TIMEOUT, proxy_timeout=DEFAULT_PROXY_TIMEOUT, api_token=None, local_store=None, block_cache=None, num_retries=0): """Initialize a new KeepClient. @@ -404,8 +415,14 @@ class KeepClient(object): 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 - 300. + * timeout: The timeout (in seconds) for HTTP requests to Keep + non-proxy servers. A tuple of two floats is interpreted as + (connection_timeout, read_timeout): see + http://docs.python-requests.org/en/latest/user/advanced/#timeouts. + Default: (3, 30). + * proxy_timeout: The timeout (in seconds) for HTTP requests to + Keep proxies. A tuple of two floats is interpreted as + (connection_timeout, read_timeout). Default: (20, 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 @@ -420,6 +437,7 @@ class KeepClient(object): * num_retries: The default number of times to retry failed requests. This will be used as the default num_retries value when get() and put() are called. Default 0. + """ self.lock = threading.Lock() if proxy is None: @@ -442,7 +460,6 @@ class KeepClient(object): self.get = self.local_store_get self.put = self.local_store_put else: - self.timeout = timeout self.num_retries = num_retries if proxy: if not proxy.endswith('/'): @@ -450,6 +467,7 @@ class KeepClient(object): self.api_token = api_token self.service_roots = [proxy] self.using_proxy = True + self.timeout = proxy_timeout self.static_service_roots = True else: # It's important to avoid instantiating an API client @@ -460,6 +478,7 @@ class KeepClient(object): self.api_token = api_client.api_token self.service_roots = None self.using_proxy = None + self.timeout = timeout self.static_service_roots = False def build_service_roots(self, force_rebuild=False): @@ -622,9 +641,8 @@ class KeepClient(object): services_to_try = [roots_map[root] for root in (local_roots + hint_roots) if roots_map[root].usable()] - http = httplib2.Http(timeout=self.timeout) for keep_service in services_to_try: - blob = keep_service.get(http, locator) + blob = keep_service.get(locator, timeout=self.timeout) if blob is not None: break loop.save_result((blob, len(services_to_try)))