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