Reland "named caches: move instead of symlinking"

client/run_isolated.py

 #!/usr/bin/env python
 # Copyright 2012 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """Runs a command with optional isolated input/output.
 
 Despite name "run_isolated", can run a generic non-isolated command specified as
 args.
 
@@ skipping 383 matching lines @@
     logging.exception('Had difficulties removing out_dir %s: %s', out_dir, e)
   stats = {
     'duration': time.time() - start,
     'items_cold': base64.b64encode(large.pack(cold)),
     'items_hot': base64.b64encode(large.pack(hot)),
   }
   return outputs_ref, success, stats
 
 
 def map_and_run(
-    command, isolated_hash, storage, isolate_cache, outputs, init_named_caches,
-    leak_temp_dir, root_dir, hard_timeout, grace_period, bot_file,
-    install_packages_fn, use_symlinks, constant_run_path):
+    command, isolated_hash, storage, isolate_cache, outputs,
+    install_named_caches, leak_temp_dir, root_dir, hard_timeout, grace_period,
+    bot_file, install_packages_fn, use_symlinks, constant_run_path):
   """Runs a command with optional isolated input/output.
 
   See run_tha_test for argument documentation.
 
   Returns metadata about the result.
   """
   assert isinstance(command, list), command
   assert root_dir or root_dir is None
   result = {
     'duration': None,
@@ skipping 80 matching lines @@
 
       # If we have an explicit list of files to return, make sure their
       # directories exist now.
       if storage and outputs:
         isolateserver.create_directories(run_dir, outputs)
 
       command = tools.fix_python_path(command)
       command = process_command(command, out_dir, bot_file)
       file_path.ensure_command_has_abs_path(command, cwd)
 
-      with init_named_caches(run_dir):
+      with install_named_caches(run_dir):
         sys.stdout.flush()
         start = time.time()
         try:
           result['exit_code'], result['had_hard_timeout'] = run_command(
               command, cwd, get_command_env(tmp_dir, cipd_info),
               hard_timeout, grace_period)
         finally:
           result['duration'] = max(time.time() - start, 0)
   except Exception as e:
     # An internal error occurred. Report accordingly so the swarming task will
@@ skipping 58 matching lines @@
         result['exit_code'] = 1
     except Exception as e:
       # Swallow any exception in the main finally clause.
       if out_dir:
         logging.exception('Leaking out_dir %s: %s', out_dir, e)
       result['internal_failure'] = str(e)
   return result
 
 
 def run_tha_test(
-    command, isolated_hash, storage, isolate_cache, outputs, init_named_caches,
-    leak_temp_dir, result_json, root_dir, hard_timeout, grace_period, bot_file,
-    install_packages_fn, use_symlinks):
+    command, isolated_hash, storage, isolate_cache, outputs,
+    install_named_caches, leak_temp_dir, result_json, root_dir, hard_timeout,
+    grace_period, bot_file, install_packages_fn, use_symlinks):
   """Runs an executable and records execution metadata.
 
   Either command or isolated_hash must be specified.
 
   If isolated_hash is specified, downloads the dependencies in the cache,
   hardlinks them into a temporary directory and runs the command specified in
   the .isolated.
 
   A temporary directory is created to hold the output files. The content inside
   this directory will be uploaded back to |storage| packaged as a .isolated
   file.
 
   Arguments:
     command: a list of string; the command to run OR optional arguments to add
              to the command stated in the .isolated file if a command was
              specified.
     isolated_hash: the SHA-1 of the .isolated file that must be retrieved to
                    recreate the tree of files to run the target executable.
                    The command specified in the .isolated is executed.
                    Mutually exclusive with command argument.
     storage: an isolateserver.Storage object to retrieve remote objects. This
              object has a reference to an isolateserver.StorageApi, which does
              the actual I/O.
     isolate_cache: an isolateserver.LocalCache to keep from retrieving the
                    same objects constantly by caching the objects retrieved.
                    Can be on-disk or in-memory.
-    init_named_caches: a function (run_dir) => context manager that creates
-                      symlinks for named caches in |run_dir|.
+    install_named_caches: a function (run_dir) => context manager that installs
+                            named caches into |run_dir|.
     leak_temp_dir: if true, the temporary directory will be deliberately leaked
                    for later examination.
     result_json: file path to dump result metadata into. If set, the process
                  exit code is always 0 unless an internal error occurred.
     root_dir: path to the directory to use to create the temporary directory. If
               not specified, a random temporary directory is created.
     hard_timeout: kills the process if it lasts more than this amount of
                   seconds.
     grace_period: number of seconds to wait between SIGTERM and SIGKILL.
     install_packages_fn: context manager dir => CipdInfo, see
@@ skipping 10 matching lines @@
       'had_hard_timeout': False,
       'internal_failure': 'Was terminated before completion',
       'outputs_ref': None,
       'version': 5,
     }
     tools.write_json(result_json, result, dense=True)
 
   # run_isolated exit code. Depends on if result_json is used or not.
   result = map_and_run(
       command, isolated_hash, storage, isolate_cache, outputs,
-      init_named_caches, leak_temp_dir, root_dir, hard_timeout, grace_period,
+      install_named_caches, leak_temp_dir, root_dir, hard_timeout, grace_period,
       bot_file, install_packages_fn, use_symlinks, True)
   logging.info('Result:\n%s', tools.format_json(result, dense=True))
 
   if result_json:
     # We've found tests to delete 'work' when quitting, causing an exception
     # here. Try to recreate the directory if necessary.
     file_path.ensure_tree(os.path.dirname(result_json))
     tools.write_json(result_json, result, dense=True)
     # Only return 1 if there was an internal error.
     return int(bool(result['internal_failure']))
@@ skipping 343 matching lines @@
   cipd.validate_cipd_options(parser, options)
 
   install_packages_fn = noop_install_packages
   if options.cipd_enabled:
     install_packages_fn = lambda run_dir: install_client_and_packages(
         run_dir, cipd.parse_package_args(options.cipd_packages),
         options.cipd_server, options.cipd_client_package,
         options.cipd_client_version, cache_dir=options.cipd_cache)
 
   @contextlib.contextmanager
-  def init_named_caches(run_dir):
+  def install_named_caches(run_dir):
     # WARNING: this function depends on "options" variable defined in the outer
     # function.
+    caches = [
+      (os.path.join(run_dir, unicode(relpath)), name)
+      for name, relpath in options.named_caches
+    ]
     with named_cache_manager.open():
-      named_cache_manager.create_symlinks(run_dir, options.named_caches)
+      for path, name in caches:
+        named_cache_manager.install(path, name)
     try:
       yield
     finally:
-      if not options.leak_temp_dir:
-        named_cache_manager.delete_symlinks(run_dir, options.named_caches)
+      with named_cache_manager.open():
+        for path, name in caches:
+          named_cache_manager.uninstall(path, name)
 
   try:
     if options.isolate_server:
       storage = isolateserver.get_storage(
           options.isolate_server, options.namespace)
       with storage:
         # Hashing schemes used by |storage| and |isolate_cache| MUST match.
         assert storage.hash_algo == isolate_cache.hash_algo
         return run_tha_test(
             args,
             options.isolated,
             storage,
             isolate_cache,
             options.output,
-            init_named_caches,
+            install_named_caches,
             options.leak_temp_dir,
             options.json, options.root_dir,
             options.hard_timeout,
             options.grace_period,
             options.bot_file,
             install_packages_fn,
             options.use_symlinks)
     return run_tha_test(
         args,
         options.isolated,
         None,
         isolate_cache,
         options.output,
-        init_named_caches,
+        install_named_caches,
         options.leak_temp_dir,
         options.json,
         options.root_dir,
         options.hard_timeout,
         options.grace_period,
         options.bot_file,
         install_packages_fn,
         options.use_symlinks)
   except (cipd.Error, named_cache.Error) as ex:
     print >> sys.stderr, ex.message
     return 1
 
 
 if __name__ == '__main__':
   subprocess42.inhibit_os_error_reporting()
   # Ensure that we are always running with the correct encoding.
   fix_encoding.fix_encoding()
   file_path.enable_symlink()
 
   sys.exit(main(sys.argv[1:]))