X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/3809452aef74876da9d2644fe6c824a22527d6ac..b8ae9ce66c24055f3bf66b0956e15c44c9dd51f0:/services/fuse/arvados_fuse/fusedir.py diff --git a/services/fuse/arvados_fuse/fusedir.py b/services/fuse/arvados_fuse/fusedir.py index 17709df1d7..3f2bcd5ec2 100644 --- a/services/fuse/arvados_fuse/fusedir.py +++ b/services/fuse/arvados_fuse/fusedir.py @@ -8,9 +8,10 @@ import functools 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 @@ -23,6 +24,7 @@ _logger = logging.getLogger('arvados.arvados_fuse') # 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: @@ -45,9 +47,10 @@ class Directory(FreshBase): """ 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") @@ -78,23 +81,23 @@ class Directory(FreshBase): _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): @@ -142,13 +145,13 @@ class Directory(FreshBase): # 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() @@ -164,9 +167,9 @@ class Directory(FreshBase): 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: @@ -181,6 +184,9 @@ class Directory(FreshBase): def flush(self): pass + def want_event_subscribe(self): + raise NotImplementedError() + def create(self, name): raise NotImplementedError() @@ -196,7 +202,22 @@ class Directory(FreshBase): 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 @@ -219,20 +240,21 @@ class CollectionDirectoryBase(Directory): 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 @@ -243,28 +265,39 @@ class CollectionDirectoryBase(Directory): 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) @@ -287,9 +320,14 @@ class CollectionDirectoryBase(Directory): 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) @@ -297,6 +335,13 @@ class CollectionDirectory(CollectionDirectoryBase): 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')) @@ -314,6 +359,9 @@ class CollectionDirectory(CollectionDirectoryBase): 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. @@ -343,7 +391,8 @@ class CollectionDirectory(CollectionDirectoryBase): 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 @@ -358,9 +407,12 @@ class CollectionDirectory(CollectionDirectoryBase): 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( @@ -391,8 +443,8 @@ class CollectionDirectory(CollectionDirectoryBase): 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: @@ -401,10 +453,12 @@ class CollectionDirectory(CollectionDirectoryBase): _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) @@ -433,6 +487,86 @@ class CollectionDirectory(CollectionDirectoryBase): # 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. @@ -447,18 +581,20 @@ class MagicDirectory(Directory): 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) @@ -470,13 +606,13 @@ will appear if it exists. # 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: @@ -484,12 +620,18 @@ will appear if it exists. 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): @@ -501,6 +643,9 @@ will appear if it exists. def clear(self, force=False): pass + def want_event_subscribe(self): + return not self.pdh_only + class RecursiveInvalidateDirectory(Directory): def invalidate(self): @@ -522,6 +667,10 @@ class TagsDirectory(RecursiveInvalidateDirectory): 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( @@ -549,6 +698,10 @@ class TagDirectory(Directory): 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( @@ -579,6 +732,9 @@ class ProjectDirectory(Directory): 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) @@ -597,6 +753,7 @@ class ProjectDirectory(Directory): 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) @@ -650,8 +807,9 @@ class ProjectDirectory(Directory): 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: @@ -663,6 +821,8 @@ class ProjectDirectory(Directory): else: return super(ProjectDirectory, self).__contains__(k) + @use_counter + @check_update def writable(self): with llfuse.lock_released: if not self._current_user: @@ -672,6 +832,8 @@ class ProjectDirectory(Directory): def persisted(self): return True + @use_counter + @check_update def mkdir(self, name): try: with llfuse.lock_released: @@ -683,6 +845,8 @@ class ProjectDirectory(Directory): _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) @@ -694,6 +858,8 @@ class ProjectDirectory(Directory): 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) @@ -717,7 +883,8 @@ class ProjectDirectory(Directory): # 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.""" @@ -731,6 +898,7 @@ class SharedDirectory(Directory): self._poll = True self._poll_time = poll_time + @use_counter def update(self): with llfuse.lock_released: all_projects = arvados.util.list_all( @@ -787,3 +955,6 @@ class SharedDirectory(Directory): 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