import subprocess
import time
+import arvados.commands._util as arv_cmd
from arvados_fuse import *
+from arvados.safeapi import ThreadSafeApiCache
+import arvados.keep
logger = logging.getLogger('arvados.arv-mount')
+class Stat(object):
+ def __init__(self, prefix, interval,
+ egr_name, ing_name,
+ egr_func, ing_func):
+ self.prefix = prefix
+ self.interval = interval
+ self.egr_name = egr_name
+ self.ing_name = ing_name
+ self.egress = egr_func
+ self.ingress = ing_func
+ self.egr = self.egress()
+ self.ing = self.ingress()
+
+ def update(self):
+ self.egr_prev = self.egr
+ self.ing_prev = self.ing
+ self.egr = self.egress()
+ self.ing = self.ingress()
+
+ delta = " -- interval %.4f seconds %d %s %d %s" % (self.interval,
+ self.egr-self.egr_prev,
+ self.egr_name,
+ self.ing-self.ing_prev,
+ self.ing_name)
+
+ sys.stderr.write("crunchstat: %s %d %s %d %s%s\n" % (self.prefix,
+ self.egr,
+ self.egr_name,
+ self.ing,
+ self.ing_name,
+ delta))
+
+def statlogger(keep, ops):
+ interval = 10
+ calls = Stat("keepcalls", interval, "put", "get",
+ keep.put_counter.get,
+ keep.get_counter.get)
+ net = Stat("net:keep0", interval, "tx", "rx",
+ keep.upload_counter.get,
+ keep.download_counter.get)
+ fuseops = Stat("fuseops", interval,"write", "read",
+ ops.write_ops_counter.get,
+ ops.read_ops_counter.get)
+ blk = Stat("blkio:0:0", interval, "write", "read",
+ ops.write_counter.get,
+ ops.read_counter.get)
+
+ while True:
+ time.sleep(interval)
+ calls.update()
+ net.update()
+ fuseops.update()
+ blk.update()
+
+
if __name__ == '__main__':
# Handle command line parameters
parser = argparse.ArgumentParser(
+ parents=[arv_cmd.retry_opt],
description='''Mount Keep data under the local filesystem. Default mode is --home''',
epilog="""
Note: When using the --exec feature, you must either specify the
help="""Mount subdirectories listed by tag.""")
mount_mode.add_argument('--by-id', action='store_true',
help="""Mount subdirectories listed by portable data hash or uuid.""")
+ mount_mode.add_argument('--by-pdh', action='store_true',
+ help="""Mount subdirectories listed by portable data hash.""")
mount_mode.add_argument('--project', type=str, help="""Mount a specific project.""")
mount_mode.add_argument('--collection', type=str, help="""Mount only the specified collection.""")
parser.add_argument('--debug', action='store_true', help="""Debug mode""")
parser.add_argument('--logfile', help="""Write debug logs and errors to the specified file (default stderr).""")
parser.add_argument('--foreground', action='store_true', help="""Run in foreground (default is to daemonize unless --exec specified)""", default=False)
+ parser.add_argument('--encoding', type=str, help="Character encoding to use for filesystem, default is utf-8 (see Python codec registry for list of available encodings)", default="utf-8")
+
+ parser.add_argument('--file-cache', type=int, help="File data cache size, in bytes (default 256MiB)", default=256*1024*1024)
+ parser.add_argument('--directory-cache', type=int, help="Directory data cache size, in bytes (default 128MiB)", default=128*1024*1024)
+
+ parser.add_argument('--read-only', action='store_false', help="Mount will be read only (default)", dest="enable_write", default=False)
+ parser.add_argument('--read-write', action='store_true', help="Mount will be read-write", dest="enable_write", default=False)
+
+ parser.add_argument('--stats', action='store_true', help="Write stats to stderr", default=False)
+
parser.add_argument('--exec', type=str, nargs=argparse.REMAINDER,
dest="exec_args", metavar=('command', 'args', '...', '--'),
help="""Mount, run a command, then unmount and exit""")
else:
daemon_ctx = None
- # Configure a logger based on command-line switches.
- # If we're using a contemporary Python SDK (mid-August 2014),
- # configure the arvados hierarchy logger.
- # Otherwise, configure the program root logger.
- base_logger = getattr(arvados, 'logger', None)
-
+ # Configure a log handler based on command-line switches.
if args.logfile:
log_handler = logging.FileHandler(args.logfile)
elif daemon_ctx:
log_handler = logging.NullHandler()
- elif base_logger:
- log_handler = arvados.log_handler
else:
- log_handler = logging.StreamHandler()
+ log_handler = None
- if base_logger is None:
- base_logger = logging.getLogger()
- else:
- base_logger.removeHandler(arvados.log_handler)
- base_logger.addHandler(log_handler)
+ if log_handler is not None:
+ arvados.logger.removeHandler(arvados.log_handler)
+ arvados.logger.addHandler(log_handler)
if args.debug:
- base_logger.setLevel(logging.DEBUG)
+ arvados.logger.setLevel(logging.DEBUG)
logger.debug("arv-mount debugging enabled")
+ logger.info("enable write is %s", args.enable_write)
+
try:
# Create the request handler
- operations = Operations(os.getuid(), os.getgid())
- api = SafeApi(arvados.config)
+ operations = Operations(os.getuid(),
+ os.getgid(),
+ encoding=args.encoding,
+ inode_cache=InodeCache(cap=args.directory_cache),
+ enable_write=args.enable_write)
+ api = ThreadSafeApiCache(apiconfig=arvados.config.settings(),
+ keep_params={"block_cache": arvados.keep.KeepBlockCache(args.file_cache)})
+
+ if args.stats:
+ statsthread = threading.Thread(target=statlogger, args=(api.keep, operations))
+ statsthread.daemon = True
+ statsthread.start()
- usr = api.users().current().execute()
+ usr = api.users().current().execute(num_retries=args.retries)
now = time.time()
- if args.by_id:
+ dir_class = None
+ dir_args = [llfuse.ROOT_INODE, operations.inodes, api, args.retries]
+ if args.by_id or args.by_pdh:
# Set up the request handler with the 'magic directory' at the root
- operations.inodes.add_entry(MagicDirectory(llfuse.ROOT_INODE, operations.inodes, api))
+ dir_class = MagicDirectory
+ dir_args.append(args.by_pdh)
elif args.by_tag:
- operations.inodes.add_entry(TagsDirectory(llfuse.ROOT_INODE, operations.inodes, api))
+ dir_class = TagsDirectory
elif args.shared:
- operations.inodes.add_entry(SharedDirectory(llfuse.ROOT_INODE, operations.inodes, api, usr))
+ dir_class = SharedDirectory
+ dir_args.append(usr)
elif args.home:
- operations.inodes.add_entry(ProjectDirectory(llfuse.ROOT_INODE, operations.inodes, api, usr))
- elif args.collection != None:
+ dir_class = ProjectDirectory
+ dir_args.append(usr)
+ dir_args.append(True)
+ elif args.collection is not None:
# Set up the request handler with the collection at the root
- operations.inodes.add_entry(CollectionDirectory(llfuse.ROOT_INODE, operations.inodes, api, args.collection))
- elif args.project != None:
- operations.inodes.add_entry(ProjectDirectory(llfuse.ROOT_INODE, operations.inodes, api, api.groups().get(uuid=args.project).execute()))
+ dir_class = CollectionDirectory
+ dir_args.append(args.collection)
+ elif args.project is not None:
+ dir_class = ProjectDirectory
+ dir_args.append(api.groups().get(uuid=args.project).execute(
+ num_retries=args.retries))
+
+ if dir_class is not None:
+ operations.inodes.add_entry(dir_class(*dir_args))
else:
- e = operations.inodes.add_entry(Directory(llfuse.ROOT_INODE))
+ e = operations.inodes.add_entry(Directory(llfuse.ROOT_INODE, operations.inodes))
+ dir_args[0] = e.inode
+
+ e._entries['by_id'] = operations.inodes.add_entry(MagicDirectory(*dir_args))
+
+ e._entries['by_tag'] = operations.inodes.add_entry(TagsDirectory(*dir_args))
- e._entries['home'] = operations.inodes.add_entry(ProjectDirectory(e.inode, operations.inodes, api, usr))
- e._entries['shared'] = operations.inodes.add_entry(SharedDirectory(e.inode, operations.inodes, api, usr))
- e._entries['by_tag'] = operations.inodes.add_entry(TagsDirectory(e.inode, operations.inodes, api))
- e._entries['by_id'] = operations.inodes.add_entry(MagicDirectory(e.inode, operations.inodes, api))
+ dir_args.append(usr)
+ dir_args.append(True)
+ e._entries['home'] = operations.inodes.add_entry(ProjectDirectory(*dir_args))
+ e._entries['shared'] = operations.inodes.add_entry(SharedDirectory(*dir_args))
text = '''
-Welcome to Arvados! This directory provides file system access to files and objects
-available on the Arvados installation located at '{}'
+Welcome to Arvados! This directory provides file system access to files and objects
+available on the Arvados installation located at '{}'
using credentials for user '{}'.
From here, the following directories are available:
opts = [optname for optname in ['allow_other', 'debug']
if getattr(args, optname)]
+ # Increase default read/write size from 4KiB to 128KiB
+ opts += ["big_writes", "max_read=131072"]
+
if args.exec_args:
# Initialize the fuse connection
llfuse.init(operations, args.mountpoint, opts)
+ # Subscribe to change events from API server
+ if not args.by_pdh:
+ operations.listen_for_events(api)
+
t = threading.Thread(None, lambda: llfuse.main())
t.start()
pass
finally:
subprocess.call(["fusermount", "-u", "-z", args.mountpoint])
+ operations.destroy()
exit(rc)
else:
try:
llfuse.init(operations, args.mountpoint, opts)
+
+ # Subscribe to change events from API server
+ operations.listen_for_events(api)
+
llfuse.main()
except Exception as e:
logger.exception('arv-mount: exception during mount')
exit(getattr(e, 'errno', 1))
+ finally:
+ operations.destroy()
projects / arvados.git / blobdiff