X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/cd1f6c0fd2f9a91b88659b90a6fd9faee576a557..2e104941dbf1e4bf92e0632cadeb946be0595d67:/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 54d9798e40..1f72635406 100755 --- a/sdk/python/arvados/commands/arv_copy.py +++ b/sdk/python/arvados/commands/arv_copy.py @@ -17,6 +17,7 @@ # arv-copy will issue an error. import argparse +import contextlib import getpass import os import re @@ -24,6 +25,7 @@ import shutil import sys import logging import tempfile +import urlparse import arvados import arvados.config @@ -32,6 +34,11 @@ import arvados.util import arvados.commands._util as arv_cmd import arvados.commands.keepdocker +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') # local_repo_dir records which git repositories from the Arvados source @@ -55,6 +62,9 @@ src_owner_uuid = None 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.') @@ -67,12 +77,15 @@ def main(): 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)') @@ -85,6 +98,13 @@ def main(): 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.') @@ -180,11 +200,19 @@ def api_for_instance(instance_name): client = arvados.api('v1', host=cfg['ARVADOS_API_HOST'], token=cfg['ARVADOS_API_TOKEN'], - insecure=api_is_insecure) + insecure=api_is_insecure, + model=OrderedJsonModel()) else: abort('need ARVADOS_API_HOST and ARVADOS_API_TOKEN for {}'.format(instance_name)) return client +# Check if git is available +def check_git_availability(): + try: + arvados.util.run_command(['git', '--help']) + except Exception: + abort('git command is not available. Please ensure git is installed.') + # copy_pipeline_instance(pi_uuid, src, dst, args) # # Copies a pipeline instance identified by pi_uuid from src to dst. @@ -209,6 +237,8 @@ def copy_pipeline_instance(pi_uuid, src, dst, args): pi = src.pipeline_instances().get(uuid=pi_uuid).execute(num_retries=args.retries) if args.recursive: + check_git_availability() + if not args.dst_git_repo: abort('--dst-git-repo is required when copying a pipeline recursively.') # Copy the pipeline template and save the copied template. @@ -245,6 +275,94 @@ def 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.iteritems(): + 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. @@ -261,7 +379,15 @@ def copy_pipeline_template(pt_uuid, src, dst, args): # 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() + if not args.dst_git_repo: abort('--dst-git-repo is required when copying a pipeline recursively.') # Copy input collections, docker images and git repos. @@ -315,10 +441,11 @@ def copy_collections(obj, src, dst, args): 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): @@ -410,12 +537,10 @@ def create_collection_from(c, src, dst, args): for link_class in ("docker_image_repo+tag", "docker_image_hash"): docker_links = src.links().list(filters=[["head_uuid", "=", collection_uuid], ["link_class", "=", link_class]]).execute(num_retries=args.retries)['items'] - for d in docker_links: - body={ - 'head_uuid': dst_collection['uuid'], - 'link_class': link_class, - 'name': d['name'], - } + for src_link in docker_links: + body = {key: src_link[key] + for key in ['link_class', 'name', 'properties']} + body['head_uuid'] = dst_collection['uuid'] body['owner_uuid'] = args.project_uuid lk = dst.links().create(body=body).execute(num_retries=args.retries) @@ -535,31 +660,30 @@ def copy_collection(obj_uuid, src, dst, 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) @@ -568,10 +692,58 @@ def copy_collection(obj_uuid, src, dst, args): # 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 = urlparse.urlsplit(url) + baseurl = urlparse.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.warn("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 @@ -589,21 +761,12 @@ def copy_collection(obj_uuid, src, dst, args): # 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)) @@ -611,17 +774,23 @@ def copy_git_repo(src_git_repo, src, dst, dst_git_repo, script_version, args): 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'