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
 898                 contents = arvados.util.list_all(self.api.groups().list,
 899                                                  self.num_retries,
 900                                                  filters=[["owner_uuid", "=", self.project_uuid],
 901                                                           ["group_class", "=", "project"]])
 902                 contents.extend(arvados.util.list_all(self.api.collections().list,
 903                                                       self.num_retries,
 904                                                       filters=[["owner_uuid", "=", self.project_uuid]]))
 905
 906             # end with llfuse.lock_released, re-acquire lock
 907
 908             self.merge(contents,
 909                        self.namefn,
 910                        samefn,
 911                        self.createDirectory)
 912             return True
 913         finally:
 914             self._updating_lock.release()
 915
 916     def _add_entry(self, i, name):
 917         ent = self.createDirectory(i)
 918         self._entries[name] = self.inodes.add_entry(ent)
 919         return self._entries[name]
 920
 921     @use_counter
 922     @check_update
 923     def __getitem__(self, k):
 924         if k == '.arvados#project':
 925             return self.project_object_file
 926         elif self._full_listing or super(ProjectDirectory, self).__contains__(k):
 927             return super(ProjectDirectory, self).__getitem__(k)
 928         with llfuse.lock_released:
 929             k2 = self.unsanitize_filename(k)
 930             if k2 == k:
 931                 namefilter = ["name", "=", k]
 932             else:
 933                 namefilter = ["name", "in", [k, k2]]
 934             contents = self.api.groups().list(filters=[["owner_uuid", "=", self.project_uuid],
 935                                                        ["group_class", "=", "project"],
 936                                                        namefilter],
 937                                               limit=2).execute(num_retries=self.num_retries)["items"]
 938             if not contents:
 939                 contents = self.api.collections().list(filters=[["owner_uuid", "=", self.project_uuid],
 940                                                                 namefilter],
 941                                                        limit=2).execute(num_retries=self.num_retries)["items"]
 942         if contents:
 943             if len(contents) > 1 and contents[1]['name'] == k:
 944                 # If "foo/bar" and "foo[SUBST]bar" both exist, use
 945                 # "foo[SUBST]bar".
 946                 contents = [contents[1]]
 947             name = self.sanitize_filename(self.namefn(contents[0]))
 948             if name != k:
 949                 raise KeyError(k)
 950             return self._add_entry(contents[0], name)
 951
 952         # Didn't find item
 953         raise KeyError(k)
 954
 955     def __contains__(self, k):
 956         if k == '.arvados#project':
 957             return True
 958         try:
 959             self[k]
 960             return True
 961         except KeyError:
 962             pass
 963         return False
 964
 965     @use_counter
 966     @check_update
 967     def writable(self):
 968         with llfuse.lock_released:
 969             if not self._current_user:
 970                 self._current_user = self.api.users().current().execute(num_retries=self.num_retries)
 971             return self._current_user["uuid"] in self.project_object.get("writable_by", [])
 972
 973     def persisted(self):
 974         return True
 975
 976     @use_counter
 977     @check_update
 978     def mkdir(self, name):
 979         try:
 980             with llfuse.lock_released:
 981                 self.api.collections().create(body={"owner_uuid": self.project_uuid,
 982                                                     "name": name,
 983                                                     "manifest_text": ""}).execute(num_retries=self.num_retries)
 984             self.invalidate()
 985         except apiclient_errors.Error as error:
 986             _logger.error(error)
 987             raise llfuse.FUSEError(errno.EEXIST)
 988
 989     @use_counter
 990     @check_update
 991     def rmdir(self, name):
 992         if name not in self:
 993             raise llfuse.FUSEError(errno.ENOENT)
 994         if not isinstance(self[name], CollectionDirectory):
 995             raise llfuse.FUSEError(errno.EPERM)
 996         if len(self[name]) > 0:
 997             raise llfuse.FUSEError(errno.ENOTEMPTY)
 998         with llfuse.lock_released:
 999             self.api.collections().delete(uuid=self[name].uuid()).execute(num_retries=self.num_retries)
1000         self.invalidate()
1001
1002     @use_counter
1003     @check_update
1004     def rename(self, name_old, name_new, src):
1005         if not isinstance(src, ProjectDirectory):
1006             raise llfuse.FUSEError(errno.EPERM)
1007
1008         ent = src[name_old]
1009
1010         if not isinstance(ent, CollectionDirectory):
1011             raise llfuse.FUSEError(errno.EPERM)
1012
1013         if name_new in self:
1014             # POSIX semantics for replacing one directory with another is
1015             # tricky (the target directory must be empty, the operation must be
1016             # atomic which isn't possible with the Arvados API as of this
1017             # writing) so don't support that.
1018             raise llfuse.FUSEError(errno.EPERM)
1019
1020         self.api.collections().update(uuid=ent.uuid(),
1021                                       body={"owner_uuid": self.uuid(),
1022                                             "name": name_new}).execute(num_retries=self.num_retries)
1023
1024         # Acually move the entry from source directory to this directory.
1025         del src._entries[name_old]
1026         self._entries[name_new] = ent
1027         self.inodes.invalidate_entry(src, name_old)
1028
1029     @use_counter
1030     def child_event(self, ev):
1031         properties = ev.get("properties") or {}
1032         old_attrs = properties.get("old_attributes") or {}
1033         new_attrs = properties.get("new_attributes") or {}
1034         old_attrs["uuid"] = ev["object_uuid"]
1035         new_attrs["uuid"] = ev["object_uuid"]
1036         old_name = self.sanitize_filename(self.namefn(old_attrs))
1037         new_name = self.sanitize_filename(self.namefn(new_attrs))
1038
1039         # create events will have a new name, but not an old name
1040         # delete events will have an old name, but not a new name
1041         # update events will have an old and new name, and they may be same or different
1042         # if they are the same, an unrelated field changed and there is nothing to do.
1043
1044         if old_attrs.get("owner_uuid") != self.project_uuid:
1045             # Was moved from somewhere else, so don't try to remove entry.
1046             old_name = None
1047         if ev.get("object_owner_uuid") != self.project_uuid:
1048             # Was moved to somewhere else, so don't try to add entry
1049             new_name = None
1050
1051         if old_attrs.get("is_trashed"):
1052             # Was previously deleted
1053             old_name = None
1054         if new_attrs.get("is_trashed"):
1055             # Has been deleted
1056             new_name = None
1057
1058         if new_name != old_name:
1059             ent = None
1060             if old_name in self._entries:
1061                 ent = self._entries[old_name]
1062                 del self._entries[old_name]
1063                 self.inodes.invalidate_entry(self, old_name)
1064
1065             if new_name:
1066                 if ent is not None:
1067                     self._entries[new_name] = ent
1068                 else:
1069                     self._add_entry(new_attrs, new_name)
1070             elif ent is not None:
1071                 self.inodes.del_entry(ent)
1072
1073
1074 class SharedDirectory(Directory):
1075     """A special directory that represents users or groups who have shared projects with me."""
1076
1077     def __init__(self, parent_inode, inodes, api, num_retries, exclude,
1078                  poll=False, poll_time=60):
1079         super(SharedDirectory, self).__init__(parent_inode, inodes, api.config)
1080         self.api = api
1081         self.num_retries = num_retries
1082         self.current_user = api.users().current().execute(num_retries=num_retries)
1083         self._poll = True
1084         self._poll_time = poll_time
1085         self._updating_lock = threading.Lock()
1086
1087     @use_counter
1088     def update(self):
1089         try:
1090             with llfuse.lock_released:
1091                 self._updating_lock.acquire()
1092                 if not self.stale():
1093                     return
1094
1095                 contents = {}
1096                 roots = []
1097                 root_owners = set()
1098                 objects = {}
1099
1100                 methods = self.api._rootDesc.get('resources')["groups"]['methods']
1101                 if 'httpMethod' in methods.get('shared', {}):
1102                     page = []
1103                     while True:
1104                         resp = self.api.groups().shared(filters=[['group_class', '=', 'project']]+page,
1105                                                         order="uuid",
1106                                                         limit=10000,
1107                                                         count="none",
1108                                                         include="owner_uuid").execute()
1109                         if not resp["items"]:
1110                             break
1111                         page = [["uuid", ">", resp["items"][len(resp["items"])-1]["uuid"]]]
1112                         for r in resp["items"]:
1113                             objects[r["uuid"]] = r
1114                             roots.append(r["uuid"])
1115                         for r in resp["included"]:
1116                             objects[r["uuid"]] = r
1117                             root_owners.add(r["uuid"])
1118                 else:
1119                     all_projects = arvados.util.list_all(
1120                         self.api.groups().list, self.num_retries,
1121                         filters=[['group_class','=','project']],
1122                         select=["uuid", "owner_uuid"])
1123                     for ob in all_projects:
1124                         objects[ob['uuid']] = ob
1125
1126                     current_uuid = self.current_user['uuid']
1127                     for ob in all_projects:
1128                         if ob['owner_uuid'] != current_uuid and ob['owner_uuid'] not in objects:
1129                             roots.append(ob['uuid'])
1130                             root_owners.add(ob['owner_uuid'])
1131
1132                     lusers = arvados.util.list_all(
1133                         self.api.users().list, self.num_retries,
1134                         filters=[['uuid','in', list(root_owners)]])
1135                     lgroups = arvados.util.list_all(
1136                         self.api.groups().list, self.num_retries,
1137                         filters=[['uuid','in', list(root_owners)+roots]])
1138
1139                     for l in lusers:
1140                         objects[l["uuid"]] = l
1141                     for l in lgroups:
1142                         objects[l["uuid"]] = l
1143
1144                 for r in root_owners:
1145                     if r in objects:
1146                         obr = objects[r]
1147                         if obr.get("name"):
1148                             contents[obr["name"]] = obr
1149                         #elif obr.get("username"):
1150                         #    contents[obr["username"]] = obr
1151                         elif "first_name" in obr:
1152                             contents[u"{} {}".format(obr["first_name"], obr["last_name"])] = obr
1153
1154                 for r in roots:
1155                     if r in objects:
1156                         obr = objects[r]
1157                         if obr['owner_uuid'] not in objects:
1158                             contents[obr["name"]] = obr
1159
1160             # end with llfuse.lock_released, re-acquire lock
1161
1162             self.merge(viewitems(contents),
1163                        lambda i: i[0],
1164                        lambda a, i: a.uuid() == i[1]['uuid'],
1165                        lambda i: ProjectDirectory(self.inode, self.inodes, self.api, self.num_retries, i[1], poll=self._poll, poll_time=self._poll_time))
1166         except Exception:
1167             _logger.exception("arv-mount shared dir error")
1168         finally:
1169             self._updating_lock.release()
1170
1171     def want_event_subscribe(self):
1172         return True