1 # Copyright (C) The Arvados Authors. All rights reserved.
3 # SPDX-License-Identifier: AGPL-3.0
15 from apiclient import errors as apiclient_errors
17 from .fusefile import StringFile, ObjectFile, FuncToJSONFile, FuseArvadosFile
18 from .fresh import FreshBase, convertTime, use_counter, check_update
20 import arvados.collection
21 from arvados.util import portable_data_hash_pattern, uuid_pattern, collection_uuid_pattern, group_uuid_pattern, user_uuid_pattern, link_uuid_pattern
23 _logger = logging.getLogger('arvados.arvados_fuse')
26 # Match any character which FUSE or Linux cannot accommodate as part
27 # of a filename. (If present in a collection filename, they will
28 # appear as underscores in the fuse mount.)
29 _disallowed_filename_characters = re.compile(r'[\x00/]')
32 class Directory(FreshBase):
33 """Generic directory object, backed by a dict.
35 Consists of a set of entries with the key representing the filename
36 and the value referencing a File or Directory object.
39 def __init__(self, parent_inode, inodes, apiconfig, enable_write):
40 """parent_inode is the integer inode number"""
42 super(Directory, self).__init__()
45 if not isinstance(parent_inode, int):
46 raise Exception("parent_inode should be an int")
47 self.parent_inode = parent_inode
49 self.apiconfig = apiconfig
51 self._mtime = time.time()
52 self._enable_write = enable_write
54 def forward_slash_subst(self):
55 if not hasattr(self, '_fsns'):
57 config = self.apiconfig()
59 self._fsns = config["Collections"]["ForwardSlashNameSubstitution"]
61 # old API server with no FSNS config
64 if self._fsns == '' or self._fsns == '/':
68 def unsanitize_filename(self, incoming):
69 """Replace ForwardSlashNameSubstitution value with /"""
70 fsns = self.forward_slash_subst()
71 if isinstance(fsns, str):
72 return incoming.replace(fsns, '/')
76 def sanitize_filename(self, dirty):
77 """Replace disallowed filename characters according to
78 ForwardSlashNameSubstitution in self.api_config."""
79 # '.' and '..' are not reachable if API server is newer than #6277
89 fsns = self.forward_slash_subst()
90 if isinstance(fsns, str):
91 dirty = dirty.replace('/', fsns)
92 return _disallowed_filename_characters.sub('_', dirty)
95 # Overridden by subclasses to implement logic to update the
96 # entries dict when the directory is stale
101 # Only used when computing the size of the disk footprint of the directory
109 def checkupdate(self):
113 except apiclient.errors.HttpError as e:
118 def __getitem__(self, item):
119 return self._entries[item]
124 return list(self._entries.items())
128 def __contains__(self, k):
129 return k in self._entries
134 return len(self._entries)
137 self.inodes.touch(self)
138 super(Directory, self).fresh()
140 def merge(self, items, fn, same, new_entry):
141 """Helper method for updating the contents of the directory.
143 Takes a list describing the new contents of the directory, reuse
144 entries that are the same in both the old and new lists, create new
145 entries, and delete old entries missing from the new list.
147 :items: iterable with new directory contents
149 :fn: function to take an entry in 'items' and return the desired file or
150 directory name, or None if this entry should be skipped
152 :same: function to compare an existing entry (a File or Directory
153 object) with an entry in the items list to determine whether to keep
156 :new_entry: function to create a new directory entry (File or Directory
157 object) from an entry in the items list.
161 oldentries = self._entries
165 name = self.sanitize_filename(fn(i))
167 if name in oldentries and same(oldentries[name], i):
168 # move existing directory entry over
169 self._entries[name] = oldentries[name]
172 _logger.debug("Adding entry '%s' to inode %i", name, self.inode)
173 # create new directory entry
176 self._entries[name] = self.inodes.add_entry(ent)
179 # delete any other directory entries that were not in found in 'items'
181 _logger.debug("Forgetting about entry '%s' on inode %i", i, self.inode)
182 self.inodes.invalidate_entry(self, i)
183 self.inodes.del_entry(oldentries[i])
187 self.inodes.invalidate_inode(self)
188 self._mtime = time.time()
193 if super(Directory, self).in_use():
195 for v in self._entries.values():
200 def has_ref(self, only_children):
201 if super(Directory, self).has_ref(only_children):
203 for v in self._entries.values():
209 """Delete all entries"""
210 oldentries = self._entries
213 oldentries[n].clear()
214 self.inodes.del_entry(oldentries[n])
217 def kernel_invalidate(self):
218 # Invalidating the dentry on the parent implies invalidating all paths
220 parent = self.inodes[self.parent_inode]
222 # Find self on the parent in order to invalidate this path.
223 # Calling the public items() method might trigger a refresh,
224 # which we definitely don't want, so read the internal dict directly.
225 for k,v in parent._entries.items():
227 self.inodes.invalidate_entry(parent, k)
239 def want_event_subscribe(self):
240 raise NotImplementedError()
242 def create(self, name):
243 raise NotImplementedError()
245 def mkdir(self, name):
246 raise NotImplementedError()
248 def unlink(self, name):
249 raise NotImplementedError()
251 def rmdir(self, name):
252 raise NotImplementedError()
254 def rename(self, name_old, name_new, src):
255 raise NotImplementedError()
258 class CollectionDirectoryBase(Directory):
259 """Represent an Arvados Collection as a directory.
261 This class is used for Subcollections, and is also the base class for
262 CollectionDirectory, which implements collection loading/saving on
265 Most operations act only the underlying Arvados `Collection` object. The
266 `Collection` object signals via a notify callback to
267 `CollectionDirectoryBase.on_event` that an item was added, removed or
268 modified. FUSE inodes and directory entries are created, deleted or
269 invalidated in response to these events.
273 def __init__(self, parent_inode, inodes, apiconfig, enable_write, collection, collection_root):
274 super(CollectionDirectoryBase, self).__init__(parent_inode, inodes, apiconfig, enable_write)
275 self.apiconfig = apiconfig
276 self.collection = collection
277 self.collection_root = collection_root
278 self.collection_record_file = None
280 def new_entry(self, name, item, mtime):
281 name = self.sanitize_filename(name)
282 if hasattr(item, "fuse_entry") and item.fuse_entry is not None:
283 if item.fuse_entry.dead is not True:
284 raise Exception("Can only reparent dead inode entry")
285 if item.fuse_entry.inode is None:
286 raise Exception("Reparented entry must still have valid inode")
287 item.fuse_entry.dead = False
288 self._entries[name] = item.fuse_entry
289 elif isinstance(item, arvados.collection.RichCollectionBase):
290 self._entries[name] = self.inodes.add_entry(CollectionDirectoryBase(self.inode, self.inodes, self.apiconfig, self._enable_write, item, self.collection_root))
291 self._entries[name].populate(mtime)
293 self._entries[name] = self.inodes.add_entry(FuseArvadosFile(self.inode, item, mtime, self._enable_write))
294 item.fuse_entry = self._entries[name]
296 def on_event(self, event, collection, name, item):
297 # These are events from the Collection object (ADD/DEL/MOD)
298 # emitted by operations on the Collection object (like
299 # "mkdirs" or "remove"), and by "update", which we need to
300 # synchronize with our FUSE objects that are assigned inodes.
301 if collection == self.collection:
302 name = self.sanitize_filename(name)
305 # It's possible for another thread to have llfuse.lock and
306 # be waiting on collection.lock. Meanwhile, we released
307 # llfuse.lock earlier in the stack, but are still holding
308 # on to the collection lock, and now we need to re-acquire
309 # llfuse.lock. If we don't release the collection lock,
310 # we'll deadlock where we're holding the collection lock
311 # waiting for llfuse.lock and the other thread is holding
312 # llfuse.lock and waiting for the collection lock.
314 # The correct locking order here is to take llfuse.lock
315 # first, then the collection lock.
317 # Since collection.lock is an RLock, it might be locked
318 # multiple times, so we need to release it multiple times,
319 # keep a count, then re-lock it the correct number of
325 self.collection.lock.release()
332 with self.collection.lock:
333 if event == arvados.collection.ADD:
334 self.new_entry(name, item, self.mtime())
335 elif event == arvados.collection.DEL:
336 ent = self._entries[name]
337 del self._entries[name]
338 self.inodes.invalidate_entry(self, name)
339 self.inodes.del_entry(ent)
340 elif event == arvados.collection.MOD:
341 if hasattr(item, "fuse_entry") and item.fuse_entry is not None:
342 self.inodes.invalidate_inode(item.fuse_entry)
343 elif name in self._entries:
344 self.inodes.invalidate_inode(self._entries[name])
346 if self.collection_record_file is not None:
347 self.collection_record_file.invalidate()
348 self.inodes.invalidate_inode(self.collection_record_file)
351 self.collection.lock.acquire()
354 def populate(self, mtime):
356 with self.collection.lock:
357 self.collection.subscribe(self.on_event)
358 for entry, item in self.collection.items():
359 self.new_entry(entry, item, self.mtime())
362 return self._enable_write and self.collection.writable()
366 self.collection_root.flush()
370 def create(self, name):
371 if not self.writable():
372 raise llfuse.FUSEError(errno.EROFS)
373 with llfuse.lock_released:
374 self.collection.open(name, "w").close()
378 def mkdir(self, name):
379 if not self.writable():
380 raise llfuse.FUSEError(errno.EROFS)
381 with llfuse.lock_released:
382 self.collection.mkdirs(name)
386 def unlink(self, name):
387 if not self.writable():
388 raise llfuse.FUSEError(errno.EROFS)
389 with llfuse.lock_released:
390 self.collection.remove(name)
395 def rmdir(self, name):
396 if not self.writable():
397 raise llfuse.FUSEError(errno.EROFS)
398 with llfuse.lock_released:
399 self.collection.remove(name)
404 def rename(self, name_old, name_new, src):
405 if not self.writable():
406 raise llfuse.FUSEError(errno.EROFS)
408 if not isinstance(src, CollectionDirectoryBase):
409 raise llfuse.FUSEError(errno.EPERM)
414 if isinstance(ent, FuseArvadosFile) and isinstance(tgt, FuseArvadosFile):
416 elif isinstance(ent, CollectionDirectoryBase) and isinstance(tgt, CollectionDirectoryBase):
418 raise llfuse.FUSEError(errno.ENOTEMPTY)
419 elif isinstance(ent, CollectionDirectoryBase) and isinstance(tgt, FuseArvadosFile):
420 raise llfuse.FUSEError(errno.ENOTDIR)
421 elif isinstance(ent, FuseArvadosFile) and isinstance(tgt, CollectionDirectoryBase):
422 raise llfuse.FUSEError(errno.EISDIR)
424 with llfuse.lock_released:
425 self.collection.rename(name_old, name_new, source_collection=src.collection, overwrite=True)
430 super(CollectionDirectoryBase, self).clear()
431 self.collection = None
434 class CollectionDirectory(CollectionDirectoryBase):
435 """Represents the root of a directory tree representing a collection."""
437 def __init__(self, parent_inode, inodes, api, num_retries, enable_write, collection_record=None, explicit_collection=None):
438 super(CollectionDirectory, self).__init__(parent_inode, inodes, api.config, enable_write, None, self)
440 self.num_retries = num_retries
443 self._poll_time = (api._rootDesc.get('blobSignatureTtl', 60*60*2) // 2)
445 _logger.debug("Error getting blobSignatureTtl from discovery document: %s", sys.exc_info()[0])
446 self._poll_time = 60*60
448 if isinstance(collection_record, dict):
449 self.collection_locator = collection_record['uuid']
450 self._mtime = convertTime(collection_record.get('modified_at'))
452 self.collection_locator = collection_record
454 self._manifest_size = 0
455 if self.collection_locator:
456 self._writable = (uuid_pattern.match(self.collection_locator) is not None) and enable_write
457 self._updating_lock = threading.Lock()
460 return i['uuid'] == self.collection_locator or i['portable_data_hash'] == self.collection_locator
463 return self._enable_write and (self.collection.writable() if self.collection is not None else self._writable)
467 if not self.writable():
469 with llfuse.lock_released:
470 with self._updating_lock:
471 if self.collection.committed():
472 self.collection.update()
474 self.collection.save()
475 self.new_collection_record(self.collection.api_response())
477 def want_event_subscribe(self):
478 return (uuid_pattern.match(self.collection_locator) is not None)
480 def new_collection(self, new_collection_record, coll_reader):
483 self.collection = coll_reader
484 self.new_collection_record(new_collection_record)
485 self.populate(self.mtime())
487 def new_collection_record(self, new_collection_record):
488 if not new_collection_record:
489 raise Exception("invalid new_collection_record")
490 self._mtime = convertTime(new_collection_record.get('modified_at'))
491 self._manifest_size = len(new_collection_record["manifest_text"])
492 self.collection_locator = new_collection_record["uuid"]
493 if self.collection_record_file is not None:
494 self.collection_record_file.invalidate()
495 self.inodes.invalidate_inode(self.collection_record_file)
496 _logger.debug("%s invalidated collection record file", self)
500 return self.collection_locator
505 if self.collection is not None and portable_data_hash_pattern.match(self.collection_locator):
506 # It's immutable, nothing to update
509 if self.collection_locator is None:
510 # No collection locator to retrieve from
514 new_collection_record = None
516 with llfuse.lock_released:
517 self._updating_lock.acquire()
521 _logger.debug("Updating collection %s inode %s", self.collection_locator, self.inode)
523 if self.collection is not None:
524 # Already have a collection object
525 self.collection.update()
526 new_collection_record = self.collection.api_response()
528 # Create a new collection object
529 if uuid_pattern.match(self.collection_locator):
530 coll_reader = arvados.collection.Collection(
531 self.collection_locator, self.api, self.api.keep,
532 num_retries=self.num_retries)
534 coll_reader = arvados.collection.CollectionReader(
535 self.collection_locator, self.api, self.api.keep,
536 num_retries=self.num_retries)
537 new_collection_record = coll_reader.api_response() or {}
538 # If the Collection only exists in Keep, there will be no API
539 # response. Fill in the fields we need.
540 if 'uuid' not in new_collection_record:
541 new_collection_record['uuid'] = self.collection_locator
542 if "portable_data_hash" not in new_collection_record:
543 new_collection_record["portable_data_hash"] = new_collection_record["uuid"]
544 if 'manifest_text' not in new_collection_record:
545 new_collection_record['manifest_text'] = coll_reader.manifest_text()
546 if 'storage_classes_desired' not in new_collection_record:
547 new_collection_record['storage_classes_desired'] = coll_reader.storage_classes_desired()
549 # end with llfuse.lock_released, re-acquire lock
551 if new_collection_record is not None:
552 if coll_reader is not None:
553 self.new_collection(new_collection_record, coll_reader)
555 self.new_collection_record(new_collection_record)
559 self._updating_lock.release()
560 except arvados.errors.NotFoundError as e:
561 _logger.error("Error fetching collection '%s': %s", self.collection_locator, e)
562 except arvados.errors.ArgumentError as detail:
563 _logger.warning("arv-mount %s: error %s", self.collection_locator, detail)
564 if new_collection_record is not None and "manifest_text" in new_collection_record:
565 _logger.warning("arv-mount manifest_text is: %s", new_collection_record["manifest_text"])
567 _logger.exception("arv-mount %s: error", self.collection_locator)
568 if new_collection_record is not None and "manifest_text" in new_collection_record:
569 _logger.error("arv-mount manifest_text is: %s", new_collection_record["manifest_text"])
574 def collection_record(self):
576 return self.collection.api_response()
580 def __getitem__(self, item):
581 if item == '.arvados#collection':
582 if self.collection_record_file is None:
583 self.collection_record_file = FuncToJSONFile(
584 self.inode, self.collection_record)
585 self.inodes.add_entry(self.collection_record_file)
586 self.invalidate() # use lookup as a signal to force update
587 return self.collection_record_file
589 return super(CollectionDirectory, self).__getitem__(item)
591 def __contains__(self, k):
592 if k == '.arvados#collection':
595 return super(CollectionDirectory, self).__contains__(k)
597 def invalidate(self):
598 if self.collection_record_file is not None:
599 self.collection_record_file.invalidate()
600 self.inodes.invalidate_inode(self.collection_record_file)
601 super(CollectionDirectory, self).invalidate()
604 return (self.collection_locator is not None)
607 # This is an empirically-derived heuristic to estimate the memory used
608 # to store this collection's metadata. Calculating the memory
609 # footprint directly would be more accurate, but also more complicated.
610 return self._manifest_size * 128
613 if self.collection is not None:
615 self.collection.save()
616 self.collection.stop_threads()
619 if self.collection is not None:
620 self.collection.stop_threads()
621 super(CollectionDirectory, self).clear()
622 self._manifest_size = 0
625 class TmpCollectionDirectory(CollectionDirectoryBase):
626 """A directory backed by an Arvados collection that never gets saved.
628 This supports using Keep as scratch space. A userspace program can
629 read the .arvados#collection file to get a current manifest in
630 order to save a snapshot of the scratch data or use it as a crunch
634 class UnsaveableCollection(arvados.collection.Collection):
640 def __init__(self, parent_inode, inodes, api_client, num_retries, enable_write, storage_classes=None):
641 collection = self.UnsaveableCollection(
642 api_client=api_client,
643 keep_client=api_client.keep,
644 num_retries=num_retries,
645 storage_classes_desired=storage_classes)
646 # This is always enable_write=True because it never tries to
647 # save to the backend
648 super(TmpCollectionDirectory, self).__init__(
649 parent_inode, inodes, api_client.config, True, collection, self)
650 self.populate(self.mtime())
652 def on_event(self, *args, **kwargs):
653 super(TmpCollectionDirectory, self).on_event(*args, **kwargs)
654 if self.collection_record_file is None:
657 # See discussion in CollectionDirectoryBase.on_event
661 self.collection.lock.release()
668 with self.collection.lock:
669 self.collection_record_file.invalidate()
670 self.inodes.invalidate_inode(self.collection_record_file)
671 _logger.debug("%s invalidated collection record", self)
674 self.collection.lock.acquire()
677 def collection_record(self):
678 with llfuse.lock_released:
681 "manifest_text": self.collection.manifest_text(),
682 "portable_data_hash": self.collection.portable_data_hash(),
683 "storage_classes_desired": self.collection.storage_classes_desired(),
686 def __contains__(self, k):
687 return (k == '.arvados#collection' or
688 super(TmpCollectionDirectory, self).__contains__(k))
691 def __getitem__(self, item):
692 if item == '.arvados#collection':
693 if self.collection_record_file is None:
694 self.collection_record_file = FuncToJSONFile(
695 self.inode, self.collection_record)
696 self.inodes.add_entry(self.collection_record_file)
697 return self.collection_record_file
698 return super(TmpCollectionDirectory, self).__getitem__(item)
709 def want_event_subscribe(self):
713 self.collection.stop_threads()
715 def invalidate(self):
716 if self.collection_record_file:
717 self.collection_record_file.invalidate()
718 super(TmpCollectionDirectory, self).invalidate()
721 class MagicDirectory(Directory):
722 """A special directory that logically contains the set of all extant keep locators.
724 When a file is referenced by lookup(), it is tested to see if it is a valid
725 keep locator to a manifest, and if so, loads the manifest contents as a
726 subdirectory of this directory with the locator as the directory name.
727 Since querying a list of all extant keep locators is impractical, only
728 collections that have already been accessed are visible to readdir().
733 This directory provides access to Arvados collections as subdirectories listed
734 by uuid (in the form 'zzzzz-4zz18-1234567890abcde') or portable data hash (in
735 the form '1234567890abcdef0123456789abcdef+123'), and Arvados projects by uuid
736 (in the form 'zzzzz-j7d0g-1234567890abcde').
738 Note that this directory will appear empty until you attempt to access a
739 specific collection or project subdirectory (such as trying to 'cd' into it),
740 at which point the collection or project will actually be looked up on the server
741 and the directory will appear if it exists.
745 def __init__(self, parent_inode, inodes, api, num_retries, enable_write, pdh_only=False, storage_classes=None):
746 super(MagicDirectory, self).__init__(parent_inode, inodes, api.config, enable_write)
748 self.num_retries = num_retries
749 self.pdh_only = pdh_only
750 self.storage_classes = storage_classes
752 def __setattr__(self, name, value):
753 super(MagicDirectory, self).__setattr__(name, value)
754 # When we're assigned an inode, add a README.
755 if ((name == 'inode') and (self.inode is not None) and
756 (not self._entries)):
757 self._entries['README'] = self.inodes.add_entry(
758 StringFile(self.inode, self.README_TEXT, time.time()))
759 # If we're the root directory, add an identical by_id subdirectory.
760 if self.inode == llfuse.ROOT_INODE:
761 self._entries['by_id'] = self.inodes.add_entry(MagicDirectory(
762 self.inode, self.inodes, self.api, self.num_retries, self._enable_write,
765 def __contains__(self, k):
766 if k in self._entries:
769 if not portable_data_hash_pattern.match(k) and (self.pdh_only or not uuid_pattern.match(k)):
775 if group_uuid_pattern.match(k):
776 project = self.api.groups().list(
777 filters=[['group_class', 'in', ['project','filter']], ["uuid", "=", k]]).execute(num_retries=self.num_retries)
778 if project[u'items_available'] == 0:
780 e = self.inodes.add_entry(ProjectDirectory(
781 self.inode, self.inodes, self.api, self.num_retries, self._enable_write,
782 project[u'items'][0], storage_classes=self.storage_classes))
784 e = self.inodes.add_entry(CollectionDirectory(
785 self.inode, self.inodes, self.api, self.num_retries, self._enable_write, k))
788 if k not in self._entries:
791 self.inodes.del_entry(e)
794 self.inodes.invalidate_entry(self, k)
795 self.inodes.del_entry(e)
797 except Exception as ex:
798 _logger.exception("arv-mount lookup '%s':", k)
800 self.inodes.del_entry(e)
803 def __getitem__(self, item):
805 return self._entries[item]
807 raise KeyError("No collection with id " + item)
812 def want_event_subscribe(self):
813 return not self.pdh_only
816 class TagsDirectory(Directory):
817 """A special directory that contains as subdirectories all tags visible to the user."""
819 def __init__(self, parent_inode, inodes, api, num_retries, enable_write, poll_time=60):
820 super(TagsDirectory, self).__init__(parent_inode, inodes, api.config, enable_write)
822 self.num_retries = num_retries
824 self._poll_time = poll_time
827 def want_event_subscribe(self):
832 with llfuse.lock_released:
833 tags = self.api.links().list(
834 filters=[['link_class', '=', 'tag'], ["name", "!=", ""]],
835 select=['name'], distinct=True, limit=1000
836 ).execute(num_retries=self.num_retries)
838 self.merge(tags['items']+[{"name": n} for n in self._extra],
840 lambda a, i: a.tag == i['name'],
841 lambda i: TagDirectory(self.inode, self.inodes, self.api, self.num_retries, self._enable_write,
842 i['name'], poll=self._poll, poll_time=self._poll_time))
846 def __getitem__(self, item):
847 if super(TagsDirectory, self).__contains__(item):
848 return super(TagsDirectory, self).__getitem__(item)
849 with llfuse.lock_released:
850 tags = self.api.links().list(
851 filters=[['link_class', '=', 'tag'], ['name', '=', item]], limit=1
852 ).execute(num_retries=self.num_retries)
854 self._extra.add(item)
856 return super(TagsDirectory, self).__getitem__(item)
860 def __contains__(self, k):
861 if super(TagsDirectory, self).__contains__(k):
871 class TagDirectory(Directory):
872 """A special directory that contains as subdirectories all collections visible
873 to the user that are tagged with a particular tag.
876 def __init__(self, parent_inode, inodes, api, num_retries, enable_write, tag,
877 poll=False, poll_time=60):
878 super(TagDirectory, self).__init__(parent_inode, inodes, api.config, enable_write)
880 self.num_retries = num_retries
883 self._poll_time = poll_time
885 def want_event_subscribe(self):
890 with llfuse.lock_released:
891 taggedcollections = self.api.links().list(
892 filters=[['link_class', '=', 'tag'],
893 ['name', '=', self.tag],
894 ['head_uuid', 'is_a', 'arvados#collection']],
896 ).execute(num_retries=self.num_retries)
897 self.merge(taggedcollections['items'],
898 lambda i: i['head_uuid'],
899 lambda a, i: a.collection_locator == i['head_uuid'],
900 lambda i: CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, self._enable_write, i['head_uuid']))
903 class ProjectDirectory(Directory):
904 """A special directory that contains the contents of a project."""
906 def __init__(self, parent_inode, inodes, api, num_retries, enable_write, project_object,
907 poll=True, poll_time=3, storage_classes=None):
908 super(ProjectDirectory, self).__init__(parent_inode, inodes, api.config, enable_write)
910 self.num_retries = num_retries
911 self.project_object = project_object
912 self.project_object_file = None
913 self.project_uuid = project_object['uuid']
915 self._poll_time = poll_time
916 self._updating_lock = threading.Lock()
917 self._current_user = None
918 self._full_listing = False
919 self.storage_classes = storage_classes
921 def want_event_subscribe(self):
924 def createDirectory(self, i):
925 if collection_uuid_pattern.match(i['uuid']):
926 return CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, self._enable_write, i)
927 elif group_uuid_pattern.match(i['uuid']):
928 return ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, self._enable_write,
929 i, self._poll, self._poll_time, self.storage_classes)
930 elif link_uuid_pattern.match(i['uuid']):
931 if i['head_kind'] == 'arvados#collection' or portable_data_hash_pattern.match(i['head_uuid']):
932 return CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, self._enable_write, i['head_uuid'])
935 elif uuid_pattern.match(i['uuid']):
936 return ObjectFile(self.parent_inode, i)
941 return self.project_uuid
944 self._full_listing = True
945 return super(ProjectDirectory, self).items()
949 if i['name'] is None or len(i['name']) == 0:
951 elif "uuid" in i and (collection_uuid_pattern.match(i['uuid']) or group_uuid_pattern.match(i['uuid'])):
952 # collection or subproject
954 elif link_uuid_pattern.match(i['uuid']) and i['head_kind'] == 'arvados#collection':
957 elif 'kind' in i and i['kind'].startswith('arvados#'):
959 return "{}.{}".format(i['name'], i['kind'][8:])
966 if self.project_object_file == None:
967 self.project_object_file = ObjectFile(self.inode, self.project_object)
968 self.inodes.add_entry(self.project_object_file)
970 if not self._full_listing:
974 if isinstance(a, CollectionDirectory) or isinstance(a, ProjectDirectory):
975 return a.uuid() == i['uuid']
976 elif isinstance(a, ObjectFile):
977 return a.uuid() == i['uuid'] and not a.stale()
981 with llfuse.lock_released:
982 self._updating_lock.acquire()
986 if group_uuid_pattern.match(self.project_uuid):
987 self.project_object = self.api.groups().get(
988 uuid=self.project_uuid).execute(num_retries=self.num_retries)
989 elif user_uuid_pattern.match(self.project_uuid):
990 self.project_object = self.api.users().get(
991 uuid=self.project_uuid).execute(num_retries=self.num_retries)
992 # do this in 2 steps until #17424 is fixed
993 contents = list(arvados.util.keyset_list_all(self.api.groups().contents,
995 num_retries=self.num_retries,
996 uuid=self.project_uuid,
997 filters=[["uuid", "is_a", "arvados#group"],
998 ["groups.group_class", "in", ["project","filter"]]]))
999 contents.extend(filter(lambda i: i["current_version_uuid"] == i["uuid"],
1000 arvados.util.keyset_list_all(self.api.groups().contents,
1002 num_retries=self.num_retries,
1003 uuid=self.project_uuid,
1004 filters=[["uuid", "is_a", "arvados#collection"]])))
1007 # end with llfuse.lock_released, re-acquire lock
1009 self.merge(contents,
1012 self.createDirectory)
1015 self._updating_lock.release()
1017 def _add_entry(self, i, name):
1018 ent = self.createDirectory(i)
1019 self._entries[name] = self.inodes.add_entry(ent)
1020 return self._entries[name]
1024 def __getitem__(self, k):
1025 if k == '.arvados#project':
1026 return self.project_object_file
1027 elif self._full_listing or super(ProjectDirectory, self).__contains__(k):
1028 return super(ProjectDirectory, self).__getitem__(k)
1029 with llfuse.lock_released:
1030 k2 = self.unsanitize_filename(k)
1032 namefilter = ["name", "=", k]
1034 namefilter = ["name", "in", [k, k2]]
1035 contents = self.api.groups().list(filters=[["owner_uuid", "=", self.project_uuid],
1036 ["group_class", "in", ["project","filter"]],
1038 limit=2).execute(num_retries=self.num_retries)["items"]
1040 contents = self.api.collections().list(filters=[["owner_uuid", "=", self.project_uuid],
1042 limit=2).execute(num_retries=self.num_retries)["items"]
1044 if len(contents) > 1 and contents[1]['name'] == k:
1045 # If "foo/bar" and "foo[SUBST]bar" both exist, use
1047 contents = [contents[1]]
1048 name = self.sanitize_filename(self.namefn(contents[0]))
1051 return self._add_entry(contents[0], name)
1056 def __contains__(self, k):
1057 if k == '.arvados#project':
1069 if not self._enable_write:
1071 with llfuse.lock_released:
1072 if not self._current_user:
1073 self._current_user = self.api.users().current().execute(num_retries=self.num_retries)
1074 return self._current_user["uuid"] in self.project_object.get("writable_by", [])
1076 def persisted(self):
1081 def mkdir(self, name):
1082 if not self.writable():
1083 raise llfuse.FUSEError(errno.EROFS)
1086 with llfuse.lock_released:
1088 "owner_uuid": self.project_uuid,
1090 "manifest_text": "" }
1091 if self.storage_classes is not None:
1092 c["storage_classes_desired"] = self.storage_classes
1094 self.api.collections().create(body=c).execute(num_retries=self.num_retries)
1095 except Exception as e:
1098 except apiclient_errors.Error as error:
1099 _logger.error(error)
1100 raise llfuse.FUSEError(errno.EEXIST)
1104 def rmdir(self, name):
1105 if not self.writable():
1106 raise llfuse.FUSEError(errno.EROFS)
1108 if name not in self:
1109 raise llfuse.FUSEError(errno.ENOENT)
1110 if not isinstance(self[name], CollectionDirectory):
1111 raise llfuse.FUSEError(errno.EPERM)
1112 if len(self[name]) > 0:
1113 raise llfuse.FUSEError(errno.ENOTEMPTY)
1114 with llfuse.lock_released:
1115 self.api.collections().delete(uuid=self[name].uuid()).execute(num_retries=self.num_retries)
1120 def rename(self, name_old, name_new, src):
1121 if not self.writable():
1122 raise llfuse.FUSEError(errno.EROFS)
1124 if not isinstance(src, ProjectDirectory):
1125 raise llfuse.FUSEError(errno.EPERM)
1129 if not isinstance(ent, CollectionDirectory):
1130 raise llfuse.FUSEError(errno.EPERM)
1132 if name_new in self:
1133 # POSIX semantics for replacing one directory with another is
1134 # tricky (the target directory must be empty, the operation must be
1135 # atomic which isn't possible with the Arvados API as of this
1136 # writing) so don't support that.
1137 raise llfuse.FUSEError(errno.EPERM)
1139 self.api.collections().update(uuid=ent.uuid(),
1140 body={"owner_uuid": self.uuid(),
1141 "name": name_new}).execute(num_retries=self.num_retries)
1143 # Acually move the entry from source directory to this directory.
1144 del src._entries[name_old]
1145 self._entries[name_new] = ent
1146 self.inodes.invalidate_entry(src, name_old)
1149 def child_event(self, ev):
1150 properties = ev.get("properties") or {}
1151 old_attrs = properties.get("old_attributes") or {}
1152 new_attrs = properties.get("new_attributes") or {}
1153 old_attrs["uuid"] = ev["object_uuid"]
1154 new_attrs["uuid"] = ev["object_uuid"]
1155 old_name = self.sanitize_filename(self.namefn(old_attrs))
1156 new_name = self.sanitize_filename(self.namefn(new_attrs))
1158 # create events will have a new name, but not an old name
1159 # delete events will have an old name, but not a new name
1160 # update events will have an old and new name, and they may be same or different
1161 # if they are the same, an unrelated field changed and there is nothing to do.
1163 if old_attrs.get("owner_uuid") != self.project_uuid:
1164 # Was moved from somewhere else, so don't try to remove entry.
1166 if ev.get("object_owner_uuid") != self.project_uuid:
1167 # Was moved to somewhere else, so don't try to add entry
1170 if old_attrs.get("is_trashed"):
1171 # Was previously deleted
1173 if new_attrs.get("is_trashed"):
1177 if new_name != old_name:
1179 if old_name in self._entries:
1180 ent = self._entries[old_name]
1181 del self._entries[old_name]
1182 self.inodes.invalidate_entry(self, old_name)
1186 self._entries[new_name] = ent
1188 self._add_entry(new_attrs, new_name)
1189 elif ent is not None:
1190 self.inodes.del_entry(ent)
1193 class SharedDirectory(Directory):
1194 """A special directory that represents users or groups who have shared projects with me."""
1196 def __init__(self, parent_inode, inodes, api, num_retries, enable_write, exclude,
1197 poll=False, poll_time=60, storage_classes=None):
1198 super(SharedDirectory, self).__init__(parent_inode, inodes, api.config, enable_write)
1200 self.num_retries = num_retries
1201 self.current_user = api.users().current().execute(num_retries=num_retries)
1203 self._poll_time = poll_time
1204 self._updating_lock = threading.Lock()
1205 self.storage_classes = storage_classes
1210 with llfuse.lock_released:
1211 self._updating_lock.acquire()
1212 if not self.stale():
1220 methods = self.api._rootDesc.get('resources')["groups"]['methods']
1221 if 'httpMethod' in methods.get('shared', {}):
1224 resp = self.api.groups().shared(filters=[['group_class', 'in', ['project','filter']]]+page,
1228 include="owner_uuid").execute()
1229 if not resp["items"]:
1231 page = [["uuid", ">", resp["items"][len(resp["items"])-1]["uuid"]]]
1232 for r in resp["items"]:
1233 objects[r["uuid"]] = r
1234 roots.append(r["uuid"])
1235 for r in resp["included"]:
1236 objects[r["uuid"]] = r
1237 root_owners.add(r["uuid"])
1239 all_projects = list(arvados.util.keyset_list_all(
1240 self.api.groups().list,
1242 num_retries=self.num_retries,
1243 filters=[['group_class','in',['project','filter']]],
1244 select=["uuid", "owner_uuid"]))
1245 for ob in all_projects:
1246 objects[ob['uuid']] = ob
1248 current_uuid = self.current_user['uuid']
1249 for ob in all_projects:
1250 if ob['owner_uuid'] != current_uuid and ob['owner_uuid'] not in objects:
1251 roots.append(ob['uuid'])
1252 root_owners.add(ob['owner_uuid'])
1254 lusers = arvados.util.keyset_list_all(
1255 self.api.users().list,
1257 num_retries=self.num_retries,
1258 filters=[['uuid','in', list(root_owners)]])
1259 lgroups = arvados.util.keyset_list_all(
1260 self.api.groups().list,
1262 num_retries=self.num_retries,
1263 filters=[['uuid','in', list(root_owners)+roots]])
1266 objects[l["uuid"]] = l
1268 objects[l["uuid"]] = l
1270 for r in root_owners:
1274 contents[obr["name"]] = obr
1275 elif "first_name" in obr:
1276 contents[u"{} {}".format(obr["first_name"], obr["last_name"])] = obr
1281 if obr['owner_uuid'] not in objects:
1282 contents[obr["name"]] = obr
1284 # end with llfuse.lock_released, re-acquire lock
1286 self.merge(contents.items(),
1288 lambda a, i: a.uuid() == i[1]['uuid'],
1289 lambda i: ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, self._enable_write,
1290 i[1], poll=self._poll, poll_time=self._poll_time, storage_classes=self.storage_classes))
1292 _logger.exception("arv-mount shared dir error")
1294 self._updating_lock.release()
1296 def want_event_subscribe(self):