sdk/python/arvados/keep.py

   1 import gflags
   2 import logging
   3 import os
   4 import pprint
   5 import sys
   6 import types
   7 import subprocess
   8 import json
   9 import UserDict
  10 import re
  11 import hashlib
  12 import string
  13 import bz2
  14 import zlib
  15 import fcntl
  16 import time
  17 import threading
  18 import timer
  19 import datetime
  20 import ssl
  21 import socket
  22 import requests
  23
  24 import arvados
  25 import arvados.config as config
  26 import arvados.errors
  27 import arvados.retry as retry
  28 import arvados.util
  29
  30 _logger = logging.getLogger('arvados.keep')
  31 global_client_object = None
  32
  33 class KeepLocator(object):
  34     EPOCH_DATETIME = datetime.datetime.utcfromtimestamp(0)
  35     HINT_RE = re.compile(r'^[A-Z][A-Za-z0-9@_-]+$')
  36
  37     def __init__(self, locator_str):
  38         self.hints = []
  39         self._perm_sig = None
  40         self._perm_expiry = None
  41         pieces = iter(locator_str.split('+'))
  42         self.md5sum = next(pieces)
  43         try:
  44             self.size = int(next(pieces))
  45         except StopIteration:
  46             self.size = None
  47         for hint in pieces:
  48             if self.HINT_RE.match(hint) is None:
  49                 raise ValueError("unrecognized hint data {}".format(hint))
  50             elif hint.startswith('A'):
  51                 self.parse_permission_hint(hint)
  52             else:
  53                 self.hints.append(hint)
  54
  55     def __str__(self):
  56         return '+'.join(
  57             str(s) for s in [self.md5sum, self.size,
  58                              self.permission_hint()] + self.hints
  59             if s is not None)
  60
  61     def _make_hex_prop(name, length):
  62         # Build and return a new property with the given name that
  63         # must be a hex string of the given length.
  64         data_name = '_{}'.format(name)
  65         def getter(self):
  66             return getattr(self, data_name)
  67         def setter(self, hex_str):
  68             if not arvados.util.is_hex(hex_str, length):
  69                 raise ValueError("{} must be a {}-digit hex string: {}".
  70                                  format(name, length, hex_str))
  71             setattr(self, data_name, hex_str)
  72         return property(getter, setter)
  73
  74     md5sum = _make_hex_prop('md5sum', 32)
  75     perm_sig = _make_hex_prop('perm_sig', 40)
  76
  77     @property
  78     def perm_expiry(self):
  79         return self._perm_expiry
  80
  81     @perm_expiry.setter
  82     def perm_expiry(self, value):
  83         if not arvados.util.is_hex(value, 1, 8):
  84             raise ValueError(
  85                 "permission timestamp must be a hex Unix timestamp: {}".
  86                 format(value))
  87         self._perm_expiry = datetime.datetime.utcfromtimestamp(int(value, 16))
  88
  89     def permission_hint(self):
  90         data = [self.perm_sig, self.perm_expiry]
  91         if None in data:
  92             return None
  93         data[1] = int((data[1] - self.EPOCH_DATETIME).total_seconds())
  94         return "A{}@{:08x}".format(*data)
  95
  96     def parse_permission_hint(self, s):
  97         try:
  98             self.perm_sig, self.perm_expiry = s[1:].split('@', 1)
  99         except IndexError:
 100             raise ValueError("bad permission hint {}".format(s))
 101
 102     def permission_expired(self, as_of_dt=None):
 103         if self.perm_expiry is None:
 104             return False
 105         elif as_of_dt is None:
 106             as_of_dt = datetime.datetime.now()
 107         return self.perm_expiry <= as_of_dt
 108
 109
 110 class Keep(object):
 111     """Simple interface to a global KeepClient object.
 112
 113     THIS CLASS IS DEPRECATED.  Please instantiate your own KeepClient with your
 114     own API client.  The global KeepClient will build an API client from the
 115     current Arvados configuration, which may not match the one you built.
 116     """
 117     _last_key = None
 118
 119     @classmethod
 120     def global_client_object(cls):
 121         global global_client_object
 122         # Previously, KeepClient would change its behavior at runtime based
 123         # on these configuration settings.  We simulate that behavior here
 124         # by checking the values and returning a new KeepClient if any of
 125         # them have changed.
 126         key = (config.get('ARVADOS_API_HOST'),
 127                config.get('ARVADOS_API_TOKEN'),
 128                config.flag_is_true('ARVADOS_API_HOST_INSECURE'),
 129                config.get('ARVADOS_KEEP_PROXY'),
 130                config.get('ARVADOS_EXTERNAL_CLIENT') == 'true',
 131                os.environ.get('KEEP_LOCAL_STORE'))
 132         if (global_client_object is None) or (cls._last_key != key):
 133             global_client_object = KeepClient()
 134             cls._last_key = key
 135         return global_client_object
 136
 137     @staticmethod
 138     def get(locator, **kwargs):
 139         return Keep.global_client_object().get(locator, **kwargs)
 140
 141     @staticmethod
 142     def put(data, **kwargs):
 143         return Keep.global_client_object().put(data, **kwargs)
 144
 145 class KeepBlockCache(object):
 146     # Default RAM cache is 256MiB
 147     def __init__(self, cache_max=(256 * 1024 * 1024)):
 148         self.cache_max = cache_max
 149         self._cache = []
 150         self._cache_lock = threading.Lock()
 151
 152     class CacheSlot(object):
 153         def __init__(self, locator):
 154             self.locator = locator
 155             self.ready = threading.Event()
 156             self.content = None
 157
 158         def get(self):
 159             self.ready.wait()
 160             return self.content
 161
 162         def set(self, value):
 163             self.content = value
 164             self.ready.set()
 165
 166         def size(self):
 167             if self.content is None:
 168                 return 0
 169             else:
 170                 return len(self.content)
 171
 172     def cap_cache(self):
 173         '''Cap the cache size to self.cache_max'''
 174         self._cache_lock.acquire()
 175         try:
 176             # Select all slots except those where ready.is_set() and content is
 177             # None (that means there was an error reading the block).
 178             self._cache = [c for c in self._cache if not (c.ready.is_set() and c.content is None)]
 179             sm = sum([slot.size() for slot in self._cache])
 180             while len(self._cache) > 0 and sm > self.cache_max:
 181                 for i in xrange(len(self._cache)-1, -1, -1):
 182                     if self._cache[i].ready.is_set():
 183                         del self._cache[i]
 184                         break
 185                 sm = sum([slot.size() for slot in self._cache])
 186         finally:
 187             self._cache_lock.release()
 188
 189     def reserve_cache(self, locator):
 190         '''Reserve a cache slot for the specified locator,
 191         or return the existing slot.'''
 192         self._cache_lock.acquire()
 193         try:
 194             # Test if the locator is already in the cache
 195             for i in xrange(0, len(self._cache)):
 196                 if self._cache[i].locator == locator:
 197                     n = self._cache[i]
 198                     if i != 0:
 199                         # move it to the front
 200                         del self._cache[i]
 201                         self._cache.insert(0, n)
 202                     return n, False
 203
 204             # Add a new cache slot for the locator
 205             n = KeepBlockCache.CacheSlot(locator)
 206             self._cache.insert(0, n)
 207             return n, True
 208         finally:
 209             self._cache_lock.release()
 210
 211 class KeepClient(object):
 212
 213     # Default Keep server connection timeout:  2 seconds
 214     # Default Keep server read timeout:      300 seconds
 215     # Default Keep proxy connection timeout:  20 seconds
 216     # Default Keep proxy read timeout:       300 seconds
 217     DEFAULT_TIMEOUT = (2, 300)
 218     DEFAULT_PROXY_TIMEOUT = (20, 300)
 219
 220     class ThreadLimiter(object):
 221         """
 222         Limit the number of threads running at a given time to
 223         {desired successes} minus {successes reported}. When successes
 224         reported == desired, wake up the remaining threads and tell
 225         them to quit.
 226
 227         Should be used in a "with" block.
 228         """
 229         def __init__(self, todo):
 230             self._todo = todo
 231             self._done = 0
 232             self._response = None
 233             self._todo_lock = threading.Semaphore(todo)
 234             self._done_lock = threading.Lock()
 235
 236         def __enter__(self):
 237             self._todo_lock.acquire()
 238             return self
 239
 240         def __exit__(self, type, value, traceback):
 241             self._todo_lock.release()
 242
 243         def shall_i_proceed(self):
 244             """
 245             Return true if the current thread should do stuff. Return
 246             false if the current thread should just stop.
 247             """
 248             with self._done_lock:
 249                 return (self._done < self._todo)
 250
 251         def save_response(self, response_body, replicas_stored):
 252             """
 253             Records a response body (a locator, possibly signed) returned by
 254             the Keep server.  It is not necessary to save more than
 255             one response, since we presume that any locator returned
 256             in response to a successful request is valid.
 257             """
 258             with self._done_lock:
 259                 self._done += replicas_stored
 260                 self._response = response_body
 261
 262         def response(self):
 263             """
 264             Returns the body from the response to a PUT request.
 265             """
 266             with self._done_lock:
 267                 return self._response
 268
 269         def done(self):
 270             """
 271             Return how many successes were reported.
 272             """
 273             with self._done_lock:
 274                 return self._done
 275
 276
 277     class KeepService(object):
 278         # Make requests to a single Keep service, and track results.
 279         HTTP_ERRORS = (requests.exceptions.RequestException,
 280                        socket.error, ssl.SSLError)
 281
 282         def __init__(self, root, **headers):
 283             self.root = root
 284             self.last_result = None
 285             self.success_flag = None
 286             self.get_headers = {'Accept': 'application/octet-stream'}
 287             self.get_headers.update(headers)
 288             self.put_headers = headers
 289
 290         def usable(self):
 291             return self.success_flag is not False
 292
 293         def finished(self):
 294             return self.success_flag is not None
 295
 296         def last_status(self):
 297             try:
 298                 return self.last_result.status_code
 299             except AttributeError:
 300                 return None
 301
 302         def get(self, locator, timeout=None):
 303             # locator is a KeepLocator object.
 304             url = self.root + str(locator)
 305             _logger.debug("Request: GET %s", url)
 306             try:
 307                 with timer.Timer() as t:
 308                     result = requests.get(url.encode('utf-8'),
 309                                           headers=self.get_headers,
 310                                           timeout=timeout)
 311             except self.HTTP_ERRORS as e:
 312                 _logger.debug("Request fail: GET %s => %s: %s",
 313                               url, type(e), str(e))
 314                 self.last_result = e
 315             else:
 316                 self.last_result = result
 317                 self.success_flag = retry.check_http_response_success(result)
 318                 content = result.content
 319                 _logger.info("%s response: %s bytes in %s msec (%.3f MiB/sec)",
 320                              self.last_status(), len(content), t.msecs,
 321                              (len(content)/(1024.0*1024))/t.secs)
 322                 if self.success_flag:
 323                     resp_md5 = hashlib.md5(content).hexdigest()
 324                     if resp_md5 == locator.md5sum:
 325                         return content
 326                     _logger.warning("Checksum fail: md5(%s) = %s",
 327                                     url, resp_md5)
 328             return None
 329
 330         def put(self, hash_s, body, timeout=None):
 331             url = self.root + hash_s
 332             _logger.debug("Request: PUT %s", url)
 333             try:
 334                 result = requests.put(url.encode('utf-8'),
 335                                       data=body,
 336                                       headers=self.put_headers,
 337                                       timeout=timeout)
 338             except self.HTTP_ERRORS as e:
 339                 _logger.debug("Request fail: PUT %s => %s: %s",
 340                               url, type(e), str(e))
 341                 self.last_result = e
 342             else:
 343                 self.last_result = result
 344                 self.success_flag = retry.check_http_response_success(result)
 345             return self.success_flag
 346
 347
 348     class KeepWriterThread(threading.Thread):
 349         """
 350         Write a blob of data to the given Keep server. On success, call
 351         save_response() of the given ThreadLimiter to save the returned
 352         locator.
 353         """
 354         def __init__(self, keep_service, **kwargs):
 355             super(KeepClient.KeepWriterThread, self).__init__()
 356             self.service = keep_service
 357             self.args = kwargs
 358             self._success = False
 359
 360         def success(self):
 361             return self._success
 362
 363         def run(self):
 364             with self.args['thread_limiter'] as limiter:
 365                 if not limiter.shall_i_proceed():
 366                     # My turn arrived, but the job has been done without
 367                     # me.
 368                     return
 369                 self.run_with_limiter(limiter)
 370
 371         def run_with_limiter(self, limiter):
 372             if self.service.finished():
 373                 return
 374             _logger.debug("KeepWriterThread %s proceeding %s %s",
 375                           str(threading.current_thread()),
 376                           self.args['data_hash'],
 377                           self.args['service_root'])
 378             self._success = bool(self.service.put(
 379                 self.args['data_hash'],
 380                 self.args['data'],
 381                 timeout=self.args.get('timeout', None)))
 382             status = self.service.last_status()
 383             if self._success:
 384                 result = self.service.last_result
 385                 _logger.debug("KeepWriterThread %s succeeded %s %s",
 386                               str(threading.current_thread()),
 387                               self.args['data_hash'],
 388                               self.args['service_root'])
 389                 # Tick the 'done' counter for the number of replica
 390                 # reported stored by the server, for the case that
 391                 # we're talking to a proxy or other backend that
 392                 # stores to multiple copies for us.
 393                 try:
 394                     replicas_stored = int(result.headers['x-keep-replicas-stored'])
 395                 except (KeyError, ValueError):
 396                     replicas_stored = 1
 397                 limiter.save_response(result.text.strip(), replicas_stored)
 398             elif status is not None:
 399                 _logger.debug("Request fail: PUT %s => %s %s",
 400                               self.args['data_hash'], status,
 401                               self.service.last_result.text)
 402
 403
 404     def __init__(self, api_client=None, proxy=None,
 405                  timeout=DEFAULT_TIMEOUT, proxy_timeout=DEFAULT_PROXY_TIMEOUT,
 406                  api_token=None, local_store=None, block_cache=None,
 407                  num_retries=0):
 408         """Initialize a new KeepClient.
 409
 410         Arguments:
 411         * api_client: The API client to use to find Keep services.  If not
 412           provided, KeepClient will build one from available Arvados
 413           configuration.
 414         * proxy: If specified, this KeepClient will send requests to this
 415           Keep proxy.  Otherwise, KeepClient will fall back to the setting
 416           of the ARVADOS_KEEP_PROXY configuration setting.  If you want to
 417           ensure KeepClient does not use a proxy, pass in an empty string.
 418         * timeout: The timeout (in seconds) for HTTP requests to Keep
 419           non-proxy servers.  A tuple of two floats is interpreted as
 420           (connection_timeout, read_timeout): see
 421           http://docs.python-requests.org/en/latest/user/advanced/#timeouts.
 422           Default: (2, 300).
 423         * proxy_timeout: The timeout (in seconds) for HTTP requests to
 424           Keep proxies. A tuple of two floats is interpreted as
 425           (connection_timeout, read_timeout). Default: (20, 300).
 426         * api_token: If you're not using an API client, but only talking
 427           directly to a Keep proxy, this parameter specifies an API token
 428           to authenticate Keep requests.  It is an error to specify both
 429           api_client and api_token.  If you specify neither, KeepClient
 430           will use one available from the Arvados configuration.
 431         * local_store: If specified, this KeepClient will bypass Keep
 432           services, and save data to the named directory.  If unspecified,
 433           KeepClient will fall back to the setting of the $KEEP_LOCAL_STORE
 434           environment variable.  If you want to ensure KeepClient does not
 435           use local storage, pass in an empty string.  This is primarily
 436           intended to mock a server for testing.
 437         * num_retries: The default number of times to retry failed requests.
 438           This will be used as the default num_retries value when get() and
 439           put() are called.  Default 0.
 440         """
 441         self.lock = threading.Lock()
 442         if proxy is None:
 443             proxy = config.get('ARVADOS_KEEP_PROXY')
 444         if api_token is None:
 445             if api_client is None:
 446                 api_token = config.get('ARVADOS_API_TOKEN')
 447             else:
 448                 api_token = api_client.api_token
 449         elif api_client is not None:
 450             raise ValueError(
 451                 "can't build KeepClient with both API client and token")
 452         if local_store is None:
 453             local_store = os.environ.get('KEEP_LOCAL_STORE')
 454
 455         self.block_cache = block_cache if block_cache else KeepBlockCache()
 456         self.timeout = timeout
 457         self.proxy_timeout = proxy_timeout
 458
 459         if local_store:
 460             self.local_store = local_store
 461             self.get = self.local_store_get
 462             self.put = self.local_store_put
 463         else:
 464             self.num_retries = num_retries
 465             if proxy:
 466                 if not proxy.endswith('/'):
 467                     proxy += '/'
 468                 self.api_token = api_token
 469                 self._keep_services = [{
 470                     'uuid': 'proxy',
 471                     '_service_root': proxy,
 472                     }]
 473                 self.using_proxy = True
 474                 self._static_services_list = True
 475             else:
 476                 # It's important to avoid instantiating an API client
 477                 # unless we actually need one, for testing's sake.
 478                 if api_client is None:
 479                     api_client = arvados.api('v1')
 480                 self.api_client = api_client
 481                 self.api_token = api_client.api_token
 482                 self._keep_services = None
 483                 self.using_proxy = None
 484                 self._static_services_list = False
 485
 486     def current_timeout(self):
 487         """Return the appropriate timeout to use for this client: the proxy
 488         timeout setting if the backend service is currently a proxy,
 489         the regular timeout setting otherwise.
 490         """
 491         # TODO(twp): the timeout should be a property of a
 492         # KeepService, not a KeepClient. See #4488.
 493         return self.proxy_timeout if self.using_proxy else self.timeout
 494
 495     def build_services_list(self, force_rebuild=False):
 496         if (self._static_services_list or
 497               (self._keep_services and not force_rebuild)):
 498             return
 499         with self.lock:
 500             try:
 501                 keep_services = self.api_client.keep_services().accessible()
 502             except Exception:  # API server predates Keep services.
 503                 keep_services = self.api_client.keep_disks().list()
 504
 505             self._keep_services = keep_services.execute().get('items')
 506             if not self._keep_services:
 507                 raise arvados.errors.NoKeepServersError()
 508
 509             self.using_proxy = any(ks.get('service_type') == 'proxy'
 510                                    for ks in self._keep_services)
 511
 512             # Precompute the base URI for each service.
 513             for r in self._keep_services:
 514                 r['_service_root'] = "{}://[{}]:{:d}/".format(
 515                     'https' if r['service_ssl_flag'] else 'http',
 516                     r['service_host'],
 517                     r['service_port'])
 518             _logger.debug(str(self._keep_services))
 519
 520     def _service_weight(self, data_hash, service_uuid):
 521         """Compute the weight of a Keep service endpoint for a data
 522         block with a known hash.
 523
 524         The weight is md5(h + u) where u is the last 15 characters of
 525         the service endpoint's UUID.
 526         """
 527         return hashlib.md5(data_hash + service_uuid[-15:]).hexdigest()
 528
 529     def weighted_service_roots(self, data_hash, force_rebuild=False):
 530         """Return an array of Keep service endpoints, in the order in
 531         which they should be probed when reading or writing data with
 532         the given hash.
 533         """
 534         self.build_services_list(force_rebuild)
 535
 536         # Sort the available services by weight (heaviest first) for
 537         # this data_hash, and return their service_roots (base URIs)
 538         # in that order.
 539         sorted_roots = [
 540             svc['_service_root'] for svc in sorted(
 541                 self._keep_services,
 542                 reverse=True,
 543                 key=lambda svc: self._service_weight(data_hash, svc['uuid']))]
 544         _logger.debug(data_hash + ': ' + str(sorted_roots))
 545         return sorted_roots
 546
 547     def map_new_services(self, roots_map, md5_s, force_rebuild, **headers):
 548         # roots_map is a dictionary, mapping Keep service root strings
 549         # to KeepService objects.  Poll for Keep services, and add any
 550         # new ones to roots_map.  Return the current list of local
 551         # root strings.
 552         headers.setdefault('Authorization', "OAuth2 %s" % (self.api_token,))
 553         local_roots = self.weighted_service_roots(md5_s, force_rebuild)
 554         for root in local_roots:
 555             if root not in roots_map:
 556                 roots_map[root] = self.KeepService(root, **headers)
 557         return local_roots
 558
 559     @staticmethod
 560     def _check_loop_result(result):
 561         # KeepClient RetryLoops should save results as a 2-tuple: the
 562         # actual result of the request, and the number of servers available
 563         # to receive the request this round.
 564         # This method returns True if there's a real result, False if
 565         # there are no more servers available, otherwise None.
 566         if isinstance(result, Exception):
 567             return None
 568         result, tried_server_count = result
 569         if (result is not None) and (result is not False):
 570             return True
 571         elif tried_server_count < 1:
 572             _logger.info("No more Keep services to try; giving up")
 573             return False
 574         else:
 575             return None
 576
 577     @retry.retry_method
 578     def get(self, loc_s, num_retries=None):
 579         """Get data from Keep.
 580
 581         This method fetches one or more blocks of data from Keep.  It
 582         sends a request each Keep service registered with the API
 583         server (or the proxy provided when this client was
 584         instantiated), then each service named in location hints, in
 585         sequence.  As soon as one service provides the data, it's
 586         returned.
 587
 588         Arguments:
 589         * loc_s: A string of one or more comma-separated locators to fetch.
 590           This method returns the concatenation of these blocks.
 591         * num_retries: The number of times to retry GET requests to
 592           *each* Keep server if it returns temporary failures, with
 593           exponential backoff.  Note that, in each loop, the method may try
 594           to fetch data from every available Keep service, along with any
 595           that are named in location hints in the locator.  The default value
 596           is set when the KeepClient is initialized.
 597         """
 598         if ',' in loc_s:
 599             return ''.join(self.get(x) for x in loc_s.split(','))
 600         locator = KeepLocator(loc_s)
 601         expect_hash = locator.md5sum
 602
 603         slot, first = self.block_cache.reserve_cache(expect_hash)
 604         if not first:
 605             v = slot.get()
 606             return v
 607
 608         # See #3147 for a discussion of the loop implementation.  Highlights:
 609         # * Refresh the list of Keep services after each failure, in case
 610         #   it's being updated.
 611         # * Retry until we succeed, we're out of retries, or every available
 612         #   service has returned permanent failure.
 613         hint_roots = ['http://keep.{}.arvadosapi.com/'.format(hint[2:])
 614                       for hint in locator.hints if hint.startswith('K@')]
 615         # Map root URLs their KeepService objects.
 616         roots_map = {root: self.KeepService(root) for root in hint_roots}
 617         blob = None
 618         loop = retry.RetryLoop(num_retries, self._check_loop_result,
 619                                backoff_start=2)
 620         for tries_left in loop:
 621             try:
 622                 local_roots = self.map_new_services(
 623                     roots_map, expect_hash,
 624                     force_rebuild=(tries_left < num_retries))
 625             except Exception as error:
 626                 loop.save_result(error)
 627                 continue
 628
 629             # Query KeepService objects that haven't returned
 630             # permanent failure, in our specified shuffle order.
 631             services_to_try = [roots_map[root]
 632                                for root in (local_roots + hint_roots)
 633                                if roots_map[root].usable()]
 634             for keep_service in services_to_try:
 635                 blob = keep_service.get(locator, timeout=self.current_timeout())
 636                 if blob is not None:
 637                     break
 638             loop.save_result((blob, len(services_to_try)))
 639
 640         # Always cache the result, then return it if we succeeded.
 641         slot.set(blob)
 642         self.block_cache.cap_cache()
 643         if loop.success():
 644             return blob
 645
 646         # No servers fulfilled the request.  Count how many responded
 647         # "not found;" if the ratio is high enough (currently 75%), report
 648         # Not Found; otherwise a generic error.
 649         # Q: Including 403 is necessary for the Keep tests to continue
 650         # passing, but maybe they should expect KeepReadError instead?
 651         not_founds = sum(1 for ks in roots_map.values()
 652                          if ks.last_status() in set([403, 404, 410]))
 653         if roots_map and ((float(not_founds) / len(roots_map)) >= .75):
 654             raise arvados.errors.NotFoundError(loc_s)
 655         else:
 656             raise arvados.errors.KeepReadError(loc_s)
 657
 658     @retry.retry_method
 659     def put(self, data, copies=2, num_retries=None):
 660         """Save data in Keep.
 661
 662         This method will get a list of Keep services from the API server, and
 663         send the data to each one simultaneously in a new thread.  Once the
 664         uploads are finished, if enough copies are saved, this method returns
 665         the most recent HTTP response body.  If requests fail to upload
 666         enough copies, this method raises KeepWriteError.
 667
 668         Arguments:
 669         * data: The string of data to upload.
 670         * copies: The number of copies that the user requires be saved.
 671           Default 2.
 672         * num_retries: The number of times to retry PUT requests to
 673           *each* Keep server if it returns temporary failures, with
 674           exponential backoff.  The default value is set when the
 675           KeepClient is initialized.
 676         """
 677         data_hash = hashlib.md5(data).hexdigest()
 678         if copies < 1:
 679             return data_hash
 680
 681         headers = {}
 682         if self.using_proxy:
 683             # Tell the proxy how many copies we want it to store
 684             headers['X-Keep-Desired-Replication'] = str(copies)
 685         roots_map = {}
 686         thread_limiter = KeepClient.ThreadLimiter(copies)
 687         loop = retry.RetryLoop(num_retries, self._check_loop_result,
 688                                backoff_start=2)
 689         for tries_left in loop:
 690             try:
 691                 local_roots = self.map_new_services(
 692                     roots_map, data_hash,
 693                     force_rebuild=(tries_left < num_retries), **headers)
 694             except Exception as error:
 695                 loop.save_result(error)
 696                 continue
 697
 698             threads = []
 699             for service_root, ks in roots_map.iteritems():
 700                 if ks.finished():
 701                     continue
 702                 t = KeepClient.KeepWriterThread(
 703                     ks,
 704                     data=data,
 705                     data_hash=data_hash,
 706                     service_root=service_root,
 707                     thread_limiter=thread_limiter,
 708                     timeout=self.current_timeout())
 709                 t.start()
 710                 threads.append(t)
 711             for t in threads:
 712                 t.join()
 713             loop.save_result((thread_limiter.done() >= copies, len(threads)))
 714
 715         if loop.success():
 716             return thread_limiter.response()
 717         raise arvados.errors.KeepWriteError(
 718             "Write fail for %s: wanted %d but wrote %d" %
 719             (data_hash, copies, thread_limiter.done()))
 720
 721     # Local storage methods need no-op num_retries arguments to keep
 722     # integration tests happy.  With better isolation they could
 723     # probably be removed again.
 724     def local_store_put(self, data, num_retries=0):
 725         md5 = hashlib.md5(data).hexdigest()
 726         locator = '%s+%d' % (md5, len(data))
 727         with open(os.path.join(self.local_store, md5 + '.tmp'), 'w') as f:
 728             f.write(data)
 729         os.rename(os.path.join(self.local_store, md5 + '.tmp'),
 730                   os.path.join(self.local_store, md5))
 731         return locator
 732
 733     def local_store_get(self, loc_s, num_retries=0):
 734         try:
 735             locator = KeepLocator(loc_s)
 736         except ValueError:
 737             raise arvados.errors.NotFoundError(
 738                 "Invalid data locator: '%s'" % loc_s)
 739         if locator.md5sum == config.EMPTY_BLOCK_LOCATOR.split('+')[0]:
 740             return ''
 741         with open(os.path.join(self.local_store, locator.md5sum), 'r') as f:
 742             return f.read()