-#! /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 os
import re
import sys
import logging
import tempfile
+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__
+
+COMMIT_HASH_RE = re.compile(r'^[0-9a-f]{1,40}$')
logger = logging.getLogger('arvados.arv-copy')
def main():
copy_opts = argparse.ArgumentParser(add_help=False)
+ copy_opts.add_argument(
+ '--version', action='version', version="%s %s" % (sys.argv[0], __version__),
+ help='Print version and exit.')
copy_opts.add_argument(
'-v', '--verbose', dest='verbose', action='store_true',
help='Verbose output.')
copy_opts.add_argument(
'-f', '--force', dest='force', action='store_true',
help='Perform copy even if the object appears to exist at the remote destination.')
+ copy_opts.add_argument(
+ '--force-filters', action='store_true', default=False,
+ help="Copy pipeline template filters verbatim, even if they act differently on the destination cluster.")
copy_opts.add_argument(
'--src', dest='source_arvados', required=True,
- help='The name of the source Arvados instance (required). May be either a pathname to a config file, or the basename of a file in $HOME/.config/arvados/instance_name.conf.')
+ 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.')
copy_opts.add_argument(
'--dst', dest='destination_arvados', required=True,
- help='The name of the destination Arvados instance (required). May be either a pathname to a config file, or the basename of a file in $HOME/.config/arvados/instance_name.conf.')
+ 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.')
copy_opts.add_argument(
'--recursive', dest='recursive', action='store_true',
help='Recursively copy any dependencies for this object. (default)')
copy_opts.add_argument(
'--project-uuid', dest='project_uuid',
help='The UUID of the project at the destination to which the pipeline should be copied.')
+ copy_opts.add_argument(
+ '--allow-git-http-src', action="store_true",
+ help='Allow cloning git repositories over insecure http')
+ copy_opts.add_argument(
+ '--allow-git-http-dst', action="store_true",
+ help='Allow pushing git repositories over insecure http')
+
copy_opts.add_argument(
'object_uuid',
help='The UUID of the object to be copied.')
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
def check_git_availability():
try:
arvados.util.run_command(['git', '--help'])
- except:
+ except Exception:
abort('git command is not available. Please ensure git is installed.')
# copy_pipeline_instance(pi_uuid, src, dst, args)
new_pi = dst.pipeline_instances().create(body=pi, ensure_unique_name=True).execute(num_retries=args.retries)
return new_pi
+def filter_iter(arg):
+ """Iterate a filter string-or-list.
+
+ Pass in a filter field that can either be a string or list.
+ This will iterate elements as if the field had been written as a list.
+ """
+ if isinstance(arg, basestring):
+ return iter((arg,))
+ else:
+ return iter(arg)
+
+def migrate_repository_filter(repo_filter, src_repository, dst_repository):
+ """Update a single repository filter in-place for the destination.
+
+ If the filter checks that the repository is src_repository, it is
+ updated to check that the repository is dst_repository. If it does
+ anything else, this function raises ValueError.
+ """
+ if src_repository is None:
+ raise ValueError("component does not specify a source repository")
+ elif dst_repository is None:
+ raise ValueError("no destination repository specified to update repository filter")
+ elif repo_filter[1:] == ['=', src_repository]:
+ repo_filter[2] = dst_repository
+ elif repo_filter[1:] == ['in', [src_repository]]:
+ repo_filter[2] = [dst_repository]
+ else:
+ raise ValueError("repository filter is not a simple source match")
+
+def migrate_script_version_filter(version_filter):
+ """Update a single script_version filter in-place for the destination.
+
+ Currently this function checks that all the filter operands are Git
+ commit hashes. If they're not, it raises ValueError to indicate that
+ the filter is not portable. It could be extended to make other
+ transformations in the future.
+ """
+ if not all(COMMIT_HASH_RE.match(v) for v in filter_iter(version_filter[2])):
+ raise ValueError("script_version filter is not limited to commit hashes")
+
+def attr_filtered(filter_, *attr_names):
+ """Return True if filter_ applies to any of attr_names, else False."""
+ return any((name == 'any') or (name in attr_names)
+ for name in filter_iter(filter_[0]))
+
+@contextlib.contextmanager
+def exception_handler(handler, *exc_types):
+ """If any exc_types are raised in the block, call handler on the exception."""
+ try:
+ yield
+ except exc_types as error:
+ handler(error)
+
+def migrate_components_filters(template_components, dst_git_repo):
+ """Update template component filters in-place for the destination.
+
+ template_components is a dictionary of components in a pipeline template.
+ This method walks over each component's filters, and updates them to have
+ identical semantics on the destination cluster. It returns a list of
+ error strings that describe what filters could not be updated safely.
+
+ dst_git_repo is the name of the destination Git repository, which can
+ be None if that is not known.
+ """
+ errors = []
+ for cname, cspec in template_components.items():
+ def add_error(errmsg):
+ errors.append("{}: {}".format(cname, errmsg))
+ if not isinstance(cspec, dict):
+ add_error("value is not a component definition")
+ continue
+ src_repository = cspec.get('repository')
+ filters = cspec.get('filters', [])
+ if not isinstance(filters, list):
+ add_error("filters are not a list")
+ continue
+ for cfilter in filters:
+ if not (isinstance(cfilter, list) and (len(cfilter) == 3)):
+ add_error("malformed filter {!r}".format(cfilter))
+ continue
+ if attr_filtered(cfilter, 'repository'):
+ with exception_handler(add_error, ValueError):
+ migrate_repository_filter(cfilter, src_repository, dst_git_repo)
+ if attr_filtered(cfilter, 'script_version'):
+ with exception_handler(add_error, ValueError):
+ migrate_script_version_filter(cfilter)
+ return errors
+
# copy_pipeline_template(pt_uuid, src, dst, args)
#
# Copies a pipeline template identified by pt_uuid from src to dst.
# fetch the pipeline template from the source instance
pt = src.pipeline_templates().get(uuid=pt_uuid).execute(num_retries=args.retries)
+ if not args.force_filters:
+ filter_errors = migrate_components_filters(pt['components'], args.dst_git_repo)
+ if filter_errors:
+ abort("Template filters cannot be copied safely. Use --force-filters to copy anyway.\n" +
+ "\n".join(filter_errors))
+
if args.recursive:
check_git_availability()
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
obj = arvados.util.portable_data_hash_pattern.sub(copy_collection_fn, obj)
obj = arvados.util.collection_uuid_pattern.sub(copy_collection_fn, obj)
return obj
- elif type(obj) == dict:
- return {v: copy_collections(obj[v], src, dst, args) for v in obj}
- elif type(obj) == list:
- return [copy_collections(v, src, dst, args) for v in obj]
+ elif isinstance(obj, dict):
+ return type(obj)((v, copy_collections(obj[v], src, dst, args))
+ for v in obj)
+ elif isinstance(obj, list):
+ return type(obj)(copy_collections(v, src, dst, args) for v in obj)
return obj
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)
else:
progress_writer = None
- for line in manifest.splitlines(True):
+ for line in manifest.splitlines():
words = line.split()
- dst_manifest_line = words[0]
+ dst_manifest += words[0]
for word in words[1:]:
try:
loc = arvados.KeepLocator(word)
- blockhash = loc.md5sum
- # copy this block if we haven't seen it before
- # (otherwise, just reuse the existing dst_locator)
- if blockhash not in dst_locators:
- logger.debug("Copying block %s (%s bytes)", blockhash, loc.size)
- if progress_writer:
- progress_writer.report(obj_uuid, bytes_written, bytes_expected)
- data = src_keep.get(word)
- dst_locator = dst_keep.put(data)
- dst_locators[blockhash] = dst_locator
- bytes_written += loc.size
- dst_manifest_line += ' ' + dst_locators[blockhash]
except ValueError:
# If 'word' can't be parsed as a locator,
# presume it's a filename.
- dst_manifest_line += ' ' + word
- dst_manifest += dst_manifest_line
- if line.endswith("\n"):
- dst_manifest += "\n"
+ dst_manifest += ' ' + word
+ continue
+ blockhash = loc.md5sum
+ # copy this block if we haven't seen it before
+ # (otherwise, just reuse the existing dst_locator)
+ if blockhash not in dst_locators:
+ logger.debug("Copying block %s (%s bytes)", blockhash, loc.size)
+ if progress_writer:
+ progress_writer.report(obj_uuid, bytes_written, bytes_expected)
+ data = src_keep.get(word)
+ dst_locator = dst_keep.put(data)
+ dst_locators[blockhash] = dst_locator
+ bytes_written += loc.size
+ dst_manifest += ' ' + dst_locators[blockhash]
+ dst_manifest += "\n"
if progress_writer:
progress_writer.report(obj_uuid, bytes_written, bytes_expected)
# Copy the manifest and save the collection.
logger.debug('saving %s with manifest: <%s>', obj_uuid, dst_manifest)
- dst_keep.put(dst_manifest.encode('utf-8'))
c['manifest_text'] = dst_manifest
return create_collection_from(c, src, dst, args)
+def select_git_url(api, repo_name, retries, allow_insecure_http, allow_insecure_http_opt):
+ r = api.repositories().list(
+ filters=[['name', '=', repo_name]]).execute(num_retries=retries)
+ if r['items_available'] != 1:
+ raise Exception('cannot identify repo {}; {} repos found'
+ .format(repo_name, r['items_available']))
+
+ https_url = [c for c in r['items'][0]["clone_urls"] if c.startswith("https:")]
+ http_url = [c for c in r['items'][0]["clone_urls"] if c.startswith("http:")]
+ other_url = [c for c in r['items'][0]["clone_urls"] if not c.startswith("http")]
+
+ priority = https_url + other_url + http_url
+
+ git_config = []
+ git_url = None
+ for url in priority:
+ if url.startswith("http"):
+ 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:
+ git_config = []
+
+ try:
+ logger.debug("trying %s", url)
+ arvados.util.run_command(["git"] + git_config + ["ls-remote", url],
+ env={"HOME": os.environ["HOME"],
+ "ARVADOS_API_TOKEN": api.api_token,
+ "GIT_ASKPASS": "/bin/false"})
+ except arvados.errors.CommandFailedError:
+ pass
+ else:
+ git_url = url
+ break
+
+ if not git_url:
+ raise Exception('Cannot access git repository, tried {}'
+ .format(priority))
+
+ if git_url.startswith("http:"):
+ if allow_insecure_http:
+ 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))
+
+ return (git_url, git_config)
+
+
# copy_git_repo(src_git_repo, src, dst, dst_git_repo, script_version, args)
#
# Copies commits from git repository 'src_git_repo' on Arvados
#
def copy_git_repo(src_git_repo, src, dst, dst_git_repo, script_version, args):
# Identify the fetch and push URLs for the git repositories.
- r = src.repositories().list(
- filters=[['name', '=', src_git_repo]]).execute(num_retries=args.retries)
- if r['items_available'] != 1:
- raise Exception('cannot identify source repo {}; {} repos found'
- .format(src_git_repo, r['items_available']))
- src_git_url = r['items'][0]['fetch_url']
- logger.debug('src_git_url: {}'.format(src_git_url))
- r = dst.repositories().list(
- filters=[['name', '=', dst_git_repo]]).execute(num_retries=args.retries)
- if r['items_available'] != 1:
- raise Exception('cannot identify destination repo {}; {} repos found'
- .format(dst_git_repo, r['items_available']))
- dst_git_push_url = r['items'][0]['push_url']
- logger.debug('dst_git_push_url: {}'.format(dst_git_push_url))
+ (src_git_url, src_git_config) = select_git_url(src, src_git_repo, args.retries, args.allow_git_http_src, "--allow-git-http-src")
+ (dst_git_url, dst_git_config) = select_git_url(dst, dst_git_repo, args.retries, args.allow_git_http_dst, "--allow-git-http-dst")
+
+ logger.debug('src_git_url: {}'.format(src_git_url))
+ logger.debug('dst_git_url: {}'.format(dst_git_url))
dst_branch = re.sub(r'\W+', '_', "{}_{}".format(src_git_url, script_version))
if src_git_repo not in local_repo_dir:
local_repo_dir[src_git_repo] = tempfile.mkdtemp()
arvados.util.run_command(
- ["git", "clone", "--bare", src_git_url,
+ ["git"] + src_git_config + ["clone", "--bare", src_git_url,
local_repo_dir[src_git_repo]],
- cwd=os.path.dirname(local_repo_dir[src_git_repo]))
+ cwd=os.path.dirname(local_repo_dir[src_git_repo]),
+ env={"HOME": os.environ["HOME"],
+ "ARVADOS_API_TOKEN": src.api_token,
+ "GIT_ASKPASS": "/bin/false"})
arvados.util.run_command(
- ["git", "remote", "add", "dst", dst_git_push_url],
+ ["git", "remote", "add", "dst", dst_git_url],
cwd=local_repo_dir[src_git_repo])
arvados.util.run_command(
["git", "branch", dst_branch, script_version],
cwd=local_repo_dir[src_git_repo])
- arvados.util.run_command(["git", "push", "dst", dst_branch],
- cwd=local_repo_dir[src_git_repo])
+ arvados.util.run_command(["git"] + dst_git_config + ["push", "dst", dst_branch],
+ cwd=local_repo_dir[src_git_repo],
+ env={"HOME": os.environ["HOME"],
+ "ARVADOS_API_TOKEN": dst.api_token,
+ "GIT_ASKPASS": "/bin/false"})
def copy_docker_images(pipeline, src, dst, args):
"""Copy any docker images named in the pipeline components'
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