services/fuse/bin/arv-mount

   1 #!/usr/bin/env python
   2
   3 import argparse
   4 import arvados
   5 import daemon
   6 import logging
   7 import os
   8 import signal
   9 import subprocess
  10 import sys
  11 import time
  12
  13 import arvados.commands._util as arv_cmd
  14 from arvados_fuse import *
  15 from arvados.safeapi import ThreadSafeApiCache
  16 import arvados.keep
  17
  18 logger = logging.getLogger('arvados.arv-mount')
  19
  20 class Stat(object):
  21     def __init__(self, prefix, interval,
  22                  egr_name, ing_name,
  23                  egr_func, ing_func):
  24         self.prefix = prefix
  25         self.interval = interval
  26         self.egr_name = egr_name
  27         self.ing_name = ing_name
  28         self.egress = egr_func
  29         self.ingress = ing_func
  30         self.egr_prev = self.egress()
  31         self.ing_prev = self.ingress()
  32
  33     def update(self):
  34         egr = self.egress()
  35         ing = self.ingress()
  36
  37         delta = " -- interval %.4f seconds %d %s %d %s" % (self.interval,
  38                                                            egr - self.egr_prev,
  39                                                            self.egr_name,
  40                                                            ing - self.ing_prev,
  41                                                            self.ing_name)
  42
  43         sys.stderr.write("crunchstat: %s %d %s %d %s%s\n" % (self.prefix,
  44                                                              egr,
  45                                                              self.egr_name,
  46                                                              ing,
  47                                                              self.ing_name,
  48                                                              delta))
  49
  50         self.egr_prev = egr
  51         self.ing_prev = ing
  52
  53
  54 def statlogger(interval, keep, ops):
  55     calls = Stat("keepcalls", interval, "put", "get",
  56                  keep.put_counter.get,
  57                  keep.get_counter.get)
  58     net = Stat("net:keep0", interval, "tx", "rx",
  59                keep.upload_counter.get,
  60                keep.download_counter.get)
  61     cache = Stat("keepcache", interval, "hit", "miss",
  62                keep.hits_counter.get,
  63                keep.misses_counter.get)
  64     fuseops = Stat("fuseops", interval,"write", "read",
  65                    ops.write_ops_counter.get,
  66                    ops.read_ops_counter.get)
  67     blk = Stat("blkio:0:0", interval, "write", "read",
  68                ops.write_counter.get,
  69                ops.read_counter.get)
  70
  71     while True:
  72         time.sleep(interval)
  73         calls.update()
  74         net.update()
  75         cache.update()
  76         fuseops.update()
  77         blk.update()
  78
  79
  80 if __name__ == '__main__':
  81     # Handle command line parameters
  82     parser = argparse.ArgumentParser(
  83         parents=[arv_cmd.retry_opt],
  84         description='''Mount Keep data under the local filesystem.  Default mode is --home''',
  85         epilog="""
  86 Note: When using the --exec feature, you must either specify the
  87 mountpoint before --exec, or mark the end of your --exec arguments
  88 with "--".
  89 """)
  90     parser.add_argument('mountpoint', type=str, help="""Mount point.""")
  91     parser.add_argument('--allow-other', action='store_true',
  92                         help="""Let other users read the mount""")
  93
  94     mount_mode = parser.add_mutually_exclusive_group()
  95
  96     mount_mode.add_argument('--all', action='store_const', const='all', dest='mode',
  97                             help="""Mount a subdirectory for each mode: home, shared, by_tag, by_id (default if no --mount-* arguments are given).""")
  98     mount_mode.add_argument('--custom', action='store_const', const=None, dest='mode',
  99                             help="""Mount a top level meta-directory with subdirectories as specified by additional --mount-* arguments (default if any --mount-* arguments are given).""")
 100     mount_mode.add_argument('--home', action='store_const', const='home', dest='mode',
 101                             help="""Mount only the user's home project.""")
 102     mount_mode.add_argument('--shared', action='store_const', const='shared', dest='mode',
 103                             help="""Mount only list of projects shared with the user.""")
 104     mount_mode.add_argument('--by-tag', action='store_const', const='by_tag', dest='mode',
 105                             help="""Mount subdirectories listed by tag.""")
 106     mount_mode.add_argument('--by-id', action='store_const', const='by_id', dest='mode',
 107                             help="""Mount subdirectories listed by portable data hash or uuid.""")
 108     mount_mode.add_argument('--by-pdh', action='store_const', const='by_pdh', dest='mode',
 109                             help="""Mount subdirectories listed by portable data hash.""")
 110     mount_mode.add_argument('--project', type=str, metavar='UUID',
 111                             help="""Mount the specified project.""")
 112     mount_mode.add_argument('--collection', type=str, metavar='UUID_or_PDH',
 113                             help="""Mount only the specified collection.""")
 114
 115     mounts = parser.add_argument_group('Custom mount options')
 116     mounts.add_argument('--mount-by-pdh',
 117                         type=str, metavar='PATH', action='append', default=[],
 118                         help="Mount each readable collection at mountpoint/PATH/P where P is the collection's portable data hash.")
 119     mounts.add_argument('--mount-by-id',
 120                         type=str, metavar='PATH', action='append', default=[],
 121                         help="Mount each readable collection at mountpoint/PATH/UUID and mountpoint/PATH/PDH where PDH is the collection's portable data hash and UUID is its UUID.")
 122     mounts.add_argument('--mount-by-tag',
 123                         type=str, metavar='PATH', action='append', default=[],
 124                         help="Mount all collections with tag TAG at mountpoint/PATH/TAG/UUID.")
 125     mounts.add_argument('--mount-home',
 126                         type=str, metavar='PATH', action='append', default=[],
 127                         help="Mount the current user's home project at mountpoint/PATH.")
 128     mounts.add_argument('--mount-shared',
 129                         type=str, metavar='PATH', action='append', default=[],
 130                         help="Mount projects shared with the current user at mountpoint/PATH.")
 131     mounts.add_argument('--mount-tmp',
 132                         type=str, metavar='PATH', action='append', default=[],
 133                         help="Create a new collection, mount it in read/write mode at mountpoint/PATH, and delete it when unmounting.")
 134
 135     parser.add_argument('--debug', action='store_true', help="""Debug mode""")
 136     parser.add_argument('--logfile', help="""Write debug logs and errors to the specified file (default stderr).""")
 137     parser.add_argument('--foreground', action='store_true', help="""Run in foreground (default is to daemonize unless --exec specified)""", default=False)
 138     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")
 139
 140     parser.add_argument('--file-cache', type=int, help="File data cache size, in bytes (default 256MiB)", default=256*1024*1024)
 141     parser.add_argument('--directory-cache', type=int, help="Directory data cache size, in bytes (default 128MiB)", default=128*1024*1024)
 142
 143     parser.add_argument('--read-only', action='store_false', help="Mount will be read only (default)", dest="enable_write", default=False)
 144     parser.add_argument('--read-write', action='store_true', help="Mount will be read-write", dest="enable_write", default=False)
 145
 146     parser.add_argument('--crunchstat-interval', type=float, help="Write stats to stderr every N seconds (default disabled)", default=0)
 147
 148     parser.add_argument('--exec', type=str, nargs=argparse.REMAINDER,
 149                         dest="exec_args", metavar=('command', 'args', '...', '--'),
 150                         help="""Mount, run a command, then unmount and exit""")
 151
 152     args = parser.parse_args()
 153     args.mountpoint = os.path.realpath(args.mountpoint)
 154     if args.logfile:
 155         args.logfile = os.path.realpath(args.logfile)
 156
 157     # Daemonize as early as possible, so we don't accidentally close
 158     # file descriptors we're using.
 159     if not (args.exec_args or args.foreground):
 160         os.chdir(args.mountpoint)
 161         daemon_ctx = daemon.DaemonContext(working_directory='.')
 162         daemon_ctx.open()
 163     else:
 164         daemon_ctx = None
 165
 166     # Configure a log handler based on command-line switches.
 167     if args.logfile:
 168         log_handler = logging.FileHandler(args.logfile)
 169     elif daemon_ctx:
 170         log_handler = logging.NullHandler()
 171     else:
 172         log_handler = None
 173
 174     if log_handler is not None:
 175         arvados.logger.removeHandler(arvados.log_handler)
 176         arvados.logger.addHandler(log_handler)
 177
 178     if args.debug:
 179         arvados.logger.setLevel(logging.DEBUG)
 180         logger.debug("arv-mount debugging enabled")
 181
 182     logger.info("enable write is %s", args.enable_write)
 183
 184     try:
 185         api = ThreadSafeApiCache(apiconfig=arvados.config.settings(),
 186                                  keep_params={"block_cache": arvados.keep.KeepBlockCache(args.file_cache)})
 187
 188         # Create the request handler
 189         operations = Operations(os.getuid(),
 190                                 os.getgid(),
 191                                 api_client=api,
 192                                 encoding=args.encoding,
 193                                 inode_cache=InodeCache(cap=args.directory_cache),
 194                                 enable_write=args.enable_write)
 195
 196         if args.crunchstat_interval:
 197             statsthread = threading.Thread(target=statlogger, args=(args.crunchstat_interval, api.keep, operations))
 198             statsthread.daemon = True
 199             statsthread.start()
 200
 201         usr = api.users().current().execute(num_retries=args.retries)
 202         now = time.time()
 203         dir_class = None
 204         dir_args = [llfuse.ROOT_INODE, operations.inodes, api, args.retries]
 205         mount_readme = False
 206
 207         if args.mode is not None and (
 208                 args.mount_by_id or
 209                 args.mount_by_pdh or
 210                 args.mount_by_tag or
 211                 args.mount_home or
 212                 args.mount_shared or
 213                 args.mount_tmp or
 214                 args.mount_collection):
 215             sys.exit("Cannot combine '{}' mode with custom --mount-* options.".
 216                      format(args.mode))
 217
 218         if args.mode in ['by_id', 'by_pdh']:
 219             # Set up the request handler with the 'magic directory' at the root
 220             dir_class = MagicDirectory
 221             dir_args.append(args.mode == 'by_pdh')
 222         elif args.mode == 'by_tag':
 223             dir_class = TagsDirectory
 224         elif args.mode == 'shared':
 225             dir_class = SharedDirectory
 226             dir_args.append(usr)
 227         elif args.mode == 'home':
 228             dir_class = ProjectDirectory
 229             dir_args.append(usr)
 230             dir_args.append(True)
 231         elif args.mode == 'all':
 232             args.mount_by_id = ['by_id']
 233             args.mount_by_tag = ['by_tag']
 234             args.mount_home = ['home']
 235             args.mount_shared = ['shared']
 236             mount_readme = True
 237         elif args.collection is not None:
 238             # Set up the request handler with the collection at the root
 239             dir_class = CollectionDirectory
 240             dir_args.append(args.collection)
 241         elif args.project is not None:
 242             dir_class = ProjectDirectory
 243             dir_args.append(api.groups().get(uuid=args.project).execute(
 244                     num_retries=args.retries))
 245
 246         if dir_class is not None:
 247             operations.inodes.add_entry(dir_class(*dir_args))
 248         else:
 249             e = operations.inodes.add_entry(Directory(llfuse.ROOT_INODE, operations.inodes))
 250             dir_args[0] = e.inode
 251
 252             def addMount(tld, name, ent):
 253                 if name in ['', '.', '..'] or '/' in name:
 254                     sys.exit("Mount point '{}' is not supported.".format(name))
 255                 tld._entries[name] = operations.inodes.add_entry(ent)
 256
 257             for name in args.mount_by_id:
 258                 addMount(e, name, MagicDirectory(*dir_args, pdh_only=False))
 259             for name in args.mount_by_pdh:
 260                 addMount(e, name, MagicDirectory(*dir_args, pdh_only=True))
 261             for name in args.mount_by_tag:
 262                 addMount(e, name, TagsDirectory(*dir_args))
 263             for name in args.mount_home:
 264                 addMount(e, name, ProjectDirectory(*dir_args, project_object=usr, poll=True))
 265             for name in args.mount_shared:
 266                 addMount(e, name, SharedDirectory(*dir_args, exclude=usr, poll=True))
 267             for name in args.mount_tmp:
 268                 addMount(e, name, TmpCollectionDirectory(*dir_args))
 269
 270             if mount_readme:
 271                 text = '''
 272 Welcome to Arvados!  This directory provides file system access to
 273 files and objects available on the Arvados installation located at
 274 '{}' using credentials for user '{}'.
 275
 276 From here, the following directories are available:
 277
 278   by_id/     Access to Keep collections by uuid or portable data hash (see by_id/README for details).
 279   by_tag/    Access to Keep collections organized by tag.
 280   home/      The contents of your home project.
 281   shared/    Projects shared with you.
 282 '''.format(arvados.config.get('ARVADOS_API_HOST'), usr['email'])
 283                 addMount(e, StringFile(e.inode, text, now))
 284
 285     except Exception:
 286         logger.exception("arv-mount: exception during API setup")
 287         exit(1)
 288
 289     # FUSE options, see mount.fuse(8)
 290     opts = [optname for optname in ['allow_other', 'debug']
 291             if getattr(args, optname)]
 292
 293     # Increase default read/write size from 4KiB to 128KiB
 294     opts += ["big_writes", "max_read=131072"]
 295
 296     if args.exec_args:
 297         # Initialize the fuse connection
 298         llfuse.init(operations, args.mountpoint, opts)
 299
 300         # Subscribe to change events from API server
 301         if args.mode != 'by_pdh':
 302             operations.listen_for_events()
 303
 304         t = threading.Thread(None, lambda: llfuse.main())
 305         t.start()
 306
 307         # wait until the driver is finished initializing
 308         operations.initlock.wait()
 309
 310         rc = 255
 311         try:
 312             sp = subprocess.Popen(args.exec_args, shell=False)
 313
 314             # forward signals to the process.
 315             signal.signal(signal.SIGINT, lambda signum, frame: sp.send_signal(signum))
 316             signal.signal(signal.SIGTERM, lambda signum, frame: sp.send_signal(signum))
 317             signal.signal(signal.SIGQUIT, lambda signum, frame: sp.send_signal(signum))
 318
 319             # wait for process to complete.
 320             rc = sp.wait()
 321
 322             # restore default signal handlers.
 323             signal.signal(signal.SIGINT, signal.SIG_DFL)
 324             signal.signal(signal.SIGTERM, signal.SIG_DFL)
 325             signal.signal(signal.SIGQUIT, signal.SIG_DFL)
 326         except Exception as e:
 327             logger.exception('arv-mount: exception during exec %s',
 328                              args.exec_args)
 329             try:
 330                 rc = e.errno
 331             except AttributeError:
 332                 pass
 333         finally:
 334             subprocess.call(["fusermount", "-u", "-z", args.mountpoint])
 335             operations.destroy()
 336
 337         exit(rc)
 338     else:
 339         try:
 340             llfuse.init(operations, args.mountpoint, opts)
 341
 342             # Subscribe to change events from API server
 343             operations.listen_for_events()
 344
 345             llfuse.main()
 346         except Exception as e:
 347             logger.exception('arv-mount: exception during mount')
 348             exit(getattr(e, 'errno', 1))
 349         finally:
 350             operations.destroy()