Improve PR with tsibley's suggestions

* Change --aws-batch-image with --image by promoting the argument to all runners.
* Improve docker image parsing by using split_image_name from cli.utils.
* Added a new cli.errors.AWSError to represent botocore.exceptions that is now caught.
* TODO: Handle job definition creation/duplication.

nextstrain/cli/errors.py

@@ -9,3 +9,7 @@ class NextstrainCliError(Exception):
 class UserError(NextstrainCliError):
     def __init__(self, message, *args, **kwargs):
         super().__init__("Error: " + message, *args, **kwargs)
+
+
+class AWSError(NextstrainCliError):
+    """Generic exception when interfacing with AWS."""

nextstrain/cli/runner/__init__.py

@@ -107,6 +107,13 @@ def register_arguments(parser: ArgumentParser, runners: List, exec: List) -> Non
         metavar = "<prog>",
         default = exec_cmd)
 
+    # Docker image
+    development.add_argument(
+        "--image",
+        help    = "Docker name to either build or run nextstrain.",
+        metavar = "<image>",
+        default = docker.DEFAULT_IMAGE)
+
     # Static exec arguments; never modified directly by the user invocation,
     # but they won't be used if --exec is changed.
     parser.set_defaults(default_exec_cmd = exec_cmd)

nextstrain/cli/runner/aws_batch/__init__.py

@@ -85,13 +85,6 @@ def register_arguments(parser) -> None:
         type    = int,
         default = DEFAULT_MEMORY)
 
-    development.add_argument(
-        "--aws-batch-image",
-        dest    = "aws_batch_image",
-        help    = "Docker image to use for the AWS Job Definition",
-        metavar = "<name>",
-        default = docker.DEFAULT_IMAGE)
-
 
 def run(opts, argv, working_volume = None, extra_env = {}) -> int:
     local_workdir = resolve_path(working_volume.src)
@@ -137,7 +130,7 @@ def run(opts, argv, working_volume = None, extra_env = {}) -> int:
         try:
             job = jobs.submit(
                 name       = run_id,
-                image      = opts.aws_batch_image,
+                image      = opts.image,
                 queue      = opts.job_queue,
                 definition = opts.job_definition,
                 cpus       = opts.cpus,

nextstrain/cli/runner/aws_batch/jobs.py

@@ -2,12 +2,13 @@
 Job handling for AWS Batch.
 """
 
-import re
 from time import time
 from typing import Callable, Generator, Iterable, Mapping, List, Optional
 from ... import hostenv, aws
-from ...util import warn
+from ...util import warn, aws_job_definition_name
+from ...errors import AWSError
 from . import logs, s3
+import botocore.exceptions
 
 
 class JobState:
@@ -139,18 +140,6 @@ def stop(self) -> None:
     def stopped(self) -> bool:
         return self.status_reason == self.STOP_REASON
 
-def job_definition_name(definition_name, docker_image):
-    """
-    Format the AWS Batch Job Definition name according to API restriction.
-
-    Returns a string.
-    """
-    docker_image_tag = docker_image.split(':')
-    name = re.sub('[^0-9a-zA-Z-]+', '-', definition_name)
-    image = re.sub('[^0-9a-zA-Z-]+', '-', docker_image_tag[0])
-    tag = re.sub('[^0-9a-zA-Z-]+', '-', docker_image_tag[1])
-    return "%s_%s_%s" % (name, image, tag)
-
 def submit(name: str,
            image: str,
            queue: str,
@@ -167,25 +156,18 @@ def submit(name: str,
     """
     batch = aws.client_with_default_region("batch")
 
-    definition_name = job_definition_name(definition, image)
+    definition_name = aws_job_definition_name(definition, image)
     try:
-        batch.register_job_definition(
-            jobDefinitionName   = definition_name,
-            type                = "container",
-            containerProperties = {
-                "image":    image,
-                "command":  [],
-            },
-            retryStrategy = {
-                "attempts": 1,
-            },
-            timeout = {
-                "attemptDurationSeconds": 14400,
-            },
-        )
-    except Exception as error:
+        job_definition = batch.describe_job_definitions(jobDefinitionName=definition_name) \
+                                    .get("jobDefinitions")
+        #TODO: Use job definition here.
+        #   * How does this code behave if a job definition with the name already exists?
+        #   * Does it throw an error? Does it silently do nothing?
+        #   * Does it create a duplicate job definition?
+
+    except botocore.exceptions.ClientError as error:
         warn(error)
-        raise Exception("Creation of job definition (%s) failed" % definition_name)
+        raise AWSError("Creation of job definition (%s) failed" % definition_name)
 
     submission = batch.submit_job(
         jobName = name,

nextstrain/cli/runner/docker.py

@@ -36,12 +36,6 @@ def register_arguments(parser) -> None:
     development = parser.add_argument_group(
         "development options for --docker")
 
-    development.add_argument(
-        "--image",
-        help    = "Container image in which to run the pathogen build",
-        metavar = "<name>",
-        default = DEFAULT_IMAGE)
-
     development.set_defaults(volumes = [])
 
     for name in COMPONENTS:

nextstrain/cli/runner/native.py

@@ -5,7 +5,8 @@
 import os
 import shutil
 from ..types import RunnerTestResults
-from ..util import exec_or_return
+from ..util import exec_or_return, warn
+from .docker import DEFAULT_IMAGE
 
 
 def register_arguments(parser) -> None:
@@ -16,9 +17,10 @@ def register_arguments(parser) -> None:
 
 
 def run(opts, argv, working_volume = None, extra_env = {}) -> int:
+    if opts.image != DEFAULT_IMAGE:
+        warn("Docker image specified ({}), but it won't be used".format(opts.image()))
     if working_volume:
         os.chdir(str(working_volume.src))
-
     return exec_or_return(argv, extra_env)
 
 

nextstrain/cli/util.py

@@ -12,6 +12,7 @@
 from sys import exit, stderr, version_info as python_version
 from textwrap import dedent, indent
 from .__version__ import __version__
+from .runner.docker import split_image_name
 
 
 def warn(*args):
@@ -213,3 +214,16 @@ def resolve_path(path: Path) -> Path:
         return path.resolve(strict = True) # type: ignore
     else:
         return path.resolve()
+
+
+def aws_job_definition_name(definition_name: str, docker_image: str) -> str:
+    """
+    Format the AWS Batch Job Definition name according to API restriction.
+
+    Returns a string.
+    """
+    docker_image_repo, docker_image_tag = split_image_name(docker_image)
+    name = re.sub('[^0-9a-zA-Z-]+', '-', definition_name)
+    image = re.sub('[^0-9a-zA-Z-]+', '-', docker_image_repo)
+    tag = re.sub('[^0-9a-zA-Z-]+', '-', docker_image_tag)
+    return "{}_{}_{}".format(name, image, tag)