X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/8b7ea9235e5761beab0dbc92d9c400574f672a15..c59670ea1fd789bf084b83fb33be3b84312dc07b:/sdk/python/arvados/collection.py diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py index 29c44b416d..a86a1f512c 100644 --- a/sdk/python/arvados/collection.py +++ b/sdk/python/arvados/collection.py @@ -27,6 +27,8 @@ import config import errors import util +_logger = logging.getLogger('arvados.collection') + def normalize_stream(s, stream): stream_tokens = [s] sortedfiles = list(stream.keys()) @@ -41,6 +43,9 @@ def normalize_stream(s, stream): blocks[b[arvados.LOCATOR]] = streamoffset streamoffset += b[arvados.BLOCKSIZE] + if len(stream_tokens) == 1: + stream_tokens.append(config.EMPTY_BLOCK_LOCATOR) + for f in sortedfiles: current_span = None fout = f.replace(' ', '\\040') @@ -87,11 +92,16 @@ def normalize(collection): class CollectionReader(object): - def __init__(self, manifest_locator_or_text): - if re.search(r'^[a-f0-9]{32}(\+\d+)?(\+\S+)*$', manifest_locator_or_text): + def __init__(self, manifest_locator_or_text, api_client=None): + self._api_client = api_client + self._keep_client = None + if re.match(r'[a-f0-9]{32}(\+\d+)?(\+\S+)*$', manifest_locator_or_text): + self._manifest_locator = manifest_locator_or_text + self._manifest_text = None + elif re.match(r'[a-z0-9]{5}-[a-z0-9]{5}-[a-z0-9]{15}$', manifest_locator_or_text): self._manifest_locator = manifest_locator_or_text self._manifest_text = None - elif re.search(r'^\S+( [a-f0-9]{32,}(\+\S+)*)*( \d+:\d+:\S+)+\n', manifest_locator_or_text): + elif re.match(r'(\S+)( [a-f0-9]{32}(\+\d+)(\+\S+)*)+( \d+:\d+:\S+)+\n', manifest_locator_or_text): self._manifest_text = manifest_locator_or_text self._manifest_locator = None else: @@ -106,17 +116,30 @@ class CollectionReader(object): pass def _populate(self): - if self._streams != None: + if self._streams is not None: return if not self._manifest_text: try: - c = arvados.api('v1').collections().get( + # As in KeepClient itself, we must wait until the last possible + # moment to instantiate an API client, in order to avoid + # tripping up clients that don't have access to an API server. + # If we do build one, make sure our Keep client uses it. + # If instantiation fails, we'll fall back to the except clause, + # just like any other Collection lookup failure. + if self._api_client is None: + self._api_client = arvados.api('v1') + self._keep_client = KeepClient(api_client=self._api_client) + if self._keep_client is None: + self._keep_client = KeepClient(api_client=self._api_client) + c = self._api_client.collections().get( uuid=self._manifest_locator).execute() self._manifest_text = c['manifest_text'] except Exception as e: - logging.warning("API lookup failed for collection %s (%s: %s)" % - (self._manifest_locator, type(e), str(e))) - self._manifest_text = Keep.get(self._manifest_locator) + _logger.warning("API lookup failed for collection %s (%s: %s)", + self._manifest_locator, type(e), str(e)) + if self._keep_client is None: + self._keep_client = KeepClient(api_client=self._api_client) + self._manifest_text = self._keep_client.get(self._manifest_locator) self._streams = [] for stream_line in self._manifest_text.split("\n"): if stream_line != '': @@ -143,14 +166,20 @@ class CollectionReader(object): for f in s.all_files(): yield f - def manifest_text(self): + def manifest_text(self, strip=False): self._populate() - return self._manifest_text + if strip: + m = ''.join([StreamReader(stream).manifest_text(strip=True) for stream in self._streams]) + return m + else: + return self._manifest_text class CollectionWriter(object): KEEP_BLOCK_SIZE = 2**26 - def __init__(self): + def __init__(self, api_client=None): + self._api_client = api_client + self._keep_client = None self._data_buffer = [] self._data_buffer_len = 0 self._current_stream_files = [] @@ -171,7 +200,11 @@ class CollectionWriter(object): def __exit__(self): self.finish() - def _do_queued_work(self): + def _prep_keep_client(self): + if self._keep_client is None: + self._keep_client = KeepClient(api_client=self._api_client) + + def do_queued_work(self): # The work queue consists of three pieces: # * _queued_file: The file object we're currently writing to the # Collection. @@ -228,7 +261,11 @@ class CollectionWriter(object): path, stream_name, max_manifest_depth = self._queued_trees[0] make_dirents = (util.listdir_recursive if (max_manifest_depth == 0) else os.listdir) - self._queue_dirents(stream_name, make_dirents(path)) + d = make_dirents(path) + if len(d) > 0: + self._queue_dirents(stream_name, d) + else: + self._queued_trees.popleft() def _queue_file(self, source, filename=None): assert (self._queued_file is None), "tried to queue more than one file" @@ -251,12 +288,12 @@ class CollectionWriter(object): def write_file(self, source, filename=None): self._queue_file(source, filename) - self._do_queued_work() + self.do_queued_work() def write_directory_tree(self, path, stream_name='.', max_manifest_depth=-1): self._queue_tree(path, stream_name, max_manifest_depth) - self._do_queued_work() + self.do_queued_work() def write(self, newdata): if hasattr(newdata, '__iter__'): @@ -272,7 +309,9 @@ class CollectionWriter(object): def flush_data(self): data_buffer = ''.join(self._data_buffer) if data_buffer != '': - self._current_stream_locators += [Keep.put(data_buffer[0:self.KEEP_BLOCK_SIZE])] + self._prep_keep_client() + self._current_stream_locators.append( + self._keep_client.put(data_buffer[0:self.KEEP_BLOCK_SIZE])) self._data_buffer = [data_buffer[self.KEEP_BLOCK_SIZE:]] self._data_buffer_len = len(self._data_buffer[0]) @@ -301,8 +340,8 @@ class CollectionWriter(object): self._current_file_pos, self._current_stream_name)) self._current_stream_files += [[self._current_file_pos, - self._current_stream_length - self._current_file_pos, - self._current_file_name]] + self._current_stream_length - self._current_file_pos, + self._current_file_name]] self._current_file_pos = self._current_stream_length def start_new_stream(self, newstreamname='.'): @@ -331,8 +370,8 @@ class CollectionWriter(object): if len(self._current_stream_locators) == 0: self._current_stream_locators += [config.EMPTY_BLOCK_LOCATOR] self._finished_streams += [[self._current_stream_name, - self._current_stream_locators, - self._current_stream_files]] + self._current_stream_locators, + self._current_stream_files]] self._current_stream_files = [] self._current_stream_length = 0 self._current_stream_locators = [] @@ -341,7 +380,24 @@ class CollectionWriter(object): self._current_file_name = None def finish(self): - return Keep.put(self.manifest_text()) + # Store the manifest in Keep and return its locator. + self._prep_keep_client() + return self._keep_client.put(self.manifest_text()) + + def stripped_manifest(self): + """ + Return the manifest for the current collection with all permission + hints removed from the locators in the manifest. + """ + raw = self.manifest_text() + clean = '' + for line in raw.split("\n"): + fields = line.split() + if len(fields) > 0: + locators = [ re.sub(r'\+A[a-z0-9@_-]+', '', x) + for x in fields[1:-1] ] + clean += fields[0] + ' ' + ' '.join(locators) + ' ' + fields[-1] + "\n" + return clean def manifest_text(self): self.finish_current_stream() @@ -355,10 +411,10 @@ class CollectionWriter(object): manifest += ' ' + ' '.join("%d:%d:%s" % (sfile[0], sfile[1], sfile[2].replace(' ', '\\040')) for sfile in stream[2]) manifest += "\n" - #print 'writer',manifest - #print 'after reader',CollectionReader(manifest).manifest_text() - - return CollectionReader(manifest).manifest_text() + if len(manifest) > 0: + return CollectionReader(manifest).manifest_text() + else: + return "" def data_locators(self): ret = [] @@ -374,12 +430,18 @@ class ResumableCollectionWriter(CollectionWriter): '_data_buffer', '_dependencies', '_finished_streams', '_queued_dirents', '_queued_trees'] - def __init__(self): + def __init__(self, api_client=None): self._dependencies = {} - super(ResumableCollectionWriter, self).__init__() + super(ResumableCollectionWriter, self).__init__(api_client) @classmethod def from_state(cls, state, *init_args, **init_kwargs): + # Try to build a new writer from scratch with the given state. + # If the state is not suitable to resume (because files have changed, + # been deleted, aren't predictable, etc.), raise a + # StaleWriterStateError. Otherwise, return the initialized writer. + # The caller is responsible for calling writer.do_queued_work() + # appropriately after it's returned. writer = cls(*init_args, **init_kwargs) for attr_name in cls.STATE_PROPS: attr_value = state[attr_name] @@ -403,13 +465,8 @@ class ResumableCollectionWriter(CollectionWriter): except IOError as error: raise errors.StaleWriterStateError( "failed to reopen active file {}: {}".format(path, error)) - writer.preresume_hook() - writer._do_queued_work() return writer - def preresume_hook(self): - pass # Subclasses can override this as desired. - def check_dependencies(self): for path, orig_stat in self._dependencies.items(): if not S_ISREG(orig_stat[ST_MODE]):