-#! /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
#
# 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
import sys
import logging
import tempfile
-import urlparse
+import urllib.parse
import arvados
import arvados.config
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__
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()
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
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):
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
# 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)
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:
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))
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(
projects / arvados.git / blobdiff