-import gflags
-import httplib
-import httplib2
import logging
import os
-import pprint
-import sys
-import types
-import subprocess
-import json
-import UserDict
import re
-import hashlib
-import string
-import bz2
-import zlib
-import fcntl
-import time
-import threading
from collections import deque
from stat import *
fout = f.replace(' ', '\\040')
for segment in stream[f]:
segmentoffset = blocks[segment[arvados.LOCATOR]] + segment[arvados.OFFSET]
- if current_span == None:
+ if current_span is None:
current_span = [segmentoffset, segmentoffset + segment[arvados.SEGMENTSIZE]]
else:
if segmentoffset == current_span[1]:
stream_tokens.append("{0}:{1}:{2}".format(current_span[0], current_span[1] - current_span[0], fout))
current_span = [segmentoffset, segmentoffset + segment[arvados.SEGMENTSIZE]]
- if current_span != None:
+ if current_span is not None:
stream_tokens.append("{0}:{1}:{2}".format(current_span[0], current_span[1] - current_span[0], fout))
- if len(stream[f]) == 0:
+ if not stream[f]:
stream_tokens.append("0:0:{0}".format(fout))
return stream_tokens
-def normalize(collection):
- streams = {}
- for s in collection.all_streams():
- for f in s.all_files():
- filestream = s.name() + "/" + f.name()
- r = filestream.rindex("/")
- streamname = filestream[:r]
- filename = filestream[r+1:]
- if streamname not in streams:
- streams[streamname] = {}
- if filename not in streams[streamname]:
- streams[streamname][filename] = []
- for r in f.segments:
- streams[streamname][filename].extend(s.locators_and_ranges(r[0], r[1]))
-
- normalized_streams = []
- sortedstreams = list(streams.keys())
- sortedstreams.sort()
- for s in sortedstreams:
- normalized_streams.append(normalize_stream(s, streams[s]))
- return normalized_streams
-
class CollectionBase(object):
def __enter__(self):
- pass
+ return self
- def __exit__(self):
+ def __exit__(self, exc_type, exc_value, traceback):
pass
def _my_keep(self):
num_retries=self.num_retries)
return self._keep_client
+ def stripped_manifest(self):
+ """
+ Return the manifest for the current collection with all
+ non-portable hints (i.e., permission signatures and other
+ hints other than size hints) removed from the locators.
+ """
+ raw = self.manifest_text()
+ clean = []
+ for line in raw.split("\n"):
+ fields = line.split()
+ if fields:
+ clean_fields = fields[:1] + [
+ (re.sub(r'\+[^\d][^\+]*', '', x)
+ if re.match(util.keep_locator_pattern, x)
+ else x)
+ for x in fields[1:]]
+ clean += [' '.join(clean_fields), "\n"]
+ return ''.join(clean)
+
class CollectionReader(CollectionBase):
def __init__(self, manifest_locator_or_text, api_client=None,
"Argument to CollectionReader must be a manifest or a collection UUID")
self._streams = None
+ def _populate_from_api_server(self):
+ # 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. Return an exception, or None if successful.
+ try:
+ if self._api_client is None:
+ self._api_client = arvados.api('v1')
+ self._keep_client = None # Make a new one with the new api.
+ c = self._api_client.collections().get(
+ uuid=self._manifest_locator).execute(
+ num_retries=self.num_retries)
+ self._manifest_text = c['manifest_text']
+ return None
+ except Exception as e:
+ return e
+
+ def _populate_from_keep(self):
+ # Retrieve a manifest directly from Keep. This has a chance of
+ # working if [a] the locator includes a permission signature
+ # or [b] the Keep services are operating in world-readable
+ # mode. Return an exception, or None if successful.
+ try:
+ self._manifest_text = self._my_keep().get(
+ self._manifest_locator, num_retries=self.num_retries)
+ except Exception as e:
+ return e
+
def _populate(self):
if self._streams is not None:
return
error_via_api = None
error_via_keep = None
- should_try_keep = (not self._manifest_text and
+ should_try_keep = ((self._manifest_text is None) and
util.keep_locator_pattern.match(
self._manifest_locator))
- if (not self._manifest_text and
+ if ((self._manifest_text is None) and
util.signed_locator_pattern.match(self._manifest_locator)):
- try:
- self._populate_from_keep()
- except e:
- error_via_keep = e
- if not self._manifest_text:
- try:
- self._populate_from_api_server()
- except Exception as e:
- if not should_try_keep:
- raise
- error_via_api = e
- if (not self._manifest_text and
+ error_via_keep = self._populate_from_keep()
+ if self._manifest_text is None:
+ error_via_api = self._populate_from_api_server()
+ if error_via_api is not None and not should_try_keep:
+ raise error_via_api
+ if ((self._manifest_text is None) and
not error_via_keep and
should_try_keep):
# Looks like a keep locator, and we didn't already try keep above
- try:
- self._populate_from_keep()
- except Exception as e:
- error_via_keep = e
- if not self._manifest_text:
+ error_via_keep = self._populate_from_keep()
+ if self._manifest_text is None:
# Nothing worked!
raise arvados.errors.NotFoundError(
("Failed to retrieve collection '{}' " +
self._streams = [sline.split()
for sline in self._manifest_text.split("\n")
if sline]
- self._streams = normalize(self)
-
- # now regenerate the manifest text based on the normalized stream
- #print "normalizing", self._manifest_text
- self._manifest_text = ''.join([StreamReader(stream, keep=self._my_keep()).manifest_text() for stream in self._streams])
- #print "result", self._manifest_text
+ def normalize(self):
+ self._populate()
+ # Rearrange streams
+ streams = {}
+ for s in self.all_streams():
+ for f in s.all_files():
+ filestream = s.name() + "/" + f.name()
+ r = filestream.rindex("/")
+ streamname = filestream[:r]
+ filename = filestream[r+1:]
+ if streamname not in streams:
+ streams[streamname] = {}
+ if filename not in streams[streamname]:
+ streams[streamname][filename] = []
+ for r in f.segments:
+ streams[streamname][filename].extend(s.locators_and_ranges(r[0], r[1]))
+
+ self._streams = [normalize_stream(s, streams[s])
+ for s in sorted(streams)]
+
+ # Regenerate the manifest text based on the normalized streams
+ self._manifest_text = ''.join(
+ [StreamReader(stream, keep=self._my_keep()).manifest_text()
+ for stream in self._streams])
def all_streams(self):
self._populate()
for f in s.all_files():
yield f
- def manifest_text(self, strip=False):
- self._populate()
- if strip:
- m = ''.join([StreamReader(stream, keep=self._my_keep()).manifest_text(strip=True) for stream in self._streams])
- return m
+ def manifest_text(self, strip=False, normalize=False):
+ if normalize:
+ cr = CollectionReader(self.manifest_text())
+ cr.normalize()
+ return cr.manifest_text(strip=strip, normalize=False)
+ elif strip:
+ return self.stripped_manifest()
else:
+ self._populate()
return self._manifest_text
self._queued_dirents = deque()
self._queued_trees = deque()
- def __exit__(self):
- self.finish()
+ def __exit__(self, exc_type, exc_value, traceback):
+ if exc_type is None:
+ self.finish()
def do_queued_work(self):
# The work queue consists of three pieces:
def _work_trees(self):
path, stream_name, max_manifest_depth = self._queued_trees[0]
- make_dirents = (util.listdir_recursive if (max_manifest_depth == 0)
- else os.listdir)
- d = make_dirents(path)
- if len(d) > 0:
+ d = util.listdir_recursive(
+ path, max_depth = (None if max_manifest_depth == 0 else 0))
+ if d:
self._queue_dirents(stream_name, d)
else:
self._queued_trees.popleft()
return self._current_file_name
def finish_current_file(self):
- if self._current_file_name == None:
+ if self._current_file_name is None:
if self._current_file_pos == self._current_stream_length:
return
raise errors.AssertionError(
# Store the manifest in Keep and return its locator.
return self._my_keep().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 portable_data_hash(self):
+ stripped = self.stripped_manifest()
+ return hashlib.md5(stripped).hexdigest() + '+' + str(len(stripped))
def manifest_text(self):
self.finish_current_stream()
manifest += ' ' + ' '.join("%d:%d:%s" % (sfile[0], sfile[1], sfile[2].replace(' ', '\\040')) for sfile in stream[2])
manifest += "\n"
- if manifest:
- return CollectionReader(manifest, self._api_client).manifest_text()
- else:
- return ""
+ return manifest
def data_locators(self):
ret = []