Merge branch '15521-keepstore-logging'
[arvados.git] / sdk / python / arvados / commands / arv_copy.py
index 2ee97b9867036ded89244d1f11c2294efccb9ef3..0ba3f0a483fac785a6d080adf5ea494c480d02b3 100755 (executable)
@@ -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
 #
 # 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
@@ -24,7 +33,7 @@ import shutil
 import sys
 import logging
 import tempfile
-import urlparse
+import urllib.parse
 
 import arvados
 import arvados.config
@@ -32,8 +41,12 @@ 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__
+
+COMMIT_HASH_RE = re.compile(r'^[0-9a-f]{1,40}$')
 
 logger = logging.getLogger('arvados.arv-copy')
 
@@ -58,6 +71,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.')
@@ -70,6 +86,9 @@ 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) - 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.')
@@ -102,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()
 
@@ -134,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
@@ -265,6 +287,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.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.
@@ -281,6 +391,12 @@ 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()
 
@@ -301,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
@@ -387,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)
@@ -468,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"
@@ -608,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:
@@ -633,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))
 
@@ -693,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(