X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/b29cc280e5220c7de3d24c3acc687be88bbcb29b..4554374c672ee56608c9ddbd6a48486fe20c90d1:/sdk/python/arvados/commands/arv_copy.py diff --git a/sdk/python/arvados/commands/arv_copy.py b/sdk/python/arvados/commands/arv_copy.py index 1f72635406..0ba3f0a483 100755 --- a/sdk/python/arvados/commands/arv_copy.py +++ b/sdk/python/arvados/commands/arv_copy.py @@ -1,4 +1,6 @@ -#! /usr/bin/env python +# Copyright (C) The Arvados Authors. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 # arv-copy [--recursive] [--no-recursive] object-uuid src dst # @@ -16,6 +18,12 @@ # instances src and dst. If either of these files is not found, # arv-copy will issue an error. +from __future__ import division +from future import standard_library +from future.utils import listvalues +standard_library.install_aliases() +from past.builtins import basestring +from builtins import object import argparse import contextlib import getpass @@ -25,7 +33,7 @@ import shutil import sys import logging import tempfile -import urlparse +import urllib.parse import arvados import arvados.config @@ -33,6 +41,7 @@ import arvados.keep import arvados.util import arvados.commands._util as arv_cmd import arvados.commands.keepdocker +import ruamel.yaml as yaml from arvados.api import OrderedJsonModel from arvados._version import __version__ @@ -112,7 +121,7 @@ def main(): copy_opts.set_defaults(recursive=True) parser = argparse.ArgumentParser( - description='Copy a pipeline instance, template or collection from one Arvados instance to another.', + description='Copy a pipeline instance, template, workflow, or collection from one Arvados instance to another.', parents=[copy_opts, arv_cmd.retry_opt]) args = parser.parse_args() @@ -144,11 +153,14 @@ def main(): set_src_owner_uuid(src_arv.pipeline_templates(), args.object_uuid, args) result = copy_pipeline_template(args.object_uuid, src_arv, dst_arv, args) + elif t == 'Workflow': + set_src_owner_uuid(src_arv.workflows(), args.object_uuid, args) + result = copy_workflow(args.object_uuid, src_arv, dst_arv, args) else: abort("cannot copy object {} of type {}".format(args.object_uuid, t)) # Clean up any outstanding temp git repositories. - for d in local_repo_dir.values(): + for d in listvalues(local_repo_dir): shutil.rmtree(d, ignore_errors=True) # If no exception was thrown and the response does not have an @@ -340,7 +352,7 @@ def migrate_components_filters(template_components, dst_git_repo): be None if that is not known. """ errors = [] - for cname, cspec in template_components.iteritems(): + for cname, cspec in template_components.items(): def add_error(errmsg): errors.append("{}: {}".format(cname, errmsg)) if not isinstance(cspec, dict): @@ -405,6 +417,64 @@ def copy_pipeline_template(pt_uuid, src, dst, args): return dst.pipeline_templates().create(body=pt, ensure_unique_name=True).execute(num_retries=args.retries) +# copy_workflow(wf_uuid, src, dst, args) +# +# Copies a workflow identified by wf_uuid from src to dst. +# +# If args.recursive is True, also copy any collections +# referenced in the workflow definition yaml. +# +# The owner_uuid of the new workflow is set to any given +# project_uuid or the user who copied the template. +# +# Returns the copied workflow object. +# +def copy_workflow(wf_uuid, src, dst, args): + # fetch the workflow from the source instance + wf = src.workflows().get(uuid=wf_uuid).execute(num_retries=args.retries) + + # copy collections and docker images + if args.recursive: + wf_def = yaml.safe_load(wf["definition"]) + if wf_def is not None: + locations = [] + docker_images = {} + graph = wf_def.get('$graph', None) + if graph is not None: + workflow_collections(graph, locations, docker_images) + else: + workflow_collections(wf_def, locations, docker_images) + + if locations: + copy_collections(locations, src, dst, args) + + for image in docker_images: + copy_docker_image(image, docker_images[image], src, dst, args) + + # copy the workflow itself + del wf['uuid'] + wf['owner_uuid'] = args.project_uuid + return dst.workflows().create(body=wf).execute(num_retries=args.retries) + +def workflow_collections(obj, locations, docker_images): + if isinstance(obj, dict): + loc = obj.get('location', None) + if loc is not None: + if loc.startswith("keep:"): + locations.append(loc[5:]) + + docker_image = obj.get('dockerImageId', None) or obj.get('dockerPull', None) + if docker_image is not None: + ds = docker_image.split(":", 1) + tag = ds[1] if len(ds)==2 else 'latest' + docker_images[ds[0]] = tag + + for x in obj: + workflow_collections(obj[x], locations, docker_images) + elif isinstance(obj, list): + for x in obj: + workflow_collections(x, locations, docker_images) + # copy_collections(obj, src, dst, args) # # Recursively copies all collections referenced by 'obj' from src @@ -491,7 +561,7 @@ def migrate_jobspec(jobspec, src, dst, dst_repo, args): # names. The return value is undefined. # def copy_git_repos(p, src, dst, dst_repo, args): - for component in p['components'].itervalues(): + for component in p['components'].values(): migrate_jobspec(component, src, dst, dst_repo, args) if 'job' in component: migrate_jobspec(component['job'], src, dst, dst_repo, args) @@ -572,10 +642,11 @@ def create_collection_from(c, src, dst, args): # def copy_collection(obj_uuid, src, dst, args): if arvados.util.keep_locator_pattern.match(obj_uuid): - # If the obj_uuid is a portable data hash, it might not be uniquely - # identified with a particular collection. As a result, it is - # ambigious as to what name to use for the copy. Apply some heuristics - # to pick which collection to get the name from. + # If the obj_uuid is a portable data hash, it might not be + # uniquely identified with a particular collection. As a + # result, it is ambiguous as to what name to use for the copy. + # Apply some heuristics to pick which collection to get the + # name from. srccol = src.collections().list( filters=[['portable_data_hash', '=', obj_uuid]], order="created_at asc" @@ -712,8 +783,8 @@ def select_git_url(api, repo_name, retries, allow_insecure_http, allow_insecure_ git_url = None for url in priority: if url.startswith("http"): - u = urlparse.urlsplit(url) - baseurl = urlparse.urlunsplit((u.scheme, u.netloc, "", "", "")) + u = urllib.parse.urlsplit(url) + baseurl = urllib.parse.urlunsplit((u.scheme, u.netloc, "", "", "")) git_config = ["-c", "credential.%s/.username=none" % baseurl, "-c", "credential.%s/.helper=!cred(){ cat >/dev/null; if [ \"$1\" = get ]; then echo password=$ARVADOS_API_TOKEN; fi; };cred" % baseurl] else: @@ -737,7 +808,7 @@ def select_git_url(api, repo_name, retries, allow_insecure_http, allow_insecure_ if git_url.startswith("http:"): if allow_insecure_http: - logger.warn("Using insecure git url %s but will allow this because %s", git_url, allow_insecure_http_opt) + logger.warning("Using insecure git url %s but will allow this because %s", git_url, allow_insecure_http_opt) else: raise Exception("Refusing to use insecure git url %s, use %s if you really want this." % (git_url, allow_insecure_http_opt)) @@ -797,7 +868,7 @@ def copy_docker_images(pipeline, src, dst, args): runtime_constraints field from src to dst.""" logger.debug('copy_docker_images: {}'.format(pipeline['uuid'])) - for c_name, c_info in pipeline['components'].iteritems(): + for c_name, c_info in pipeline['components'].items(): if ('runtime_constraints' in c_info and 'docker_image' in c_info['runtime_constraints']): copy_docker_image(