import threading
from apiclient import errors as apiclient_errors
import errno
+import time
-from fusefile import StringFile, ObjectFile, FuseArvadosFile
-from fresh import FreshBase, convertTime, use_counter
+from fusefile import StringFile, ObjectFile, FuncToJSONFile, FuseArvadosFile
+from fresh import FreshBase, convertTime, use_counter, check_update
import arvados.collection
from arvados.util import portable_data_hash_pattern, uuid_pattern, collection_uuid_pattern, group_uuid_pattern, user_uuid_pattern, link_uuid_pattern
# appear as underscores in the fuse mount.)
_disallowed_filename_characters = re.compile('[\x00/]')
+# '.' and '..' are not reachable if API server is newer than #6277
def sanitize_filename(dirty):
"""Replace disallowed filename characters with harmless "_"."""
if dirty is None:
"""
def __init__(self, parent_inode, inodes):
+ """parent_inode is the integer inode number"""
+
super(Directory, self).__init__()
- """parent_inode is the integer inode number"""
self.inode = None
if not isinstance(parent_inode, int):
raise Exception("parent_inode should be an int")
_logger.warn(e)
@use_counter
+ @check_update
def __getitem__(self, item):
- self.checkupdate()
return self._entries[item]
@use_counter
+ @check_update
def items(self):
- self.checkupdate()
return list(self._entries.items())
@use_counter
+ @check_update
def __contains__(self, k):
- self.checkupdate()
return k in self._entries
@use_counter
+ @check_update
def __len__(self):
- self.checkupdate()
return len(self._entries)
def fresh(self):
# delete any other directory entries that were not in found in 'items'
for i in oldentries:
- _logger.debug("Forgetting about entry '%s' on inode %i", str(i), self.inode)
- llfuse.invalidate_entry(self.inode, str(i))
+ _logger.debug("Forgetting about entry '%s' on inode %i", i, self.inode)
+ self.inodes.invalidate_entry(self.inode, i.encode(self.inodes.encoding))
self.inodes.del_entry(oldentries[i])
changed = True
if changed:
- llfuse.invalidate_inode(self.inode)
+ self.inodes.invalidate_inode(self.inode)
self._mtime = time.time()
self.fresh()
self._entries = oldentries
return False
for n in oldentries:
- llfuse.invalidate_entry(self.inode, str(n))
+ self.inodes.invalidate_entry(self.inode, n.encode(self.inodes.encoding))
self.inodes.del_entry(oldentries[n])
- llfuse.invalidate_inode(self.inode)
+ self.inodes.invalidate_inode(self.inode)
self.invalidate()
return True
else:
def flush(self):
pass
+ def want_event_subscribe(self):
+ raise NotImplementedError()
+
def create(self, name):
raise NotImplementedError()
def rename(self, name_old, name_new, src):
raise NotImplementedError()
+
class CollectionDirectoryBase(Directory):
+ """Represent an Arvados Collection as a directory.
+
+ This class is used for Subcollections, and is also the base class for
+ CollectionDirectory, which implements collection loading/saving on
+ Collection records.
+
+ Most operations act only the underlying Arvados `Collection` object. The
+ `Collection` object signals via a notify callback to
+ `CollectionDirectoryBase.on_event` that an item was added, removed or
+ modified. FUSE inodes and directory entries are created, deleted or
+ invalidated in response to these events.
+
+ """
+
def __init__(self, parent_inode, inodes, collection):
super(CollectionDirectoryBase, self).__init__(parent_inode, inodes)
self.collection = collection
def on_event(self, event, collection, name, item):
if collection == self.collection:
- _logger.debug("%s %s %s %s", event, collection, name, item)
+ name = sanitize_filename(name)
+ _logger.debug("collection notify %s %s %s %s", event, collection, name, item)
with llfuse.lock:
if event == arvados.collection.ADD:
self.new_entry(name, item, self.mtime())
elif event == arvados.collection.DEL:
ent = self._entries[name]
del self._entries[name]
- llfuse.invalidate_entry(self.inode, name)
+ self.inodes.invalidate_entry(self.inode, name.encode(self.inodes.encoding))
self.inodes.del_entry(ent)
elif event == arvados.collection.MOD:
if hasattr(item, "fuse_entry") and item.fuse_entry is not None:
- llfuse.invalidate_inode(item.fuse_entry.inode)
+ self.inodes.invalidate_inode(item.fuse_entry.inode)
elif name in self._entries:
- llfuse.invalidate_inode(self._entries[name].inode)
+ self.inodes.invalidate_inode(self._entries[name].inode)
def populate(self, mtime):
self._mtime = mtime
def writable(self):
return self.collection.writable()
+ @use_counter
def flush(self):
with llfuse.lock_released:
self.collection.root_collection().save()
+ @use_counter
+ @check_update
def create(self, name):
with llfuse.lock_released:
self.collection.open(name, "w").close()
+ @use_counter
+ @check_update
def mkdir(self, name):
with llfuse.lock_released:
self.collection.mkdirs(name)
+ @use_counter
+ @check_update
def unlink(self, name):
with llfuse.lock_released:
self.collection.remove(name)
+ self.flush()
+ @use_counter
+ @check_update
def rmdir(self, name):
with llfuse.lock_released:
self.collection.remove(name)
+ self.flush()
+ @use_counter
+ @check_update
def rename(self, name_old, name_new, src):
if not isinstance(src, CollectionDirectoryBase):
raise llfuse.FUSEError(errno.EPERM)
if name_new in self:
ent = src[name_old]
tgt = self[name_new]
- if isinstance(FuseArvadosFile, ent) and isinstance(FuseArvadosFile, tgt):
+ if isinstance(ent, FuseArvadosFile) and isinstance(tgt, FuseArvadosFile):
pass
- elif isinstance(CollectionDirectoryBase, ent) and isinstance(CollectionDirectoryBase, tgt):
+ elif isinstance(ent, CollectionDirectoryBase) and isinstance(tgt, CollectionDirectoryBase):
if len(tgt) > 0:
raise llfuse.FUSEError(errno.ENOTEMPTY)
- elif isinstance(CollectionDirectoryBase, ent) and isinstance(FuseArvadosFile, tgt):
+ elif isinstance(ent, CollectionDirectoryBase) and isinstance(tgt, FuseArvadosFile):
raise llfuse.FUSEError(errno.ENOTDIR)
- elif isinstance(FuseArvadosFile, ent) and isinstance(CollectionDirectoryBase, tgt):
+ elif isinstance(ent, FuseArvadosFile) and isinstance(tgt, CollectionDirectoryBase):
raise llfuse.FUSEError(errno.EISDIR)
with llfuse.lock_released:
self.flush()
src.flush()
+ def clear(self, force=False):
+ r = super(CollectionDirectoryBase, self).clear(force)
+ self.collection = None
+ return r
+
class CollectionDirectory(CollectionDirectoryBase):
- """Represents the root of a directory tree holding a collection."""
+ """Represents the root of a directory tree representing a collection."""
def __init__(self, parent_inode, inodes, api, num_retries, collection_record=None, explicit_collection=None):
super(CollectionDirectory, self).__init__(parent_inode, inodes, None)
self.num_retries = num_retries
self.collection_record_file = None
self.collection_record = None
+ self._poll = True
+ try:
+ self._poll_time = (api._rootDesc.get('blobSignatureTtl', 60*60*2)/2)
+ except:
+ _logger.debug("Error getting blobSignatureTtl from discovery document: %s", sys.exc_info()[0])
+ self._poll_time = 60*60
+
if isinstance(collection_record, dict):
self.collection_locator = collection_record['uuid']
self._mtime = convertTime(collection_record.get('modified_at'))
def writable(self):
return self.collection.writable() if self.collection is not None else self._writable
+ def want_event_subscribe(self):
+ return (uuid_pattern.match(self.collection_locator) is not None)
+
# Used by arv-web.py to switch the contents of the CollectionDirectory
def change_collection(self, new_locator):
"""Switch the contents of the CollectionDirectory.
def uuid(self):
return self.collection_locator
- def update(self):
+ @use_counter
+ def update(self, to_record_version=None):
try:
if self.collection_record is not None and portable_data_hash_pattern.match(self.collection_locator):
return True
if not self.stale():
return
- _logger.debug("Updating %s", self.collection_locator)
- if self.collection:
- self.collection.update()
+ _logger.debug("Updating %s", to_record_version)
+ if self.collection is not None:
+ if self.collection.known_past_version(to_record_version):
+ _logger.debug("%s already processed %s", self.collection_locator, to_record_version)
+ else:
+ self.collection.update()
else:
if uuid_pattern.match(self.collection_locator):
coll_reader = arvados.collection.Collection(
return True
finally:
self._updating_lock.release()
- except arvados.errors.NotFoundError:
- _logger.exception("arv-mount %s: error", self.collection_locator)
+ except arvados.errors.NotFoundError as e:
+ _logger.error("Error fetching collection '%s': %s", self.collection_locator, e)
except arvados.errors.ArgumentError as detail:
_logger.warning("arv-mount %s: error %s", self.collection_locator, detail)
if self.collection_record is not None and "manifest_text" in self.collection_record:
_logger.exception("arv-mount %s: error", self.collection_locator)
if self.collection_record is not None and "manifest_text" in self.collection_record:
_logger.error("arv-mount manifest_text is: %s", self.collection_record["manifest_text"])
+ self.invalidate()
return False
+ @use_counter
+ @check_update
def __getitem__(self, item):
- self.checkupdate()
if item == '.arvados#collection':
if self.collection_record_file is None:
self.collection_record_file = ObjectFile(self.inode, self.collection_record)
# footprint directly would be more accurate, but also more complicated.
return self._manifest_size * 128
+ def finalize(self):
+ if self.collection is not None:
+ if self.writable():
+ self.collection.save()
+ self.collection.stop_threads()
+
+
+class TmpCollectionDirectory(CollectionDirectoryBase):
+ """A directory backed by an Arvados collection that never gets saved.
+
+ This supports using Keep as scratch space. A userspace program can
+ read the .arvados#collection file to get a current manifest in
+ order to save a snapshot of the scratch data or use it as a crunch
+ job output.
+ """
+
+ class UnsaveableCollection(arvados.collection.Collection):
+ def save(self):
+ pass
+ def save_new(self):
+ pass
+
+ def __init__(self, parent_inode, inodes, api_client, num_retries):
+ collection = self.UnsaveableCollection(
+ api_client=api_client,
+ keep_client=api_client.keep,
+ num_retries=num_retries)
+ super(TmpCollectionDirectory, self).__init__(
+ parent_inode, inodes, collection)
+ self.collection_record_file = None
+ self.populate(self.mtime())
+
+ def on_event(self, *args, **kwargs):
+ super(TmpCollectionDirectory, self).on_event(*args, **kwargs)
+ if self.collection_record_file:
+ with llfuse.lock:
+ self.collection_record_file.invalidate()
+ self.inodes.invalidate_inode(self.collection_record_file.inode)
+ _logger.debug("%s invalidated collection record", self)
+
+ def collection_record(self):
+ with llfuse.lock_released:
+ return {
+ "uuid": None,
+ "manifest_text": self.collection.manifest_text(),
+ "portable_data_hash": self.collection.portable_data_hash(),
+ }
+
+ def __contains__(self, k):
+ return (k == '.arvados#collection' or
+ super(TmpCollectionDirectory, self).__contains__(k))
+
+ @use_counter
+ def __getitem__(self, item):
+ if item == '.arvados#collection':
+ if self.collection_record_file is None:
+ self.collection_record_file = FuncToJSONFile(
+ self.inode, self.collection_record)
+ self.inodes.add_entry(self.collection_record_file)
+ return self.collection_record_file
+ return super(TmpCollectionDirectory, self).__getitem__(item)
+
+ def persisted(self):
+ return False
+
+ def writable(self):
+ return True
+
+ def want_event_subscribe(self):
+ return False
+
+ def finalize(self):
+ self.collection.stop_threads()
+
+ def invalidate(self):
+ if self.collection_record_file:
+ self.collection_record_file.invalidate()
+ super(TmpCollectionDirectory, self).invalidate()
+
+
class MagicDirectory(Directory):
"""A special directory that logically contains the set of all extant keep locators.
README_TEXT = """
This directory provides access to Arvados collections as subdirectories listed
by uuid (in the form 'zzzzz-4zz18-1234567890abcde') or portable data hash (in
-the form '1234567890abcdefghijklmnopqrstuv+123').
+the form '1234567890abcdef0123456789abcdef+123').
Note that this directory will appear empty until you attempt to access a
specific collection subdirectory (such as trying to 'cd' into it), at which
point the collection will actually be looked up on the server and the directory
will appear if it exists.
+
""".lstrip()
- def __init__(self, parent_inode, inodes, api, num_retries):
+ def __init__(self, parent_inode, inodes, api, num_retries, pdh_only=False):
super(MagicDirectory, self).__init__(parent_inode, inodes)
self.api = api
self.num_retries = num_retries
+ self.pdh_only = pdh_only
def __setattr__(self, name, value):
super(MagicDirectory, self).__setattr__(name, value)
# If we're the root directory, add an identical by_id subdirectory.
if self.inode == llfuse.ROOT_INODE:
self._entries['by_id'] = self.inodes.add_entry(MagicDirectory(
- self.inode, self.inodes, self.api, self.num_retries))
+ self.inode, self.inodes, self.api, self.num_retries, self.pdh_only))
def __contains__(self, k):
if k in self._entries:
return True
- if not portable_data_hash_pattern.match(k) and not uuid_pattern.match(k):
+ if not portable_data_hash_pattern.match(k) and (self.pdh_only or not uuid_pattern.match(k)):
return False
try:
self.inode, self.inodes, self.api, self.num_retries, k))
if e.update():
- self._entries[k] = e
+ if k not in self._entries:
+ self._entries[k] = e
+ else:
+ self.inodes.del_entry(e)
return True
else:
+ self.inodes.invalidate_entry(self.inode, k)
+ self.inodes.del_entry(e)
return False
- except Exception as e:
- _logger.debug('arv-mount exception keep %s', e)
+ except Exception as ex:
+ _logger.debug('arv-mount exception keep %s', ex)
+ self.inodes.del_entry(e)
return False
def __getitem__(self, item):
def clear(self, force=False):
pass
+ def want_event_subscribe(self):
+ return not self.pdh_only
+
class RecursiveInvalidateDirectory(Directory):
def invalidate(self):
self._poll = True
self._poll_time = poll_time
+ def want_event_subscribe(self):
+ return True
+
+ @use_counter
def update(self):
with llfuse.lock_released:
tags = self.api.links().list(
self._poll = poll
self._poll_time = poll_time
+ def want_event_subscribe(self):
+ return True
+
+ @use_counter
def update(self):
with llfuse.lock_released:
taggedcollections = self.api.links().list(
self._updating_lock = threading.Lock()
self._current_user = None
+ def want_event_subscribe(self):
+ return True
+
def createDirectory(self, i):
if collection_uuid_pattern.match(i['uuid']):
return CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i)
def uuid(self):
return self.project_uuid
+ @use_counter
def update(self):
if self.project_object_file == None:
self.project_object_file = ObjectFile(self.inode, self.project_object)
finally:
self._updating_lock.release()
+ @use_counter
+ @check_update
def __getitem__(self, item):
- self.checkupdate()
if item == '.arvados#project':
return self.project_object_file
else:
else:
return super(ProjectDirectory, self).__contains__(k)
+ @use_counter
+ @check_update
def writable(self):
with llfuse.lock_released:
if not self._current_user:
def persisted(self):
return True
+ @use_counter
+ @check_update
def mkdir(self, name):
try:
with llfuse.lock_released:
_logger.error(error)
raise llfuse.FUSEError(errno.EEXIST)
+ @use_counter
+ @check_update
def rmdir(self, name):
if name not in self:
raise llfuse.FUSEError(errno.ENOENT)
self.api.collections().delete(uuid=self[name].uuid()).execute(num_retries=self.num_retries)
self.invalidate()
+ @use_counter
+ @check_update
def rename(self, name_old, name_new, src):
if not isinstance(src, ProjectDirectory):
raise llfuse.FUSEError(errno.EPERM)
# Acually move the entry from source directory to this directory.
del src._entries[name_old]
self._entries[name_new] = ent
- llfuse.invalidate_entry(src.inode, name_old)
+ self.inodes.invalidate_entry(src.inode, name_old.encode(self.inodes.encoding))
+
class SharedDirectory(Directory):
"""A special directory that represents users or groups who have shared projects with me."""
self._poll = True
self._poll_time = poll_time
+ @use_counter
def update(self):
with llfuse.lock_released:
all_projects = arvados.util.list_all(
lambda i: ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, i[1], poll=self._poll, poll_time=self._poll_time))
except Exception:
_logger.exception()
+
+ def want_event_subscribe(self):
+ return True
projects / arvados.git / blobdiff