sdk/python/arvados/commands/arv_copy.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: Apache-2.0
   4
   5 # arv-copy [--recursive] [--no-recursive] object-uuid src dst
   6 #
   7 # Copies an object from Arvados instance src to instance dst.
   8 #
   9 # By default, arv-copy recursively copies any dependent objects
  10 # necessary to make the object functional in the new instance
  11 # (e.g. for a pipeline instance, arv-copy copies the pipeline
  12 # template, input collection, docker images, git repositories). If
  13 # --no-recursive is given, arv-copy copies only the single record
  14 # identified by object-uuid.
  15 #
  16 # The user must have files $HOME/.config/arvados/{src}.conf and
  17 # $HOME/.config/arvados/{dst}.conf with valid login credentials for
  18 # instances src and dst.  If either of these files is not found,
  19 # arv-copy will issue an error.
  20
  21 from __future__ import division
  22 from future import standard_library
  23 from future.utils import listvalues
  24 standard_library.install_aliases()
  25 from past.builtins import basestring
  26 from builtins import object
  27 import argparse
  28 import contextlib
  29 import getpass
  30 import os
  31 import re
  32 import shutil
  33 import sys
  34 import logging
  35 import tempfile
  36 import urllib.parse
  37
  38 import arvados
  39 import arvados.config
  40 import arvados.keep
  41 import arvados.util
  42 import arvados.commands._util as arv_cmd
  43 import arvados.commands.keepdocker
  44 import ruamel.yaml as yaml
  45
  46 from arvados.api import OrderedJsonModel
  47 from arvados._version import __version__
  48
  49 COMMIT_HASH_RE = re.compile(r'^[0-9a-f]{1,40}$')
  50
  51 logger = logging.getLogger('arvados.arv-copy')
  52
  53 # local_repo_dir records which git repositories from the Arvados source
  54 # instance have been checked out locally during this run, and to which
  55 # directories.
  56 # e.g. if repository 'twp' from src_arv has been cloned into
  57 # /tmp/gitfHkV9lu44A then local_repo_dir['twp'] = '/tmp/gitfHkV9lu44A'
  58 #
  59 local_repo_dir = {}
  60
  61 # List of collections that have been copied in this session, and their
  62 # destination collection UUIDs.
  63 collections_copied = {}
  64
  65 # Set of (repository, script_version) two-tuples of commits copied in git.
  66 scripts_copied = set()
  67
  68 # The owner_uuid of the object being copied
  69 src_owner_uuid = None
  70
  71 def main():
  72     copy_opts = argparse.ArgumentParser(add_help=False)
  73
  74     copy_opts.add_argument(
  75         '--version', action='version', version="%s %s" % (sys.argv[0], __version__),
  76         help='Print version and exit.')
  77     copy_opts.add_argument(
  78         '-v', '--verbose', dest='verbose', action='store_true',
  79         help='Verbose output.')
  80     copy_opts.add_argument(
  81         '--progress', dest='progress', action='store_true',
  82         help='Report progress on copying collections. (default)')
  83     copy_opts.add_argument(
  84         '--no-progress', dest='progress', action='store_false',
  85         help='Do not report progress on copying collections.')
  86     copy_opts.add_argument(
  87         '-f', '--force', dest='force', action='store_true',
  88         help='Perform copy even if the object appears to exist at the remote destination.')
  89     copy_opts.add_argument(
  90         '--force-filters', action='store_true', default=False,
  91         help="Copy pipeline template filters verbatim, even if they act differently on the destination cluster.")
  92     copy_opts.add_argument(
  93         '--src', dest='source_arvados', required=True,
  94         help='The name of the source Arvados instance (required) - points at an Arvados config file. May be either a pathname to a config file, or (for example) "foo" as shorthand for $HOME/.config/arvados/foo.conf.')
  95     copy_opts.add_argument(
  96         '--dst', dest='destination_arvados', required=True,
  97         help='The name of the destination Arvados instance (required) - points at an Arvados config file. May be either a pathname to a config file, or (for example) "foo" as shorthand for $HOME/.config/arvados/foo.conf.')
  98     copy_opts.add_argument(
  99         '--recursive', dest='recursive', action='store_true',
 100         help='Recursively copy any dependencies for this object. (default)')
 101     copy_opts.add_argument(
 102         '--no-recursive', dest='recursive', action='store_false',
 103         help='Do not copy any dependencies. NOTE: if this option is given, the copied object will need to be updated manually in order to be functional.')
 104     copy_opts.add_argument(
 105         '--dst-git-repo', dest='dst_git_repo',
 106         help='The name of the destination git repository. Required when copying a pipeline recursively.')
 107     copy_opts.add_argument(
 108         '--project-uuid', dest='project_uuid',
 109         help='The UUID of the project at the destination to which the pipeline should be copied.')
 110     copy_opts.add_argument(
 111         '--allow-git-http-src', action="store_true",
 112         help='Allow cloning git repositories over insecure http')
 113     copy_opts.add_argument(
 114         '--allow-git-http-dst', action="store_true",
 115         help='Allow pushing git repositories over insecure http')
 116
 117     copy_opts.add_argument(
 118         'object_uuid',
 119         help='The UUID of the object to be copied.')
 120     copy_opts.set_defaults(progress=True)
 121     copy_opts.set_defaults(recursive=True)
 122
 123     parser = argparse.ArgumentParser(
 124         description='Copy a pipeline instance, template, workflow, or collection from one Arvados instance to another.',
 125         parents=[copy_opts, arv_cmd.retry_opt])
 126     args = parser.parse_args()
 127
 128     if args.verbose:
 129         logger.setLevel(logging.DEBUG)
 130     else:
 131         logger.setLevel(logging.INFO)
 132
 133     # Create API clients for the source and destination instances
 134     src_arv = api_for_instance(args.source_arvados)
 135     dst_arv = api_for_instance(args.destination_arvados)
 136
 137     if not args.project_uuid:
 138         args.project_uuid = dst_arv.users().current().execute(num_retries=args.retries)["uuid"]
 139
 140     # Identify the kind of object we have been given, and begin copying.
 141     t = uuid_type(src_arv, args.object_uuid)
 142     if t == 'Collection':
 143         set_src_owner_uuid(src_arv.collections(), args.object_uuid, args)
 144         result = copy_collection(args.object_uuid,
 145                                  src_arv, dst_arv,
 146                                  args)
 147     elif t == 'PipelineInstance':
 148         set_src_owner_uuid(src_arv.pipeline_instances(), args.object_uuid, args)
 149         result = copy_pipeline_instance(args.object_uuid,
 150                                         src_arv, dst_arv,
 151                                         args)
 152     elif t == 'PipelineTemplate':
 153         set_src_owner_uuid(src_arv.pipeline_templates(), args.object_uuid, args)
 154         result = copy_pipeline_template(args.object_uuid,
 155                                         src_arv, dst_arv, args)
 156     elif t == 'Workflow':
 157         set_src_owner_uuid(src_arv.workflows(), args.object_uuid, args)
 158         result = copy_workflow(args.object_uuid, src_arv, dst_arv, args)
 159     else:
 160         abort("cannot copy object {} of type {}".format(args.object_uuid, t))
 161
 162     # Clean up any outstanding temp git repositories.
 163     for d in listvalues(local_repo_dir):
 164         shutil.rmtree(d, ignore_errors=True)
 165
 166     # If no exception was thrown and the response does not have an
 167     # error_token field, presume success
 168     if 'error_token' in result or 'uuid' not in result:
 169         logger.error("API server returned an error result: {}".format(result))
 170         exit(1)
 171
 172     logger.info("")
 173     logger.info("Success: created copy with uuid {}".format(result['uuid']))
 174     exit(0)
 175
 176 def set_src_owner_uuid(resource, uuid, args):
 177     global src_owner_uuid
 178     c = resource.get(uuid=uuid).execute(num_retries=args.retries)
 179     src_owner_uuid = c.get("owner_uuid")
 180
 181 # api_for_instance(instance_name)
 182 #
 183 #     Creates an API client for the Arvados instance identified by
 184 #     instance_name.
 185 #
 186 #     If instance_name contains a slash, it is presumed to be a path
 187 #     (either local or absolute) to a file with Arvados configuration
 188 #     settings.
 189 #
 190 #     Otherwise, it is presumed to be the name of a file in
 191 #     $HOME/.config/arvados/instance_name.conf
 192 #
 193 def api_for_instance(instance_name):
 194     if '/' in instance_name:
 195         config_file = instance_name
 196     else:
 197         config_file = os.path.join(os.environ['HOME'], '.config', 'arvados', "{}.conf".format(instance_name))
 198
 199     try:
 200         cfg = arvados.config.load(config_file)
 201     except (IOError, OSError) as e:
 202         abort(("Could not open config file {}: {}\n" +
 203                "You must make sure that your configuration tokens\n" +
 204                "for Arvados instance {} are in {} and that this\n" +
 205                "file is readable.").format(
 206                    config_file, e, instance_name, config_file))
 207
 208     if 'ARVADOS_API_HOST' in cfg and 'ARVADOS_API_TOKEN' in cfg:
 209         api_is_insecure = (
 210             cfg.get('ARVADOS_API_HOST_INSECURE', '').lower() in set(
 211                 ['1', 't', 'true', 'y', 'yes']))
 212         client = arvados.api('v1',
 213                              host=cfg['ARVADOS_API_HOST'],
 214                              token=cfg['ARVADOS_API_TOKEN'],
 215                              insecure=api_is_insecure,
 216                              model=OrderedJsonModel())
 217     else:
 218         abort('need ARVADOS_API_HOST and ARVADOS_API_TOKEN for {}'.format(instance_name))
 219     return client
 220
 221 # Check if git is available
 222 def check_git_availability():
 223     try:
 224         arvados.util.run_command(['git', '--help'])
 225     except Exception:
 226         abort('git command is not available. Please ensure git is installed.')
 227
 228 # copy_pipeline_instance(pi_uuid, src, dst, args)
 229 #
 230 #    Copies a pipeline instance identified by pi_uuid from src to dst.
 231 #
 232 #    If the args.recursive option is set:
 233 #      1. Copies all input collections
 234 #           * For each component in the pipeline, include all collections
 235 #             listed as job dependencies for that component)
 236 #      2. Copy docker images
 237 #      3. Copy git repositories
 238 #      4. Copy the pipeline template
 239 #
 240 #    The only changes made to the copied pipeline instance are:
 241 #      1. The original pipeline instance UUID is preserved in
 242 #         the 'properties' hash as 'copied_from_pipeline_instance_uuid'.
 243 #      2. The pipeline_template_uuid is changed to the new template uuid.
 244 #      3. The owner_uuid of the instance is changed to the user who
 245 #         copied it.
 246 #
 247 def copy_pipeline_instance(pi_uuid, src, dst, args):
 248     # Fetch the pipeline instance record.
 249     pi = src.pipeline_instances().get(uuid=pi_uuid).execute(num_retries=args.retries)
 250
 251     if args.recursive:
 252         check_git_availability()
 253
 254         if not args.dst_git_repo:
 255             abort('--dst-git-repo is required when copying a pipeline recursively.')
 256         # Copy the pipeline template and save the copied template.
 257         if pi.get('pipeline_template_uuid', None):
 258             pt = copy_pipeline_template(pi['pipeline_template_uuid'],
 259                                         src, dst, args)
 260
 261         # Copy input collections, docker images and git repos.
 262         pi = copy_collections(pi, src, dst, args)
 263         copy_git_repos(pi, src, dst, args.dst_git_repo, args)
 264         copy_docker_images(pi, src, dst, args)
 265
 266         # Update the fields of the pipeline instance with the copied
 267         # pipeline template.
 268         if pi.get('pipeline_template_uuid', None):
 269             pi['pipeline_template_uuid'] = pt['uuid']
 270
 271     else:
 272         # not recursive
 273         logger.info("Copying only pipeline instance %s.", pi_uuid)
 274         logger.info("You are responsible for making sure all pipeline dependencies have been updated.")
 275
 276     # Update the pipeline instance properties, and create the new
 277     # instance at dst.
 278     pi['properties']['copied_from_pipeline_instance_uuid'] = pi_uuid
 279     pi['description'] = "Pipeline copied from {}\n\n{}".format(
 280         pi_uuid,
 281         pi['description'] if pi.get('description', None) else '')
 282
 283     pi['owner_uuid'] = args.project_uuid
 284
 285     del pi['uuid']
 286
 287     new_pi = dst.pipeline_instances().create(body=pi, ensure_unique_name=True).execute(num_retries=args.retries)
 288     return new_pi
 289
 290 def filter_iter(arg):
 291     """Iterate a filter string-or-list.
 292
 293     Pass in a filter field that can either be a string or list.
 294     This will iterate elements as if the field had been written as a list.
 295     """
 296     if isinstance(arg, basestring):
 297         return iter((arg,))
 298     else:
 299         return iter(arg)
 300
 301 def migrate_repository_filter(repo_filter, src_repository, dst_repository):
 302     """Update a single repository filter in-place for the destination.
 303
 304     If the filter checks that the repository is src_repository, it is
 305     updated to check that the repository is dst_repository.  If it does
 306     anything else, this function raises ValueError.
 307     """
 308     if src_repository is None:
 309         raise ValueError("component does not specify a source repository")
 310     elif dst_repository is None:
 311         raise ValueError("no destination repository specified to update repository filter")
 312     elif repo_filter[1:] == ['=', src_repository]:
 313         repo_filter[2] = dst_repository
 314     elif repo_filter[1:] == ['in', [src_repository]]:
 315         repo_filter[2] = [dst_repository]
 316     else:
 317         raise ValueError("repository filter is not a simple source match")
 318
 319 def migrate_script_version_filter(version_filter):
 320     """Update a single script_version filter in-place for the destination.
 321
 322     Currently this function checks that all the filter operands are Git
 323     commit hashes.  If they're not, it raises ValueError to indicate that
 324     the filter is not portable.  It could be extended to make other
 325     transformations in the future.
 326     """
 327     if not all(COMMIT_HASH_RE.match(v) for v in filter_iter(version_filter[2])):
 328         raise ValueError("script_version filter is not limited to commit hashes")
 329
 330 def attr_filtered(filter_, *attr_names):
 331     """Return True if filter_ applies to any of attr_names, else False."""
 332     return any((name == 'any') or (name in attr_names)
 333                for name in filter_iter(filter_[0]))
 334
 335 @contextlib.contextmanager
 336 def exception_handler(handler, *exc_types):
 337     """If any exc_types are raised in the block, call handler on the exception."""
 338     try:
 339         yield
 340     except exc_types as error:
 341         handler(error)
 342
 343 def migrate_components_filters(template_components, dst_git_repo):
 344     """Update template component filters in-place for the destination.
 345
 346     template_components is a dictionary of components in a pipeline template.
 347     This method walks over each component's filters, and updates them to have
 348     identical semantics on the destination cluster.  It returns a list of
 349     error strings that describe what filters could not be updated safely.
 350
 351     dst_git_repo is the name of the destination Git repository, which can
 352     be None if that is not known.
 353     """
 354     errors = []
 355     for cname, cspec in template_components.items():
 356         def add_error(errmsg):
 357             errors.append("{}: {}".format(cname, errmsg))
 358         if not isinstance(cspec, dict):
 359             add_error("value is not a component definition")
 360             continue
 361         src_repository = cspec.get('repository')
 362         filters = cspec.get('filters', [])
 363         if not isinstance(filters, list):
 364             add_error("filters are not a list")
 365             continue
 366         for cfilter in filters:
 367             if not (isinstance(cfilter, list) and (len(cfilter) == 3)):
 368                 add_error("malformed filter {!r}".format(cfilter))
 369                 continue
 370             if attr_filtered(cfilter, 'repository'):
 371                 with exception_handler(add_error, ValueError):
 372                     migrate_repository_filter(cfilter, src_repository, dst_git_repo)
 373             if attr_filtered(cfilter, 'script_version'):
 374                 with exception_handler(add_error, ValueError):
 375                     migrate_script_version_filter(cfilter)
 376     return errors
 377
 378 # copy_pipeline_template(pt_uuid, src, dst, args)
 379 #
 380 #    Copies a pipeline template identified by pt_uuid from src to dst.
 381 #
 382 #    If args.recursive is True, also copy any collections, docker
 383 #    images and git repositories that this template references.
 384 #
 385 #    The owner_uuid of the new template is changed to that of the user
 386 #    who copied the template.
 387 #
 388 #    Returns the copied pipeline template object.
 389 #
 390 def copy_pipeline_template(pt_uuid, src, dst, args):
 391     # fetch the pipeline template from the source instance
 392     pt = src.pipeline_templates().get(uuid=pt_uuid).execute(num_retries=args.retries)
 393
 394     if not args.force_filters:
 395         filter_errors = migrate_components_filters(pt['components'], args.dst_git_repo)
 396         if filter_errors:
 397             abort("Template filters cannot be copied safely. Use --force-filters to copy anyway.\n" +
 398                   "\n".join(filter_errors))
 399
 400     if args.recursive:
 401         check_git_availability()
 402
 403         if not args.dst_git_repo:
 404             abort('--dst-git-repo is required when copying a pipeline recursively.')
 405         # Copy input collections, docker images and git repos.
 406         pt = copy_collections(pt, src, dst, args)
 407         copy_git_repos(pt, src, dst, args.dst_git_repo, args)
 408         copy_docker_images(pt, src, dst, args)
 409
 410     pt['description'] = "Pipeline template copied from {}\n\n{}".format(
 411         pt_uuid,
 412         pt['description'] if pt.get('description', None) else '')
 413     pt['name'] = "{} copied from {}".format(pt.get('name', ''), pt_uuid)
 414     del pt['uuid']
 415
 416     pt['owner_uuid'] = args.project_uuid
 417
 418     return dst.pipeline_templates().create(body=pt, ensure_unique_name=True).execute(num_retries=args.retries)
 419
 420 # copy_workflow(wf_uuid, src, dst, args)
 421 #
 422 #    Copies a workflow identified by wf_uuid from src to dst.
 423 #
 424 #    If args.recursive is True, also copy any collections
 425 #      referenced in the workflow definition yaml.
 426 #
 427 #    The owner_uuid of the new workflow is set to any given
 428 #      project_uuid or the user who copied the template.
 429 #
 430 #    Returns the copied workflow object.
 431 #
 432 def copy_workflow(wf_uuid, src, dst, args):
 433     # fetch the workflow from the source instance
 434     wf = src.workflows().get(uuid=wf_uuid).execute(num_retries=args.retries)
 435
 436     # copy collections and docker images
 437     if args.recursive:
 438         wf_def = yaml.safe_load(wf["definition"])
 439         if wf_def is not None:
 440             locations = []
 441             docker_images = {}
 442             graph = wf_def.get('$graph', None)
 443             if graph is not None:
 444                 workflow_collections(graph, locations, docker_images)
 445             else:
 446                 workflow_collections(wf_def, locations, docker_images)
 447
 448             if locations:
 449                 copy_collections(locations, src, dst, args)
 450
 451             for image in docker_images:
 452                 copy_docker_image(image, docker_images[image], src, dst, args)
 453
 454     # copy the workflow itself
 455     del wf['uuid']
 456     wf['owner_uuid'] = args.project_uuid
 457     return dst.workflows().create(body=wf).execute(num_retries=args.retries)
 458
 459 def workflow_collections(obj, locations, docker_images):
 460     if isinstance(obj, dict):
 461         loc = obj.get('location', None)
 462         if loc is not None:
 463             if loc.startswith("keep:"):
 464                 locations.append(loc[5:])
 465
 466         docker_image = obj.get('dockerImageId', None) or obj.get('dockerPull', None)
 467         if docker_image is not None:
 468             ds = docker_image.split(":", 1)
 469             tag = ds[1] if len(ds)==2 else 'latest'
 470             docker_images[ds[0]] = tag
 471
 472         for x in obj:
 473             workflow_collections(obj[x], locations, docker_images)
 474     elif isinstance(obj, list):
 475         for x in obj:
 476             workflow_collections(x, locations, docker_images)
 477
 478 # copy_collections(obj, src, dst, args)
 479 #
 480 #    Recursively copies all collections referenced by 'obj' from src
 481 #    to dst.  obj may be a dict or a list, in which case we run
 482 #    copy_collections on every value it contains. If it is a string,
 483 #    search it for any substring that matches a collection hash or uuid
 484 #    (this will find hidden references to collections like
 485 #      "input0": "$(file 3229739b505d2b878b62aed09895a55a+142/HWI-ST1027_129_D0THKACXX.1_1.fastq)")
 486 #
 487 #    Returns a copy of obj with any old collection uuids replaced by
 488 #    the new ones.
 489 #
 490 def copy_collections(obj, src, dst, args):
 491
 492     def copy_collection_fn(collection_match):
 493         """Helper function for regex substitution: copies a single collection,
 494         identified by the collection_match MatchObject, to the
 495         destination.  Returns the destination collection uuid (or the
 496         portable data hash if that's what src_id is).
 497
 498         """
 499         src_id = collection_match.group(0)
 500         if src_id not in collections_copied:
 501             dst_col = copy_collection(src_id, src, dst, args)
 502             if src_id in [dst_col['uuid'], dst_col['portable_data_hash']]:
 503                 collections_copied[src_id] = src_id
 504             else:
 505                 collections_copied[src_id] = dst_col['uuid']
 506         return collections_copied[src_id]
 507
 508     if isinstance(obj, basestring):
 509         # Copy any collections identified in this string to dst, replacing
 510         # them with the dst uuids as necessary.
 511         obj = arvados.util.portable_data_hash_pattern.sub(copy_collection_fn, obj)
 512         obj = arvados.util.collection_uuid_pattern.sub(copy_collection_fn, obj)
 513         return obj
 514     elif isinstance(obj, dict):
 515         return type(obj)((v, copy_collections(obj[v], src, dst, args))
 516                          for v in obj)
 517     elif isinstance(obj, list):
 518         return type(obj)(copy_collections(v, src, dst, args) for v in obj)
 519     return obj
 520
 521 def migrate_jobspec(jobspec, src, dst, dst_repo, args):
 522     """Copy a job's script to the destination repository, and update its record.
 523
 524     Given a jobspec dictionary, this function finds the referenced script from
 525     src and copies it to dst and dst_repo.  It also updates jobspec in place to
 526     refer to names on the destination.
 527     """
 528     repo = jobspec.get('repository')
 529     if repo is None:
 530         return
 531     # script_version is the "script_version" parameter from the source
 532     # component or job.  If no script_version was supplied in the
 533     # component or job, it is a mistake in the pipeline, but for the
 534     # purposes of copying the repository, default to "master".
 535     script_version = jobspec.get('script_version') or 'master'
 536     script_key = (repo, script_version)
 537     if script_key not in scripts_copied:
 538         copy_git_repo(repo, src, dst, dst_repo, script_version, args)
 539         scripts_copied.add(script_key)
 540     jobspec['repository'] = dst_repo
 541     repo_dir = local_repo_dir[repo]
 542     for version_key in ['script_version', 'supplied_script_version']:
 543         if version_key in jobspec:
 544             jobspec[version_key] = git_rev_parse(jobspec[version_key], repo_dir)
 545
 546 # copy_git_repos(p, src, dst, dst_repo, args)
 547 #
 548 #    Copies all git repositories referenced by pipeline instance or
 549 #    template 'p' from src to dst.
 550 #
 551 #    For each component c in the pipeline:
 552 #      * Copy git repositories named in c['repository'] and c['job']['repository'] if present
 553 #      * Rename script versions:
 554 #          * c['script_version']
 555 #          * c['job']['script_version']
 556 #          * c['job']['supplied_script_version']
 557 #        to the commit hashes they resolve to, since any symbolic
 558 #        names (tags, branches) are not preserved in the destination repo.
 559 #
 560 #    The pipeline object is updated in place with the new repository
 561 #    names.  The return value is undefined.
 562 #
 563 def copy_git_repos(p, src, dst, dst_repo, args):
 564     for component in p['components'].values():
 565         migrate_jobspec(component, src, dst, dst_repo, args)
 566         if 'job' in component:
 567             migrate_jobspec(component['job'], src, dst, dst_repo, args)
 568
 569 def total_collection_size(manifest_text):
 570     """Return the total number of bytes in this collection (excluding
 571     duplicate blocks)."""
 572
 573     total_bytes = 0
 574     locators_seen = {}
 575     for line in manifest_text.splitlines():
 576         words = line.split()
 577         for word in words[1:]:
 578             try:
 579                 loc = arvados.KeepLocator(word)
 580             except ValueError:
 581                 continue  # this word isn't a locator, skip it
 582             if loc.md5sum not in locators_seen:
 583                 locators_seen[loc.md5sum] = True
 584                 total_bytes += loc.size
 585
 586     return total_bytes
 587
 588 def create_collection_from(c, src, dst, args):
 589     """Create a new collection record on dst, and copy Docker metadata if
 590     available."""
 591
 592     collection_uuid = c['uuid']
 593     del c['uuid']
 594
 595     if not c["name"]:
 596         c['name'] = "copied from " + collection_uuid
 597
 598     if 'properties' in c:
 599         del c['properties']
 600
 601     c['owner_uuid'] = args.project_uuid
 602
 603     dst_collection = dst.collections().create(body=c, ensure_unique_name=True).execute(num_retries=args.retries)
 604
 605     # Create docker_image_repo+tag and docker_image_hash links
 606     # at the destination.
 607     for link_class in ("docker_image_repo+tag", "docker_image_hash"):
 608         docker_links = src.links().list(filters=[["head_uuid", "=", collection_uuid], ["link_class", "=", link_class]]).execute(num_retries=args.retries)['items']
 609
 610         for src_link in docker_links:
 611             body = {key: src_link[key]
 612                     for key in ['link_class', 'name', 'properties']}
 613             body['head_uuid'] = dst_collection['uuid']
 614             body['owner_uuid'] = args.project_uuid
 615
 616             lk = dst.links().create(body=body).execute(num_retries=args.retries)
 617             logger.debug('created dst link {}'.format(lk))
 618
 619     return dst_collection
 620
 621 # copy_collection(obj_uuid, src, dst, args)
 622 #
 623 #    Copies the collection identified by obj_uuid from src to dst.
 624 #    Returns the collection object created at dst.
 625 #
 626 #    If args.progress is True, produce a human-friendly progress
 627 #    report.
 628 #
 629 #    If a collection with the desired portable_data_hash already
 630 #    exists at dst, and args.force is False, copy_collection returns
 631 #    the existing collection without copying any blocks.  Otherwise
 632 #    (if no collection exists or if args.force is True)
 633 #    copy_collection copies all of the collection data blocks from src
 634 #    to dst.
 635 #
 636 #    For this application, it is critical to preserve the
 637 #    collection's manifest hash, which is not guaranteed with the
 638 #    arvados.CollectionReader and arvados.CollectionWriter classes.
 639 #    Copying each block in the collection manually, followed by
 640 #    the manifest block, ensures that the collection's manifest
 641 #    hash will not change.
 642 #
 643 def copy_collection(obj_uuid, src, dst, args):
 644     if arvados.util.keep_locator_pattern.match(obj_uuid):
 645         # If the obj_uuid is a portable data hash, it might not be uniquely
 646         # identified with a particular collection.  As a result, it is
 647         # ambigious as to what name to use for the copy.  Apply some heuristics
 648         # to pick which collection to get the name from.
 649         srccol = src.collections().list(
 650             filters=[['portable_data_hash', '=', obj_uuid]],
 651             order="created_at asc"
 652             ).execute(num_retries=args.retries)
 653
 654         items = srccol.get("items")
 655
 656         if not items:
 657             logger.warning("Could not find collection with portable data hash %s", obj_uuid)
 658             return
 659
 660         c = None
 661
 662         if len(items) == 1:
 663             # There's only one collection with the PDH, so use that.
 664             c = items[0]
 665         if not c:
 666             # See if there is a collection that's in the same project
 667             # as the root item (usually a pipeline) being copied.
 668             for i in items:
 669                 if i.get("owner_uuid") == src_owner_uuid and i.get("name"):
 670                     c = i
 671                     break
 672         if not c:
 673             # Didn't find any collections located in the same project, so
 674             # pick the oldest collection that has a name assigned to it.
 675             for i in items:
 676                 if i.get("name"):
 677                     c = i
 678                     break
 679         if not c:
 680             # None of the collections have names (?!), so just pick the
 681             # first one.
 682             c = items[0]
 683
 684         # list() doesn't return manifest text (and we don't want it to,
 685         # because we don't need the same maninfest text sent to us 50
 686         # times) so go and retrieve the collection object directly
 687         # which will include the manifest text.
 688         c = src.collections().get(uuid=c["uuid"]).execute(num_retries=args.retries)
 689     else:
 690         # Assume this is an actual collection uuid, so fetch it directly.
 691         c = src.collections().get(uuid=obj_uuid).execute(num_retries=args.retries)
 692
 693     # If a collection with this hash already exists at the
 694     # destination, and 'force' is not true, just return that
 695     # collection.
 696     if not args.force:
 697         if 'portable_data_hash' in c:
 698             colhash = c['portable_data_hash']
 699         else:
 700             colhash = c['uuid']
 701         dstcol = dst.collections().list(
 702             filters=[['portable_data_hash', '=', colhash]]
 703         ).execute(num_retries=args.retries)
 704         if dstcol['items_available'] > 0:
 705             for d in dstcol['items']:
 706                 if ((args.project_uuid == d['owner_uuid']) and
 707                     (c.get('name') == d['name']) and
 708                     (c['portable_data_hash'] == d['portable_data_hash'])):
 709                     return d
 710             c['manifest_text'] = dst.collections().get(
 711                 uuid=dstcol['items'][0]['uuid']
 712             ).execute(num_retries=args.retries)['manifest_text']
 713             return create_collection_from(c, src, dst, args)
 714
 715     # Fetch the collection's manifest.
 716     manifest = c['manifest_text']
 717     logger.debug("Copying collection %s with manifest: <%s>", obj_uuid, manifest)
 718
 719     # Copy each block from src_keep to dst_keep.
 720     # Use the newly signed locators returned from dst_keep to build
 721     # a new manifest as we go.
 722     src_keep = arvados.keep.KeepClient(api_client=src, num_retries=args.retries)
 723     dst_keep = arvados.keep.KeepClient(api_client=dst, num_retries=args.retries)
 724     dst_manifest = ""
 725     dst_locators = {}
 726     bytes_written = 0
 727     bytes_expected = total_collection_size(manifest)
 728     if args.progress:
 729         progress_writer = ProgressWriter(human_progress)
 730     else:
 731         progress_writer = None
 732
 733     for line in manifest.splitlines():
 734         words = line.split()
 735         dst_manifest += words[0]
 736         for word in words[1:]:
 737             try:
 738                 loc = arvados.KeepLocator(word)
 739             except ValueError:
 740                 # If 'word' can't be parsed as a locator,
 741                 # presume it's a filename.
 742                 dst_manifest += ' ' + word
 743                 continue
 744             blockhash = loc.md5sum
 745             # copy this block if we haven't seen it before
 746             # (otherwise, just reuse the existing dst_locator)
 747             if blockhash not in dst_locators:
 748                 logger.debug("Copying block %s (%s bytes)", blockhash, loc.size)
 749                 if progress_writer:
 750                     progress_writer.report(obj_uuid, bytes_written, bytes_expected)
 751                 data = src_keep.get(word)
 752                 dst_locator = dst_keep.put(data)
 753                 dst_locators[blockhash] = dst_locator
 754                 bytes_written += loc.size
 755             dst_manifest += ' ' + dst_locators[blockhash]
 756         dst_manifest += "\n"
 757
 758     if progress_writer:
 759         progress_writer.report(obj_uuid, bytes_written, bytes_expected)
 760         progress_writer.finish()
 761
 762     # Copy the manifest and save the collection.
 763     logger.debug('saving %s with manifest: <%s>', obj_uuid, dst_manifest)
 764
 765     c['manifest_text'] = dst_manifest
 766     return create_collection_from(c, src, dst, args)
 767
 768 def select_git_url(api, repo_name, retries, allow_insecure_http, allow_insecure_http_opt):
 769     r = api.repositories().list(
 770         filters=[['name', '=', repo_name]]).execute(num_retries=retries)
 771     if r['items_available'] != 1:
 772         raise Exception('cannot identify repo {}; {} repos found'
 773                         .format(repo_name, r['items_available']))
 774
 775     https_url = [c for c in r['items'][0]["clone_urls"] if c.startswith("https:")]
 776     http_url = [c for c in r['items'][0]["clone_urls"] if c.startswith("http:")]
 777     other_url = [c for c in r['items'][0]["clone_urls"] if not c.startswith("http")]
 778
 779     priority = https_url + other_url + http_url
 780
 781     git_config = []
 782     git_url = None
 783     for url in priority:
 784         if url.startswith("http"):
 785             u = urllib.parse.urlsplit(url)
 786             baseurl = urllib.parse.urlunsplit((u.scheme, u.netloc, "", "", ""))
 787             git_config = ["-c", "credential.%s/.username=none" % baseurl,
 788                           "-c", "credential.%s/.helper=!cred(){ cat >/dev/null; if [ \"$1\" = get ]; then echo password=$ARVADOS_API_TOKEN; fi; };cred" % baseurl]
 789         else:
 790             git_config = []
 791
 792         try:
 793             logger.debug("trying %s", url)
 794             arvados.util.run_command(["git"] + git_config + ["ls-remote", url],
 795                                       env={"HOME": os.environ["HOME"],
 796                                            "ARVADOS_API_TOKEN": api.api_token,
 797                                            "GIT_ASKPASS": "/bin/false"})
 798         except arvados.errors.CommandFailedError:
 799             pass
 800         else:
 801             git_url = url
 802             break
 803
 804     if not git_url:
 805         raise Exception('Cannot access git repository, tried {}'
 806                         .format(priority))
 807
 808     if git_url.startswith("http:"):
 809         if allow_insecure_http:
 810             logger.warning("Using insecure git url %s but will allow this because %s", git_url, allow_insecure_http_opt)
 811         else:
 812             raise Exception("Refusing to use insecure git url %s, use %s if you really want this." % (git_url, allow_insecure_http_opt))
 813
 814     return (git_url, git_config)
 815
 816
 817 # copy_git_repo(src_git_repo, src, dst, dst_git_repo, script_version, args)
 818 #
 819 #    Copies commits from git repository 'src_git_repo' on Arvados
 820 #    instance 'src' to 'dst_git_repo' on 'dst'.  Both src_git_repo
 821 #    and dst_git_repo are repository names, not UUIDs (i.e. "arvados"
 822 #    or "jsmith")
 823 #
 824 #    All commits will be copied to a destination branch named for the
 825 #    source repository URL.
 826 #
 827 #    The destination repository must already exist.
 828 #
 829 #    The user running this command must be authenticated
 830 #    to both repositories.
 831 #
 832 def copy_git_repo(src_git_repo, src, dst, dst_git_repo, script_version, args):
 833     # Identify the fetch and push URLs for the git repositories.
 834
 835     (src_git_url, src_git_config) = select_git_url(src, src_git_repo, args.retries, args.allow_git_http_src, "--allow-git-http-src")
 836     (dst_git_url, dst_git_config) = select_git_url(dst, dst_git_repo, args.retries, args.allow_git_http_dst, "--allow-git-http-dst")
 837
 838     logger.debug('src_git_url: {}'.format(src_git_url))
 839     logger.debug('dst_git_url: {}'.format(dst_git_url))
 840
 841     dst_branch = re.sub(r'\W+', '_', "{}_{}".format(src_git_url, script_version))
 842
 843     # Copy git commits from src repo to dst repo.
 844     if src_git_repo not in local_repo_dir:
 845         local_repo_dir[src_git_repo] = tempfile.mkdtemp()
 846         arvados.util.run_command(
 847             ["git"] + src_git_config + ["clone", "--bare", src_git_url,
 848              local_repo_dir[src_git_repo]],
 849             cwd=os.path.dirname(local_repo_dir[src_git_repo]),
 850             env={"HOME": os.environ["HOME"],
 851                  "ARVADOS_API_TOKEN": src.api_token,
 852                  "GIT_ASKPASS": "/bin/false"})
 853         arvados.util.run_command(
 854             ["git", "remote", "add", "dst", dst_git_url],
 855             cwd=local_repo_dir[src_git_repo])
 856     arvados.util.run_command(
 857         ["git", "branch", dst_branch, script_version],
 858         cwd=local_repo_dir[src_git_repo])
 859     arvados.util.run_command(["git"] + dst_git_config + ["push", "dst", dst_branch],
 860                              cwd=local_repo_dir[src_git_repo],
 861                              env={"HOME": os.environ["HOME"],
 862                                   "ARVADOS_API_TOKEN": dst.api_token,
 863                                   "GIT_ASKPASS": "/bin/false"})
 864
 865 def copy_docker_images(pipeline, src, dst, args):
 866     """Copy any docker images named in the pipeline components'
 867     runtime_constraints field from src to dst."""
 868
 869     logger.debug('copy_docker_images: {}'.format(pipeline['uuid']))
 870     for c_name, c_info in pipeline['components'].items():
 871         if ('runtime_constraints' in c_info and
 872             'docker_image' in c_info['runtime_constraints']):
 873             copy_docker_image(
 874                 c_info['runtime_constraints']['docker_image'],
 875                 c_info['runtime_constraints'].get('docker_image_tag', 'latest'),
 876                 src, dst, args)
 877
 878
 879 def copy_docker_image(docker_image, docker_image_tag, src, dst, args):
 880     """Copy the docker image identified by docker_image and
 881     docker_image_tag from src to dst. Create appropriate
 882     docker_image_repo+tag and docker_image_hash links at dst.
 883
 884     """
 885
 886     logger.debug('copying docker image {}:{}'.format(docker_image, docker_image_tag))
 887
 888     # Find the link identifying this docker image.
 889     docker_image_list = arvados.commands.keepdocker.list_images_in_arv(
 890         src, args.retries, docker_image, docker_image_tag)
 891     if docker_image_list:
 892         image_uuid, image_info = docker_image_list[0]
 893         logger.debug('copying collection {} {}'.format(image_uuid, image_info))
 894
 895         # Copy the collection it refers to.
 896         dst_image_col = copy_collection(image_uuid, src, dst, args)
 897     elif arvados.util.keep_locator_pattern.match(docker_image):
 898         dst_image_col = copy_collection(docker_image, src, dst, args)
 899     else:
 900         logger.warning('Could not find docker image {}:{}'.format(docker_image, docker_image_tag))
 901
 902 # git_rev_parse(rev, repo)
 903 #
 904 #    Returns the 40-character commit hash corresponding to 'rev' in
 905 #    git repository 'repo' (which must be the path of a local git
 906 #    repository)
 907 #
 908 def git_rev_parse(rev, repo):
 909     gitout, giterr = arvados.util.run_command(
 910         ['git', 'rev-parse', rev], cwd=repo)
 911     return gitout.strip()
 912
 913 # uuid_type(api, object_uuid)
 914 #
 915 #    Returns the name of the class that object_uuid belongs to, based on
 916 #    the second field of the uuid.  This function consults the api's
 917 #    schema to identify the object class.
 918 #
 919 #    It returns a string such as 'Collection', 'PipelineInstance', etc.
 920 #
 921 #    Special case: if handed a Keep locator hash, return 'Collection'.
 922 #
 923 def uuid_type(api, object_uuid):
 924     if re.match(r'^[a-f0-9]{32}\+[0-9]+(\+[A-Za-z0-9+-]+)?$', object_uuid):
 925         return 'Collection'
 926     p = object_uuid.split('-')
 927     if len(p) == 3:
 928         type_prefix = p[1]
 929         for k in api._schema.schemas:
 930             obj_class = api._schema.schemas[k].get('uuidPrefix', None)
 931             if type_prefix == obj_class:
 932                 return k
 933     return None
 934
 935 def abort(msg, code=1):
 936     logger.info("arv-copy: %s", msg)
 937     exit(code)
 938
 939
 940 # Code for reporting on the progress of a collection upload.
 941 # Stolen from arvados.commands.put.ArvPutCollectionWriter
 942 # TODO(twp): figure out how to refactor into a shared library
 943 # (may involve refactoring some arvados.commands.arv_copy.copy_collection
 944 # code)
 945
 946 def machine_progress(obj_uuid, bytes_written, bytes_expected):
 947     return "{} {}: {} {} written {} total\n".format(
 948         sys.argv[0],
 949         os.getpid(),
 950         obj_uuid,
 951         bytes_written,
 952         -1 if (bytes_expected is None) else bytes_expected)
 953
 954 def human_progress(obj_uuid, bytes_written, bytes_expected):
 955     if bytes_expected:
 956         return "\r{}: {}M / {}M {:.1%} ".format(
 957             obj_uuid,
 958             bytes_written >> 20, bytes_expected >> 20,
 959             float(bytes_written) / bytes_expected)
 960     else:
 961         return "\r{}: {} ".format(obj_uuid, bytes_written)
 962
 963 class ProgressWriter(object):
 964     _progress_func = None
 965     outfile = sys.stderr
 966
 967     def __init__(self, progress_func):
 968         self._progress_func = progress_func
 969
 970     def report(self, obj_uuid, bytes_written, bytes_expected):
 971         if self._progress_func is not None:
 972             self.outfile.write(
 973                 self._progress_func(obj_uuid, bytes_written, bytes_expected))
 974
 975     def finish(self):
 976         self.outfile.write("\n")
 977
 978 if __name__ == '__main__':
 979     main()