services/fuse/arvados_fuse/fusedir.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4
   5 from __future__ import absolute_import
   6 from __future__ import division
   7 from future.utils import viewitems
   8 from future.utils import itervalues
   9 from builtins import dict
  10 import apiclient
  11 import arvados
  12 import errno
  13 import functools
  14 import llfuse
  15 import logging
  16 import re
  17 import sys
  18 import threading
  19 import time
  20 from apiclient import errors as apiclient_errors
  21
  22 from .fusefile import StringFile, ObjectFile, FuncToJSONFile, FuseArvadosFile
  23 from .fresh import FreshBase, convertTime, use_counter, check_update
  24
  25 import arvados.collection
  26 from arvados.util import portable_data_hash_pattern, uuid_pattern, collection_uuid_pattern, group_uuid_pattern, user_uuid_pattern, link_uuid_pattern
  27
  28 _logger = logging.getLogger('arvados.arvados_fuse')
  29
  30
  31 # Match any character which FUSE or Linux cannot accommodate as part
  32 # of a filename. (If present in a collection filename, they will
  33 # appear as underscores in the fuse mount.)
  34 _disallowed_filename_characters = re.compile('[\x00/]')
  35
  36
  37 class Directory(FreshBase):
  38     """Generic directory object, backed by a dict.
  39
  40     Consists of a set of entries with the key representing the filename
  41     and the value referencing a File or Directory object.
  42     """
  43
  44     def __init__(self, parent_inode, inodes, apiconfig):
  45         """parent_inode is the integer inode number"""
  46
  47         super(Directory, self).__init__()
  48
  49         self.inode = None
  50         if not isinstance(parent_inode, int):
  51             raise Exception("parent_inode should be an int")
  52         self.parent_inode = parent_inode
  53         self.inodes = inodes
  54         self.apiconfig = apiconfig
  55         self._entries = {}
  56         self._mtime = time.time()
  57
  58     def forward_slash_subst(self):
  59         if not hasattr(self, '_fsns'):
  60             self._fsns = None
  61             config = self.apiconfig()
  62             try:
  63                 self._fsns = config["Collections"]["ForwardSlashNameSubstitution"]
  64             except KeyError:
  65                 # old API server with no FSNS config
  66                 self._fsns = '_'
  67             else:
  68                 if self._fsns == '' or self._fsns == '/':
  69                     self._fsns = None
  70         return self._fsns
  71
  72     def unsanitize_filename(self, incoming):
  73         """Replace ForwardSlashNameSubstitution value with /"""
  74         fsns = self.forward_slash_subst()
  75         if isinstance(fsns, str):
  76             return incoming.replace(fsns, '/')
  77         else:
  78             return incoming
  79
  80     def sanitize_filename(self, dirty):
  81         """Replace disallowed filename characters according to
  82         ForwardSlashNameSubstitution in self.api_config."""
  83         # '.' and '..' are not reachable if API server is newer than #6277
  84         if dirty is None:
  85             return None
  86         elif dirty == '':
  87             return '_'
  88         elif dirty == '.':
  89             return '_'
  90         elif dirty == '..':
  91             return '__'
  92         else:
  93             fsns = self.forward_slash_subst()
  94             if isinstance(fsns, str):
  95                 dirty = dirty.replace('/', fsns)
  96             return _disallowed_filename_characters.sub('_', dirty)
  97
  98
  99     #  Overridden by subclasses to implement logic to update the
 100     #  entries dict when the directory is stale
 101     @use_counter
 102     def update(self):
 103         pass
 104
 105     # Only used when computing the size of the disk footprint of the directory
 106     # (stub)
 107     def size(self):
 108         return 0
 109
 110     def persisted(self):
 111         return False
 112
 113     def checkupdate(self):
 114         if self.stale():
 115             try:
 116                 self.update()
 117             except apiclient.errors.HttpError as e:
 118                 _logger.warn(e)
 119
 120     @use_counter
 121     @check_update
 122     def __getitem__(self, item):
 123         return self._entries[item]
 124
 125     @use_counter
 126     @check_update
 127     def items(self):
 128         return list(self._entries.items())
 129
 130     @use_counter
 131     @check_update
 132     def __contains__(self, k):
 133         return k in self._entries
 134
 135     @use_counter
 136     @check_update
 137     def __len__(self):
 138         return len(self._entries)
 139
 140     def fresh(self):
 141         self.inodes.touch(self)
 142         super(Directory, self).fresh()
 143
 144     def merge(self, items, fn, same, new_entry):
 145         """Helper method for updating the contents of the directory.
 146
 147         Takes a list describing the new contents of the directory, reuse
 148         entries that are the same in both the old and new lists, create new
 149         entries, and delete old entries missing from the new list.
 150
 151         :items: iterable with new directory contents
 152
 153         :fn: function to take an entry in 'items' and return the desired file or
 154         directory name, or None if this entry should be skipped
 155
 156         :same: function to compare an existing entry (a File or Directory
 157         object) with an entry in the items list to determine whether to keep
 158         the existing entry.
 159
 160         :new_entry: function to create a new directory entry (File or Directory
 161         object) from an entry in the items list.
 162
 163         """
 164
 165         oldentries = self._entries
 166         self._entries = {}
 167         changed = False
 168         for i in items:
 169             name = self.sanitize_filename(fn(i))
 170             if name:
 171                 if name in oldentries and same(oldentries[name], i):
 172                     # move existing directory entry over
 173                     self._entries[name] = oldentries[name]
 174                     del oldentries[name]
 175                 else:
 176                     _logger.debug("Adding entry '%s' to inode %i", name, self.inode)
 177                     # create new directory entry
 178                     ent = new_entry(i)
 179                     if ent is not None:
 180                         self._entries[name] = self.inodes.add_entry(ent)
 181                         changed = True
 182
 183         # delete any other directory entries that were not in found in 'items'
 184         for i in oldentries:
 185             _logger.debug("Forgetting about entry '%s' on inode %i", i, self.inode)
 186             self.inodes.invalidate_entry(self, i)
 187             self.inodes.del_entry(oldentries[i])
 188             changed = True
 189
 190         if changed:
 191             self.inodes.invalidate_inode(self)
 192             self._mtime = time.time()
 193
 194         self.fresh()
 195
 196     def in_use(self):
 197         if super(Directory, self).in_use():
 198             return True
 199         for v in itervalues(self._entries):
 200             if v.in_use():
 201                 return True
 202         return False
 203
 204     def has_ref(self, only_children):
 205         if super(Directory, self).has_ref(only_children):
 206             return True
 207         for v in itervalues(self._entries):
 208             if v.has_ref(False):
 209                 return True
 210         return False
 211
 212     def clear(self):
 213         """Delete all entries"""
 214         oldentries = self._entries
 215         self._entries = {}
 216         for n in oldentries:
 217             oldentries[n].clear()
 218             self.inodes.del_entry(oldentries[n])
 219         self.invalidate()
 220
 221     def kernel_invalidate(self):
 222         # Invalidating the dentry on the parent implies invalidating all paths
 223         # below it as well.
 224         parent = self.inodes[self.parent_inode]
 225
 226         # Find self on the parent in order to invalidate this path.
 227         # Calling the public items() method might trigger a refresh,
 228         # which we definitely don't want, so read the internal dict directly.
 229         for k,v in viewitems(parent._entries):
 230             if v is self:
 231                 self.inodes.invalidate_entry(parent, k)
 232                 break
 233
 234     def mtime(self):
 235         return self._mtime
 236
 237     def writable(self):
 238         return False
 239
 240     def flush(self):
 241         pass
 242
 243     def want_event_subscribe(self):
 244         raise NotImplementedError()
 245
 246     def create(self, name):
 247         raise NotImplementedError()
 248
 249     def mkdir(self, name):
 250         raise NotImplementedError()
 251
 252     def unlink(self, name):
 253         raise NotImplementedError()
 254
 255     def rmdir(self, name):
 256         raise NotImplementedError()
 257
 258     def rename(self, name_old, name_new, src):
 259         raise NotImplementedError()
 260
 261
 262 class CollectionDirectoryBase(Directory):
 263     """Represent an Arvados Collection as a directory.
 264
 265     This class is used for Subcollections, and is also the base class for
 266     CollectionDirectory, which implements collection loading/saving on
 267     Collection records.
 268
 269     Most operations act only the underlying Arvados `Collection` object.  The
 270     `Collection` object signals via a notify callback to
 271     `CollectionDirectoryBase.on_event` that an item was added, removed or
 272     modified.  FUSE inodes and directory entries are created, deleted or
 273     invalidated in response to these events.
 274
 275     """
 276
 277     def __init__(self, parent_inode, inodes, apiconfig, collection):
 278         super(CollectionDirectoryBase, self).__init__(parent_inode, inodes, apiconfig)
 279         self.apiconfig = apiconfig
 280         self.collection = collection
 281
 282     def new_entry(self, name, item, mtime):
 283         name = self.sanitize_filename(name)
 284         if hasattr(item, "fuse_entry") and item.fuse_entry is not None:
 285             if item.fuse_entry.dead is not True:
 286                 raise Exception("Can only reparent dead inode entry")
 287             if item.fuse_entry.inode is None:
 288                 raise Exception("Reparented entry must still have valid inode")
 289             item.fuse_entry.dead = False
 290             self._entries[name] = item.fuse_entry
 291         elif isinstance(item, arvados.collection.RichCollectionBase):
 292             self._entries[name] = self.inodes.add_entry(CollectionDirectoryBase(self.inode, self.inodes, self.apiconfig, item))
 293             self._entries[name].populate(mtime)
 294         else:
 295             self._entries[name] = self.inodes.add_entry(FuseArvadosFile(self.inode, item, mtime))
 296         item.fuse_entry = self._entries[name]
 297
 298     def on_event(self, event, collection, name, item):
 299         if collection == self.collection:
 300             name = self.sanitize_filename(name)
 301             _logger.debug("collection notify %s %s %s %s", event, collection, name, item)
 302             with llfuse.lock:
 303                 if event == arvados.collection.ADD:
 304                     self.new_entry(name, item, self.mtime())
 305                 elif event == arvados.collection.DEL:
 306                     ent = self._entries[name]
 307                     del self._entries[name]
 308                     self.inodes.invalidate_entry(self, name)
 309                     self.inodes.del_entry(ent)
 310                 elif event == arvados.collection.MOD:
 311                     if hasattr(item, "fuse_entry") and item.fuse_entry is not None:
 312                         self.inodes.invalidate_inode(item.fuse_entry)
 313                     elif name in self._entries:
 314                         self.inodes.invalidate_inode(self._entries[name])
 315
 316     def populate(self, mtime):
 317         self._mtime = mtime
 318         self.collection.subscribe(self.on_event)
 319         for entry, item in viewitems(self.collection):
 320             self.new_entry(entry, item, self.mtime())
 321
 322     def writable(self):
 323         return self.collection.writable()
 324
 325     @use_counter
 326     def flush(self):
 327         with llfuse.lock_released:
 328             self.collection.root_collection().save()
 329
 330     @use_counter
 331     @check_update
 332     def create(self, name):
 333         with llfuse.lock_released:
 334             self.collection.open(name, "w").close()
 335
 336     @use_counter
 337     @check_update
 338     def mkdir(self, name):
 339         with llfuse.lock_released:
 340             self.collection.mkdirs(name)
 341
 342     @use_counter
 343     @check_update
 344     def unlink(self, name):
 345         with llfuse.lock_released:
 346             self.collection.remove(name)
 347         self.flush()
 348
 349     @use_counter
 350     @check_update
 351     def rmdir(self, name):
 352         with llfuse.lock_released:
 353             self.collection.remove(name)
 354         self.flush()
 355
 356     @use_counter
 357     @check_update
 358     def rename(self, name_old, name_new, src):
 359         if not isinstance(src, CollectionDirectoryBase):
 360             raise llfuse.FUSEError(errno.EPERM)
 361
 362         if name_new in self:
 363             ent = src[name_old]
 364             tgt = self[name_new]
 365             if isinstance(ent, FuseArvadosFile) and isinstance(tgt, FuseArvadosFile):
 366                 pass
 367             elif isinstance(ent, CollectionDirectoryBase) and isinstance(tgt, CollectionDirectoryBase):
 368                 if len(tgt) > 0:
 369                     raise llfuse.FUSEError(errno.ENOTEMPTY)
 370             elif isinstance(ent, CollectionDirectoryBase) and isinstance(tgt, FuseArvadosFile):
 371                 raise llfuse.FUSEError(errno.ENOTDIR)
 372             elif isinstance(ent, FuseArvadosFile) and isinstance(tgt, CollectionDirectoryBase):
 373                 raise llfuse.FUSEError(errno.EISDIR)
 374
 375         with llfuse.lock_released:
 376             self.collection.rename(name_old, name_new, source_collection=src.collection, overwrite=True)
 377         self.flush()
 378         src.flush()
 379
 380     def clear(self):
 381         super(CollectionDirectoryBase, self).clear()
 382         self.collection = None
 383
 384
 385 class CollectionDirectory(CollectionDirectoryBase):
 386     """Represents the root of a directory tree representing a collection."""
 387
 388     def __init__(self, parent_inode, inodes, api, num_retries, collection_record=None, explicit_collection=None):
 389         super(CollectionDirectory, self).__init__(parent_inode, inodes, api.config, None)
 390         self.api = api
 391         self.num_retries = num_retries
 392         self.collection_record_file = None
 393         self.collection_record = None
 394         self._poll = True
 395         try:
 396             self._poll_time = (api._rootDesc.get('blobSignatureTtl', 60*60*2) // 2)
 397         except:
 398             _logger.debug("Error getting blobSignatureTtl from discovery document: %s", sys.exc_info()[0])
 399             self._poll_time = 60*60
 400
 401         if isinstance(collection_record, dict):
 402             self.collection_locator = collection_record['uuid']
 403             self._mtime = convertTime(collection_record.get('modified_at'))
 404         else:
 405             self.collection_locator = collection_record
 406             self._mtime = 0
 407         self._manifest_size = 0
 408         if self.collection_locator:
 409             self._writable = (uuid_pattern.match(self.collection_locator) is not None)
 410         self._updating_lock = threading.Lock()
 411
 412     def same(self, i):
 413         return i['uuid'] == self.collection_locator or i['portable_data_hash'] == self.collection_locator
 414
 415     def writable(self):
 416         return self.collection.writable() if self.collection is not None else self._writable
 417
 418     def want_event_subscribe(self):
 419         return (uuid_pattern.match(self.collection_locator) is not None)
 420
 421     # Used by arv-web.py to switch the contents of the CollectionDirectory
 422     def change_collection(self, new_locator):
 423         """Switch the contents of the CollectionDirectory.
 424
 425         Must be called with llfuse.lock held.
 426         """
 427
 428         self.collection_locator = new_locator
 429         self.collection_record = None
 430         self.update()
 431
 432     def new_collection(self, new_collection_record, coll_reader):
 433         if self.inode:
 434             self.clear()
 435
 436         self.collection_record = new_collection_record
 437
 438         if self.collection_record:
 439             self._mtime = convertTime(self.collection_record.get('modified_at'))
 440             self.collection_locator = self.collection_record["uuid"]
 441             if self.collection_record_file is not None:
 442                 self.collection_record_file.update(self.collection_record)
 443
 444         self.collection = coll_reader
 445         self.populate(self.mtime())
 446
 447     def uuid(self):
 448         return self.collection_locator
 449
 450     @use_counter
 451     def update(self, to_record_version=None):
 452         try:
 453             if self.collection_record is not None and portable_data_hash_pattern.match(self.collection_locator):
 454                 return True
 455
 456             if self.collection_locator is None:
 457                 self.fresh()
 458                 return True
 459
 460             try:
 461                 with llfuse.lock_released:
 462                     self._updating_lock.acquire()
 463                     if not self.stale():
 464                         return
 465
 466                     _logger.debug("Updating collection %s inode %s to record version %s", self.collection_locator, self.inode, to_record_version)
 467                     if self.collection is not None:
 468                         if self.collection.known_past_version(to_record_version):
 469                             _logger.debug("%s already processed %s", self.collection_locator, to_record_version)
 470                         else:
 471                             self.collection.update()
 472                     else:
 473                         if uuid_pattern.match(self.collection_locator):
 474                             coll_reader = arvados.collection.Collection(
 475                                 self.collection_locator, self.api, self.api.keep,
 476                                 num_retries=self.num_retries)
 477                         else:
 478                             coll_reader = arvados.collection.CollectionReader(
 479                                 self.collection_locator, self.api, self.api.keep,
 480                                 num_retries=self.num_retries)
 481                         new_collection_record = coll_reader.api_response() or {}
 482                         # If the Collection only exists in Keep, there will be no API
 483                         # response.  Fill in the fields we need.
 484                         if 'uuid' not in new_collection_record:
 485                             new_collection_record['uuid'] = self.collection_locator
 486                         if "portable_data_hash" not in new_collection_record:
 487                             new_collection_record["portable_data_hash"] = new_collection_record["uuid"]
 488                         if 'manifest_text' not in new_collection_record:
 489                             new_collection_record['manifest_text'] = coll_reader.manifest_text()
 490
 491                         if self.collection_record is None or self.collection_record["portable_data_hash"] != new_collection_record.get("portable_data_hash"):
 492                             self.new_collection(new_collection_record, coll_reader)
 493
 494                         self._manifest_size = len(coll_reader.manifest_text())
 495                         _logger.debug("%s manifest_size %i", self, self._manifest_size)
 496                 # end with llfuse.lock_released, re-acquire lock
 497
 498                 self.fresh()
 499                 return True
 500             finally:
 501                 self._updating_lock.release()
 502         except arvados.errors.NotFoundError as e:
 503             _logger.error("Error fetching collection '%s': %s", self.collection_locator, e)
 504         except arvados.errors.ArgumentError as detail:
 505             _logger.warning("arv-mount %s: error %s", self.collection_locator, detail)
 506             if self.collection_record is not None and "manifest_text" in self.collection_record:
 507                 _logger.warning("arv-mount manifest_text is: %s", self.collection_record["manifest_text"])
 508         except Exception:
 509             _logger.exception("arv-mount %s: error", self.collection_locator)
 510             if self.collection_record is not None and "manifest_text" in self.collection_record:
 511                 _logger.error("arv-mount manifest_text is: %s", self.collection_record["manifest_text"])
 512         self.invalidate()
 513         return False
 514
 515     @use_counter
 516     @check_update
 517     def __getitem__(self, item):
 518         if item == '.arvados#collection':
 519             if self.collection_record_file is None:
 520                 self.collection_record_file = ObjectFile(self.inode, self.collection_record)
 521                 self.inodes.add_entry(self.collection_record_file)
 522             return self.collection_record_file
 523         else:
 524             return super(CollectionDirectory, self).__getitem__(item)
 525
 526     def __contains__(self, k):
 527         if k == '.arvados#collection':
 528             return True
 529         else:
 530             return super(CollectionDirectory, self).__contains__(k)
 531
 532     def invalidate(self):
 533         self.collection_record = None
 534         self.collection_record_file = None
 535         super(CollectionDirectory, self).invalidate()
 536
 537     def persisted(self):
 538         return (self.collection_locator is not None)
 539
 540     def objsize(self):
 541         # This is an empirically-derived heuristic to estimate the memory used
 542         # to store this collection's metadata.  Calculating the memory
 543         # footprint directly would be more accurate, but also more complicated.
 544         return self._manifest_size * 128
 545
 546     def finalize(self):
 547         if self.collection is not None:
 548             if self.writable():
 549                 self.collection.save()
 550             self.collection.stop_threads()
 551
 552     def clear(self):
 553         if self.collection is not None:
 554             self.collection.stop_threads()
 555         super(CollectionDirectory, self).clear()
 556         self._manifest_size = 0
 557
 558
 559 class TmpCollectionDirectory(CollectionDirectoryBase):
 560     """A directory backed by an Arvados collection that never gets saved.
 561
 562     This supports using Keep as scratch space. A userspace program can
 563     read the .arvados#collection file to get a current manifest in
 564     order to save a snapshot of the scratch data or use it as a crunch
 565     job output.
 566     """
 567
 568     class UnsaveableCollection(arvados.collection.Collection):
 569         def save(self):
 570             pass
 571         def save_new(self):
 572             pass
 573
 574     def __init__(self, parent_inode, inodes, api_client, num_retries):
 575         collection = self.UnsaveableCollection(
 576             api_client=api_client,
 577             keep_client=api_client.keep,
 578             num_retries=num_retries)
 579         super(TmpCollectionDirectory, self).__init__(
 580             parent_inode, inodes, api_client.config, collection)
 581         self.collection_record_file = None
 582         self.populate(self.mtime())
 583
 584     def on_event(self, *args, **kwargs):
 585         super(TmpCollectionDirectory, self).on_event(*args, **kwargs)
 586         if self.collection_record_file:
 587             with llfuse.lock:
 588                 self.collection_record_file.invalidate()
 589             self.inodes.invalidate_inode(self.collection_record_file)
 590             _logger.debug("%s invalidated collection record", self)
 591
 592     def collection_record(self):
 593         with llfuse.lock_released:
 594             return {
 595                 "uuid": None,
 596                 "manifest_text": self.collection.manifest_text(),
 597                 "portable_data_hash": self.collection.portable_data_hash(),
 598             }
 599
 600     def __contains__(self, k):
 601         return (k == '.arvados#collection' or
 602                 super(TmpCollectionDirectory, self).__contains__(k))
 603
 604     @use_counter
 605     def __getitem__(self, item):
 606         if item == '.arvados#collection':
 607             if self.collection_record_file is None:
 608                 self.collection_record_file = FuncToJSONFile(
 609                     self.inode, self.collection_record)
 610                 self.inodes.add_entry(self.collection_record_file)
 611             return self.collection_record_file
 612         return super(TmpCollectionDirectory, self).__getitem__(item)
 613
 614     def persisted(self):
 615         return False
 616
 617     def writable(self):
 618         return True
 619
 620     def want_event_subscribe(self):
 621         return False
 622
 623     def finalize(self):
 624         self.collection.stop_threads()
 625
 626     def invalidate(self):
 627         if self.collection_record_file:
 628             self.collection_record_file.invalidate()
 629         super(TmpCollectionDirectory, self).invalidate()
 630
 631
 632 class MagicDirectory(Directory):
 633     """A special directory that logically contains the set of all extant keep locators.
 634
 635     When a file is referenced by lookup(), it is tested to see if it is a valid
 636     keep locator to a manifest, and if so, loads the manifest contents as a
 637     subdirectory of this directory with the locator as the directory name.
 638     Since querying a list of all extant keep locators is impractical, only
 639     collections that have already been accessed are visible to readdir().
 640
 641     """
 642
 643     README_TEXT = """
 644 This directory provides access to Arvados collections as subdirectories listed
 645 by uuid (in the form 'zzzzz-4zz18-1234567890abcde') or portable data hash (in
 646 the form '1234567890abcdef0123456789abcdef+123'), and Arvados projects by uuid
 647 (in the form 'zzzzz-j7d0g-1234567890abcde').
 648
 649 Note that this directory will appear empty until you attempt to access a
 650 specific collection or project subdirectory (such as trying to 'cd' into it),
 651 at which point the collection or project will actually be looked up on the server
 652 and the directory will appear if it exists.
 653
 654 """.lstrip()
 655
 656     def __init__(self, parent_inode, inodes, api, num_retries, pdh_only=False):
 657         super(MagicDirectory, self).__init__(parent_inode, inodes, api.config)
 658         self.api = api
 659         self.num_retries = num_retries
 660         self.pdh_only = pdh_only
 661
 662     def __setattr__(self, name, value):
 663         super(MagicDirectory, self).__setattr__(name, value)
 664         # When we're assigned an inode, add a README.
 665         if ((name == 'inode') and (self.inode is not None) and
 666               (not self._entries)):
 667             self._entries['README'] = self.inodes.add_entry(
 668                 StringFile(self.inode, self.README_TEXT, time.time()))
 669             # If we're the root directory, add an identical by_id subdirectory.
 670             if self.inode == llfuse.ROOT_INODE:
 671                 self._entries['by_id'] = self.inodes.add_entry(MagicDirectory(
 672                         self.inode, self.inodes, self.api, self.num_retries, self.pdh_only))
 673
 674     def __contains__(self, k):
 675         if k in self._entries:
 676             return True
 677
 678         if not portable_data_hash_pattern.match(k) and (self.pdh_only or not uuid_pattern.match(k)):
 679             return False
 680
 681         try:
 682             e = None
 683
 684             if group_uuid_pattern.match(k):
 685                 project = self.api.groups().list(
 686                     filters=[['group_class', '=', 'project'], ["uuid", "=", k]]).execute(num_retries=self.num_retries)
 687                 if project[u'items_available'] == 0:
 688                     return False
 689                 e = self.inodes.add_entry(ProjectDirectory(
 690                     self.inode, self.inodes, self.api, self.num_retries, project[u'items'][0]))
 691             else:
 692                 e = self.inodes.add_entry(CollectionDirectory(
 693                         self.inode, self.inodes, self.api, self.num_retries, k))
 694
 695             if e.update():
 696                 if k not in self._entries:
 697                     self._entries[k] = e
 698                 else:
 699                     self.inodes.del_entry(e)
 700                 return True
 701             else:
 702                 self.inodes.invalidate_entry(self, k)
 703                 self.inodes.del_entry(e)
 704                 return False
 705         except Exception as ex:
 706             _logger.exception("arv-mount lookup '%s':", k)
 707             if e is not None:
 708                 self.inodes.del_entry(e)
 709             return False
 710
 711     def __getitem__(self, item):
 712         if item in self:
 713             return self._entries[item]
 714         else:
 715             raise KeyError("No collection with id " + item)
 716
 717     def clear(self):
 718         pass
 719
 720     def want_event_subscribe(self):
 721         return not self.pdh_only
 722
 723
 724 class TagsDirectory(Directory):
 725     """A special directory that contains as subdirectories all tags visible to the user."""
 726
 727     def __init__(self, parent_inode, inodes, api, num_retries, poll_time=60):
 728         super(TagsDirectory, self).__init__(parent_inode, inodes, api.config)
 729         self.api = api
 730         self.num_retries = num_retries
 731         self._poll = True
 732         self._poll_time = poll_time
 733         self._extra = set()
 734
 735     def want_event_subscribe(self):
 736         return True
 737
 738     @use_counter
 739     def update(self):
 740         with llfuse.lock_released:
 741             tags = self.api.links().list(
 742                 filters=[['link_class', '=', 'tag'], ["name", "!=", ""]],
 743                 select=['name'], distinct=True, limit=1000
 744                 ).execute(num_retries=self.num_retries)
 745         if "items" in tags:
 746             self.merge(tags['items']+[{"name": n} for n in self._extra],
 747                        lambda i: i['name'],
 748                        lambda a, i: a.tag == i['name'],
 749                        lambda i: TagDirectory(self.inode, self.inodes, self.api, self.num_retries, i['name'], poll=self._poll, poll_time=self._poll_time))
 750
 751     @use_counter
 752     @check_update
 753     def __getitem__(self, item):
 754         if super(TagsDirectory, self).__contains__(item):
 755             return super(TagsDirectory, self).__getitem__(item)
 756         with llfuse.lock_released:
 757             tags = self.api.links().list(
 758                 filters=[['link_class', '=', 'tag'], ['name', '=', item]], limit=1
 759             ).execute(num_retries=self.num_retries)
 760         if tags["items"]:
 761             self._extra.add(item)
 762             self.update()
 763         return super(TagsDirectory, self).__getitem__(item)
 764
 765     @use_counter
 766     @check_update
 767     def __contains__(self, k):
 768         if super(TagsDirectory, self).__contains__(k):
 769             return True
 770         try:
 771             self[k]
 772             return True
 773         except KeyError:
 774             pass
 775         return False
 776
 777
 778 class TagDirectory(Directory):
 779     """A special directory that contains as subdirectories all collections visible
 780     to the user that are tagged with a particular tag.
 781     """
 782
 783     def __init__(self, parent_inode, inodes, api, num_retries, tag,
 784                  poll=False, poll_time=60):
 785         super(TagDirectory, self).__init__(parent_inode, inodes, api.config)
 786         self.api = api
 787         self.num_retries = num_retries
 788         self.tag = tag
 789         self._poll = poll
 790         self._poll_time = poll_time
 791
 792     def want_event_subscribe(self):
 793         return True
 794
 795     @use_counter
 796     def update(self):
 797         with llfuse.lock_released:
 798             taggedcollections = self.api.links().list(
 799                 filters=[['link_class', '=', 'tag'],
 800                          ['name', '=', self.tag],
 801                          ['head_uuid', 'is_a', 'arvados#collection']],
 802                 select=['head_uuid']
 803                 ).execute(num_retries=self.num_retries)
 804         self.merge(taggedcollections['items'],
 805                    lambda i: i['head_uuid'],
 806                    lambda a, i: a.collection_locator == i['head_uuid'],
 807                    lambda i: CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i['head_uuid']))
 808
 809
 810 class ProjectDirectory(Directory):
 811     """A special directory that contains the contents of a project."""
 812
 813     def __init__(self, parent_inode, inodes, api, num_retries, project_object,
 814                  poll=False, poll_time=60):
 815         super(ProjectDirectory, self).__init__(parent_inode, inodes, api.config)
 816         self.api = api
 817         self.num_retries = num_retries
 818         self.project_object = project_object
 819         self.project_object_file = None
 820         self.project_uuid = project_object['uuid']
 821         self._poll = poll
 822         self._poll_time = poll_time
 823         self._updating_lock = threading.Lock()
 824         self._current_user = None
 825         self._full_listing = False
 826
 827     def want_event_subscribe(self):
 828         return True
 829
 830     def createDirectory(self, i):
 831         if collection_uuid_pattern.match(i['uuid']):
 832             return CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i)
 833         elif group_uuid_pattern.match(i['uuid']):
 834             return ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, i, self._poll, self._poll_time)
 835         elif link_uuid_pattern.match(i['uuid']):
 836             if i['head_kind'] == 'arvados#collection' or portable_data_hash_pattern.match(i['head_uuid']):
 837                 return CollectionDirectory(self.inode, self.inodes, self.api, self.num_retries, i['head_uuid'])
 838             else:
 839                 return None
 840         elif uuid_pattern.match(i['uuid']):
 841             return ObjectFile(self.parent_inode, i)
 842         else:
 843             return None
 844
 845     def uuid(self):
 846         return self.project_uuid
 847
 848     def items(self):
 849         self._full_listing = True
 850         return super(ProjectDirectory, self).items()
 851
 852     def namefn(self, i):
 853         if 'name' in i:
 854             if i['name'] is None or len(i['name']) == 0:
 855                 return None
 856             elif "uuid" in i and (collection_uuid_pattern.match(i['uuid']) or group_uuid_pattern.match(i['uuid'])):
 857                 # collection or subproject
 858                 return i['name']
 859             elif link_uuid_pattern.match(i['uuid']) and i['head_kind'] == 'arvados#collection':
 860                 # name link
 861                 return i['name']
 862             elif 'kind' in i and i['kind'].startswith('arvados#'):
 863                 # something else
 864                 return "{}.{}".format(i['name'], i['kind'][8:])
 865         else:
 866             return None
 867
 868
 869     @use_counter
 870     def update(self):
 871         if self.project_object_file == None:
 872             self.project_object_file = ObjectFile(self.inode, self.project_object)
 873             self.inodes.add_entry(self.project_object_file)
 874
 875         if not self._full_listing:
 876             return True
 877
 878         def samefn(a, i):
 879             if isinstance(a, CollectionDirectory) or isinstance(a, ProjectDirectory):
 880                 return a.uuid() == i['uuid']
 881             elif isinstance(a, ObjectFile):
 882                 return a.uuid() == i['uuid'] and not a.stale()
 883             return False
 884
 885         try:
 886             with llfuse.lock_released:
 887                 self._updating_lock.acquire()
 888                 if not self.stale():
 889                     return
 890
 891                 if group_uuid_pattern.match(self.project_uuid):
 892                     self.project_object = self.api.groups().get(
 893                         uuid=self.project_uuid).execute(num_retries=self.num_retries)
 894                 elif user_uuid_pattern.match(self.project_uuid):
 895                     self.project_object = self.api.users().get(
 896                         uuid=self.project_uuid).execute(num_retries=self.num_retries)
 897                 # do this in 2 steps until #17424 is fixed
 898                 contents = arvados.util.list_all(self.api.groups().contents,
 899                                                  self.num_retries,
 900                                                  uuid=self.project_uuid,
 901                                                  filters=[["uuid", "is_a", "arvados#group"],
 902                                                           ["group_class", "=", "project"]])
 903                 contents.extend(arvados.util.list_all(self.api.groups().contents,
 904                                                       self.num_retries,
 905                                                       uuid=self.project_uuid,
 906                                                       filters=[["uuid", "is_a", "arvados#collection"]]))
 907
 908             # end with llfuse.lock_released, re-acquire lock
 909
 910             self.merge(contents,
 911                        self.namefn,
 912                        samefn,
 913                        self.createDirectory)
 914             return True
 915         finally:
 916             self._updating_lock.release()
 917
 918     def _add_entry(self, i, name):
 919         ent = self.createDirectory(i)
 920         self._entries[name] = self.inodes.add_entry(ent)
 921         return self._entries[name]
 922
 923     @use_counter
 924     @check_update
 925     def __getitem__(self, k):
 926         if k == '.arvados#project':
 927             return self.project_object_file
 928         elif self._full_listing or super(ProjectDirectory, self).__contains__(k):
 929             return super(ProjectDirectory, self).__getitem__(k)
 930         with llfuse.lock_released:
 931             k2 = self.unsanitize_filename(k)
 932             if k2 == k:
 933                 namefilter = ["name", "=", k]
 934             else:
 935                 namefilter = ["name", "in", [k, k2]]
 936             contents = self.api.groups().list(filters=[["owner_uuid", "=", self.project_uuid],
 937                                                        ["group_class", "=", "project"],
 938                                                        namefilter],
 939                                               limit=2).execute(num_retries=self.num_retries)["items"]
 940             if not contents:
 941                 contents = self.api.collections().list(filters=[["owner_uuid", "=", self.project_uuid],
 942                                                                 namefilter],
 943                                                        limit=2).execute(num_retries=self.num_retries)["items"]
 944         if contents:
 945             if len(contents) > 1 and contents[1]['name'] == k:
 946                 # If "foo/bar" and "foo[SUBST]bar" both exist, use
 947                 # "foo[SUBST]bar".
 948                 contents = [contents[1]]
 949             name = self.sanitize_filename(self.namefn(contents[0]))
 950             if name != k:
 951                 raise KeyError(k)
 952             return self._add_entry(contents[0], name)
 953
 954         # Didn't find item
 955         raise KeyError(k)
 956
 957     def __contains__(self, k):
 958         if k == '.arvados#project':
 959             return True
 960         try:
 961             self[k]
 962             return True
 963         except KeyError:
 964             pass
 965         return False
 966
 967     @use_counter
 968     @check_update
 969     def writable(self):
 970         with llfuse.lock_released:
 971             if not self._current_user:
 972                 self._current_user = self.api.users().current().execute(num_retries=self.num_retries)
 973             return self._current_user["uuid"] in self.project_object.get("writable_by", [])
 974
 975     def persisted(self):
 976         return True
 977
 978     @use_counter
 979     @check_update
 980     def mkdir(self, name):
 981         try:
 982             with llfuse.lock_released:
 983                 self.api.collections().create(body={"owner_uuid": self.project_uuid,
 984                                                     "name": name,
 985                                                     "manifest_text": ""}).execute(num_retries=self.num_retries)
 986             self.invalidate()
 987         except apiclient_errors.Error as error:
 988             _logger.error(error)
 989             raise llfuse.FUSEError(errno.EEXIST)
 990
 991     @use_counter
 992     @check_update
 993     def rmdir(self, name):
 994         if name not in self:
 995             raise llfuse.FUSEError(errno.ENOENT)
 996         if not isinstance(self[name], CollectionDirectory):
 997             raise llfuse.FUSEError(errno.EPERM)
 998         if len(self[name]) > 0:
 999             raise llfuse.FUSEError(errno.ENOTEMPTY)
1000         with llfuse.lock_released:
1001             self.api.collections().delete(uuid=self[name].uuid()).execute(num_retries=self.num_retries)
1002         self.invalidate()
1003
1004     @use_counter
1005     @check_update
1006     def rename(self, name_old, name_new, src):
1007         if not isinstance(src, ProjectDirectory):
1008             raise llfuse.FUSEError(errno.EPERM)
1009
1010         ent = src[name_old]
1011
1012         if not isinstance(ent, CollectionDirectory):
1013             raise llfuse.FUSEError(errno.EPERM)
1014
1015         if name_new in self:
1016             # POSIX semantics for replacing one directory with another is
1017             # tricky (the target directory must be empty, the operation must be
1018             # atomic which isn't possible with the Arvados API as of this
1019             # writing) so don't support that.
1020             raise llfuse.FUSEError(errno.EPERM)
1021
1022         self.api.collections().update(uuid=ent.uuid(),
1023                                       body={"owner_uuid": self.uuid(),
1024                                             "name": name_new}).execute(num_retries=self.num_retries)
1025
1026         # Acually move the entry from source directory to this directory.
1027         del src._entries[name_old]
1028         self._entries[name_new] = ent
1029         self.inodes.invalidate_entry(src, name_old)
1030
1031     @use_counter
1032     def child_event(self, ev):
1033         properties = ev.get("properties") or {}
1034         old_attrs = properties.get("old_attributes") or {}
1035         new_attrs = properties.get("new_attributes") or {}
1036         old_attrs["uuid"] = ev["object_uuid"]
1037         new_attrs["uuid"] = ev["object_uuid"]
1038         old_name = self.sanitize_filename(self.namefn(old_attrs))
1039         new_name = self.sanitize_filename(self.namefn(new_attrs))
1040
1041         # create events will have a new name, but not an old name
1042         # delete events will have an old name, but not a new name
1043         # update events will have an old and new name, and they may be same or different
1044         # if they are the same, an unrelated field changed and there is nothing to do.
1045
1046         if old_attrs.get("owner_uuid") != self.project_uuid:
1047             # Was moved from somewhere else, so don't try to remove entry.
1048             old_name = None
1049         if ev.get("object_owner_uuid") != self.project_uuid:
1050             # Was moved to somewhere else, so don't try to add entry
1051             new_name = None
1052
1053         if old_attrs.get("is_trashed"):
1054             # Was previously deleted
1055             old_name = None
1056         if new_attrs.get("is_trashed"):
1057             # Has been deleted
1058             new_name = None
1059
1060         if new_name != old_name:
1061             ent = None
1062             if old_name in self._entries:
1063                 ent = self._entries[old_name]
1064                 del self._entries[old_name]
1065                 self.inodes.invalidate_entry(self, old_name)
1066
1067             if new_name:
1068                 if ent is not None:
1069                     self._entries[new_name] = ent
1070                 else:
1071                     self._add_entry(new_attrs, new_name)
1072             elif ent is not None:
1073                 self.inodes.del_entry(ent)
1074
1075
1076 class SharedDirectory(Directory):
1077     """A special directory that represents users or groups who have shared projects with me."""
1078
1079     def __init__(self, parent_inode, inodes, api, num_retries, exclude,
1080                  poll=False, poll_time=60):
1081         super(SharedDirectory, self).__init__(parent_inode, inodes, api.config)
1082         self.api = api
1083         self.num_retries = num_retries
1084         self.current_user = api.users().current().execute(num_retries=num_retries)
1085         self._poll = True
1086         self._poll_time = poll_time
1087         self._updating_lock = threading.Lock()
1088
1089     @use_counter
1090     def update(self):
1091         try:
1092             with llfuse.lock_released:
1093                 self._updating_lock.acquire()
1094                 if not self.stale():
1095                     return
1096
1097                 contents = {}
1098                 roots = []
1099                 root_owners = set()
1100                 objects = {}
1101
1102                 methods = self.api._rootDesc.get('resources')["groups"]['methods']
1103                 if 'httpMethod' in methods.get('shared', {}):
1104                     page = []
1105                     while True:
1106                         resp = self.api.groups().shared(filters=[['group_class', '=', 'project']]+page,
1107                                                         order="uuid",
1108                                                         limit=10000,
1109                                                         count="none",
1110                                                         include="owner_uuid").execute()
1111                         if not resp["items"]:
1112                             break
1113                         page = [["uuid", ">", resp["items"][len(resp["items"])-1]["uuid"]]]
1114                         for r in resp["items"]:
1115                             objects[r["uuid"]] = r
1116                             roots.append(r["uuid"])
1117                         for r in resp["included"]:
1118                             objects[r["uuid"]] = r
1119                             root_owners.add(r["uuid"])
1120                 else:
1121                     all_projects = arvados.util.list_all(
1122                         self.api.groups().list, self.num_retries,
1123                         filters=[['group_class','=','project']],
1124                         select=["uuid", "owner_uuid"])
1125                     for ob in all_projects:
1126                         objects[ob['uuid']] = ob
1127
1128                     current_uuid = self.current_user['uuid']
1129                     for ob in all_projects:
1130                         if ob['owner_uuid'] != current_uuid and ob['owner_uuid'] not in objects:
1131                             roots.append(ob['uuid'])
1132                             root_owners.add(ob['owner_uuid'])
1133
1134                     lusers = arvados.util.list_all(
1135                         self.api.users().list, self.num_retries,
1136                         filters=[['uuid','in', list(root_owners)]])
1137                     lgroups = arvados.util.list_all(
1138                         self.api.groups().list, self.num_retries,
1139                         filters=[['uuid','in', list(root_owners)+roots]])
1140
1141                     for l in lusers:
1142                         objects[l["uuid"]] = l
1143                     for l in lgroups:
1144                         objects[l["uuid"]] = l
1145
1146                 for r in root_owners:
1147                     if r in objects:
1148                         obr = objects[r]
1149                         if obr.get("name"):
1150                             contents[obr["name"]] = obr
1151                         #elif obr.get("username"):
1152                         #    contents[obr["username"]] = obr
1153                         elif "first_name" in obr:
1154                             contents[u"{} {}".format(obr["first_name"], obr["last_name"])] = obr
1155
1156                 for r in roots:
1157                     if r in objects:
1158                         obr = objects[r]
1159                         if obr['owner_uuid'] not in objects:
1160                             contents[obr["name"]] = obr
1161
1162             # end with llfuse.lock_released, re-acquire lock
1163
1164             self.merge(viewitems(contents),
1165                        lambda i: i[0],
1166                        lambda a, i: a.uuid() == i[1]['uuid'],
1167                        lambda i: ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, i[1], poll=self._poll, poll_time=self._poll_time))
1168         except Exception:
1169             _logger.exception("arv-mount shared dir error")
1170         finally:
1171             self._updating_lock.release()
1172
1173     def want_event_subscribe(self):
1174         return True