X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/f81a5a1adb403f9c81159239e92880843dca3c93..af901a9a0434c9b5ed8b9349835a1749f45f95d3:/sdk/python/arvados/keep.py diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py index c0c5c7ee47..1b076b61ae 100644 --- a/sdk/python/arvados/keep.py +++ b/sdk/python/arvados/keep.py @@ -14,7 +14,7 @@ import re import socket import ssl import string -import StringIO +import cStringIO import subprocess import sys import threading @@ -76,7 +76,7 @@ class KeepLocator(object): 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: {}". + raise ValueError("{} is not a {}-digit hex string: {}". format(name, length, hex_str)) setattr(self, data_name, hex_str) return property(getter, setter) @@ -357,7 +357,7 @@ class KeepClient(object): try: with timer.Timer() as t: self._headers = {} - response_body = StringIO.StringIO() + response_body = cStringIO.StringIO() curl.setopt(pycurl.NOSIGNAL, 1) curl.setopt(pycurl.OPENSOCKETFUNCTION, self._socket_open) curl.setopt(pycurl.URL, url.encode('utf-8')) @@ -420,12 +420,20 @@ class KeepClient(object): curl = self._get_user_agent() try: self._headers = {} - response_body = StringIO.StringIO() + body_reader = cStringIO.StringIO(body) + response_body = cStringIO.StringIO() curl.setopt(pycurl.NOSIGNAL, 1) curl.setopt(pycurl.OPENSOCKETFUNCTION, self._socket_open) curl.setopt(pycurl.URL, url.encode('utf-8')) - curl.setopt(pycurl.POSTFIELDS, body) - curl.setopt(pycurl.CUSTOMREQUEST, 'PUT') + # Using UPLOAD tells cURL to wait for a "go ahead" from the + # Keep server (in the form of a HTTP/1.1 "100 Continue" + # response) instead of sending the request body immediately. + # This allows the server to reject the request if the request + # is invalid or the server is read-only, without waiting for + # the client to send the entire block. + curl.setopt(pycurl.UPLOAD, True) + curl.setopt(pycurl.INFILESIZE, len(body)) + curl.setopt(pycurl.READFUNCTION, body_reader.read) curl.setopt(pycurl.HTTPHEADER, [ '{}: {}'.format(k,v) for k,v in self.put_headers.iteritems()]) curl.setopt(pycurl.WRITEFUNCTION, response_body.write) @@ -640,8 +648,10 @@ class KeepClient(object): 'uuid': 'proxy', '_service_root': proxy, }] + self._writable_services = self._keep_services self.using_proxy = True self._static_services_list = True + self.max_replicas_per_service = None else: # It's important to avoid instantiating an API client # unless we actually need one, for testing's sake. @@ -651,8 +661,10 @@ class KeepClient(object): self.api_token = api_client.api_token self._gateway_services = {} self._keep_services = None + self._writable_services = None self.using_proxy = None self._static_services_list = False + self.max_replicas_per_service = None def current_timeout(self, attempt_number): """Return the appropriate timeout to use for this client. @@ -703,10 +715,19 @@ class KeepClient(object): self._keep_services = [ ks for ks in accessible if ks.get('service_type') in ['disk', 'proxy']] + self._writable_services = [ + ks for ks in accessible + if (ks.get('service_type') in ['disk', 'proxy']) and (True != ks.get('read_only'))] _logger.debug(str(self._keep_services)) self.using_proxy = any(ks.get('service_type') == 'proxy' for ks in self._keep_services) + # Set max_replicas_per_service to 1 for disk typed services. + # In case of, non-disk typed services, we will use None to mean unknown. + self.max_replicas_per_service = 1 + for ks in accessible: + if ('disk' != ks.get('service_type')) and not ks.get('read_only'): + self.max_replicas_per_service = None def _service_weight(self, data_hash, service_uuid): """Compute the weight of a Keep service endpoint for a data @@ -717,7 +738,7 @@ class KeepClient(object): """ return hashlib.md5(data_hash + service_uuid[-15:]).hexdigest() - def weighted_service_roots(self, locator, force_rebuild=False): + def weighted_service_roots(self, locator, force_rebuild=False, need_writable=False): """Return an array of Keep service endpoints, in the order in which they should be probed when reading or writing data with the given hash+hints. @@ -725,15 +746,14 @@ class KeepClient(object): self.build_services_list(force_rebuild) sorted_roots = [] - # Use the services indicated by the given +K@... remote # service hints, if any are present and can be resolved to a # URI. for hint in locator.hints: if hint.startswith('K@'): if len(hint) == 7: - sorted_roots.append( - "https://keep.{}.arvadosapi.com/".format(hint[2:])) + sorted_roots.append( + "https://keep.{}.arvadosapi.com/".format(hint[2:])) elif len(hint) == 29: svc = self._gateway_services.get(hint[2:]) if svc: @@ -742,21 +762,24 @@ class KeepClient(object): # Sort the available local services by weight (heaviest first) # for this locator, and return their service_roots (base URIs) # in that order. + use_services = self._keep_services + if need_writable: + use_services = self._writable_services sorted_roots.extend([ svc['_service_root'] for svc in sorted( - self._keep_services, + use_services, reverse=True, key=lambda svc: self._service_weight(locator.md5sum, svc['uuid']))]) _logger.debug("{}: {}".format(locator, sorted_roots)) return sorted_roots - def map_new_services(self, roots_map, locator, force_rebuild, **headers): + def map_new_services(self, roots_map, locator, force_rebuild, need_writable, **headers): # roots_map is a dictionary, mapping Keep service root strings # to KeepService objects. Poll for Keep services, and add any # new ones to roots_map. Return the current list of local # root strings. headers.setdefault('Authorization', "OAuth2 %s" % (self.api_token,)) - local_roots = self.weighted_service_roots(locator, force_rebuild) + local_roots = self.weighted_service_roots(locator, force_rebuild, need_writable) for root in local_roots: if root not in roots_map: roots_map[root] = self.KeepService( @@ -850,7 +873,8 @@ class KeepClient(object): try: sorted_roots = self.map_new_services( roots_map, locator, - force_rebuild=(tries_left < num_retries)) + force_rebuild=(tries_left < num_retries), + need_writable=False) except Exception as error: loop.save_result(error) continue @@ -912,26 +936,26 @@ class KeepClient(object): if isinstance(data, unicode): data = data.encode("ascii") elif not isinstance(data, str): - raise arvados.errors.ArgumentError("Argument 'data' to KeepClient.put must be type 'str'") + raise arvados.errors.ArgumentError("Argument 'data' to KeepClient.put is not type 'str'") data_hash = hashlib.md5(data).hexdigest() + loc_s = data_hash + '+' + str(len(data)) if copies < 1: - return data_hash - locator = KeepLocator(data_hash + '+' + str(len(data))) + return loc_s + locator = KeepLocator(loc_s) headers = {} - if self.using_proxy: - # Tell the proxy how many copies we want it to store - headers['X-Keep-Desired-Replication'] = str(copies) + # Tell the proxy how many copies we want it to store + headers['X-Keep-Desired-Replication'] = str(copies) roots_map = {} - thread_limiter = KeepClient.ThreadLimiter(copies) + thread_limiter = KeepClient.ThreadLimiter(1 if self.max_replicas_per_service is None else copies) loop = retry.RetryLoop(num_retries, self._check_loop_result, backoff_start=2) for tries_left in loop: try: local_roots = self.map_new_services( roots_map, locator, - force_rebuild=(tries_left < num_retries), **headers) + force_rebuild=(tries_left < num_retries), need_writable=True, **headers) except Exception as error: loop.save_result(error) continue