Fix ECR

backend/src/main/scala/cromwell/backend/standard/StandardAsyncExecutionActor.scala

@@ -1232,8 +1232,10 @@ trait StandardAsyncExecutionActor
     */
   def handleExecutionResult(status: StandardAsyncRunState,
                             oldHandle: StandardAsyncPendingExecutionHandle): Future[ExecutionHandle] = {
-
+
+    // get the memory retry code.
     def memoryRetryRC: Future[Boolean] = {
+      // convert int to boolean
       def returnCodeAsBoolean(codeAsOption: Option[String]): Boolean = {
         codeAsOption match {
           case Some(codeAsString) =>
@@ -1250,26 +1252,47 @@ trait StandardAsyncExecutionActor
           case None => false
         }
       }
-
+      // read if the file exists
       def readMemoryRetryRCFile(fileExists: Boolean): Future[Option[String]] = {
         if (fileExists)
           asyncIo.contentAsStringAsync(jobPaths.memoryRetryRC, None, failOnOverflow = false).map(Option(_))
         else
           Future.successful(None)
       }
-
+      //finally : assign the yielded variable
       for {
         fileExists <- asyncIo.existsAsync(jobPaths.memoryRetryRC)
         retryCheckRCAsOption <- readMemoryRetryRCFile(fileExists)
         retryWithMoreMemory = returnCodeAsBoolean(retryCheckRCAsOption)
       } yield retryWithMoreMemory
     }
 
+    // get the exit code of the job.
+    def JobExitCode: Future[String] = {
+
+      // read if the file exists
+      def readRCFile(fileExists: Boolean): Future[String] = {
+        if (fileExists)
+          asyncIo.contentAsStringAsync(jobPaths.returnCode, None, failOnOverflow = false)
+        else {
+          jobLogger.warn("RC file not found. Setting job to failed & waiting 5m before retry.")
+          Thread.sleep(300000)
+          Future("1")
+        }
+      }
+      //finally : assign the yielded variable
+      for {
+        fileExists <- asyncIo.existsAsync(jobPaths.returnCode)
+        jobRC <- readRCFile(fileExists)
+      } yield jobRC
+    }
+
+    // get path to sderr
     val stderr = jobPaths.standardPaths.error
     lazy val stderrAsOption: Option[Path] = Option(stderr)
-
+    // get the three needed variables, using functions above or direct assignment.
     val stderrSizeAndReturnCodeAndMemoryRetry = for {
-      returnCodeAsString <- asyncIo.contentAsStringAsync(jobPaths.returnCode, None, failOnOverflow = false)
+      returnCodeAsString <- JobExitCode
       // Only check stderr size if we need to, otherwise this results in a lot of unnecessary I/O that
       // may fail due to race conditions on quickly-executing jobs.
       stderrSize <- if (failOnStdErr) asyncIo.sizeAsync(stderr) else Future.successful(0L)

dockerHashing/src/main/scala/cromwell/docker/local/DockerCliFlow.scala

@@ -109,17 +109,25 @@ object DockerCliFlow {
   /** Utility for converting the flow image id to the format output by the docker cli. */
   private def cliKeyFromImageId(context: DockerInfoContext): DockerCliKey = {
     val imageId = context.dockerImageID
-    (imageId.host, imageId.repository) match {
-      case (None, None) =>
-        // For docker hub images (host == None), and don't include "library".
-        val repository = imageId.image
-        val tag = imageId.reference
-        DockerCliKey(repository, tag)
-      case _ =>
-        // For all other images, include the host and repository.
-        val repository = s"${imageId.hostAsString}${imageId.nameWithDefaultRepository}"
-        val tag = imageId.reference
-        DockerCliKey(repository, tag)
+    // private aws ECR does not have library, check for ECR in docker host.
+    if ( imageId.hostAsString.matches(raw"\d+\.dkr\.ecr\..+\.amazonaws\.com/") ) {
+       val repository = s"${imageId.hostAsString}${imageId.image}"
+       val tag = imageId.reference
+       DockerCliKey(repository, tag)
+    } else {
+      (imageId.host, imageId.repository) match {
+        case (None, None) =>
+          // For docker hub images (host == None), and don't include "library".
+          val repository = imageId.image
+          val tag = imageId.reference
+          DockerCliKey(repository, tag)
+        case _ =>
+          // For all other images, include the host and repository.
+          val repository = s"${imageId.hostAsString}${imageId.nameWithDefaultRepository}"
+          val tag = imageId.reference
+          DockerCliKey(repository, tag)
+
+       }
     }
   }
 }

dockerHashing/src/main/scala/cromwell/docker/registryv2/flows/aws/AmazonEcr.scala

@@ -4,12 +4,14 @@ import cats.effect.IO
 import cromwell.docker.{DockerImageIdentifier, DockerInfoActor, DockerRegistryConfig}
 import org.http4s.AuthScheme
 import org.http4s.client.Client
+import org.slf4j.{Logger, LoggerFactory}
 import software.amazon.awssdk.services.ecr.EcrClient
 
 import scala.compat.java8.OptionConverters._
 import scala.concurrent.Future
 
 class AmazonEcr(override val config: DockerRegistryConfig, ecrClient: EcrClient = EcrClient.create()) extends AmazonEcrAbstract(config) {
+  private val logger: Logger = LoggerFactory.getLogger(this.getClass)
 
   override protected val authorizationScheme: AuthScheme = AuthScheme.Basic
 
@@ -30,6 +32,7 @@ class AmazonEcr(override val config: DockerRegistryConfig, ecrClient: EcrClient
   override def accepts(dockerImageIdentifier: DockerImageIdentifier): Boolean = dockerImageIdentifier.hostAsString.contains("amazonaws.com")
 
   override protected def getToken(dockerInfoContext: DockerInfoActor.DockerInfoContext)(implicit client: Client[IO]): IO[Option[String]] = {
+    logger.info("obtaining access token for '{}'", dockerInfoContext.dockerImageID.fullName)
     val eventualMaybeToken = Future(ecrClient.getAuthorizationToken
       .authorizationData()
       .stream()

dockerHashing/src/main/scala/cromwell/docker/registryv2/flows/aws/AmazonEcrAbstract.scala

@@ -6,31 +6,41 @@ import cromwell.docker.registryv2.DockerRegistryV2Abstract
 import cromwell.docker.registryv2.flows.aws.EcrUtils.{EcrForbidden, EcrNotFound, EcrUnauthorized}
 import org.apache.commons.codec.digest.DigestUtils
 import org.http4s.{Header, Response, Status}
+import org.slf4j.{Logger, LoggerFactory}
 
 abstract class AmazonEcrAbstract(override val config: DockerRegistryConfig) extends DockerRegistryV2Abstract(config) {
+  private val logger: Logger = LoggerFactory.getLogger(this.getClass)
+
 
   /**
     * Not used as getToken is overridden
     */
-  override protected def authorizationServerHostName(dockerImageIdentifier: DockerImageIdentifier): String = ""
+  override protected def authorizationServerHostName(dockerImageIdentifier: DockerImageIdentifier): String = {
+    logger.warn("A call has been made to authorizationServerHostName for an ECR container but ECR auth should be done via SDK. This call will be ignored")
+    ""
+  }
 
   /**
     * Not used as getToken is overridden
     */
-  override protected def buildTokenRequestHeaders(dockerInfoContext: DockerInfoActor.DockerInfoContext): List[Header] = List.empty
+  override protected def buildTokenRequestHeaders(dockerInfoContext: DockerInfoActor.DockerInfoContext): List[Header] = {
+    logger.warn("A call has been made to buildTokenRequestHeaders for an ECR container but ECR auth should be done via SDK. This call will be ignored")
+    List.empty
+  }
 
   /**
-    * Amazon ECR repositories don't have a digest header in responses so we must made it from the manifest body
+    * Amazon ECR repositories don't have a digest header in responses so we must make it from the manifest body
     */
   override protected def getDigestFromResponse(response: Response[IO]): IO[DockerHashResult] = response match {
     case Status.Successful(r) => digestManifest(r.bodyText)
-    case Status.Unauthorized(_) => IO.raiseError(new EcrUnauthorized)
+    case Status.Unauthorized(_) => IO.raiseError(new EcrUnauthorized("Not authorized to obtain a digest from this registry. Your token may be invalid"))
     case Status.NotFound(_) => IO.raiseError(new EcrNotFound)
     case Status.Forbidden(_) => IO.raiseError(new EcrForbidden)
     case failed => failed.as[String].flatMap(body => IO.raiseError(new Exception(s"Failed to get manifest: $body")))
   }
 
   private def digestManifest(bodyText: fs2.Stream[IO, String]): IO[DockerHashResult] = {
+    logger.info("computing sha256 digest for container manifest")
     bodyText
       .compile
       .string

dockerHashing/src/main/scala/cromwell/docker/registryv2/flows/aws/AmazonEcrPublic.scala

@@ -3,13 +3,17 @@ package cromwell.docker.registryv2.flows.aws
 import cats.effect.IO
 import cromwell.docker.{DockerImageIdentifier, DockerInfoActor, DockerRegistryConfig}
 import org.http4s.client.Client
+import org.slf4j.{Logger, LoggerFactory}
 import software.amazon.awssdk.services.ecrpublic.EcrPublicClient
 import software.amazon.awssdk.services.ecrpublic.model.GetAuthorizationTokenRequest
 
 import scala.concurrent.Future
 
 
 class AmazonEcrPublic(override val config: DockerRegistryConfig, ecrClient: EcrPublicClient = EcrPublicClient.create()) extends AmazonEcrAbstract(config) {
+ private val logger: Logger = LoggerFactory.getLogger(this.getClass)
+
+
   /**
     * public.ecr.aws
     */
@@ -23,7 +27,7 @@ class AmazonEcrPublic(override val config: DockerRegistryConfig, ecrClient: EcrP
 
 
   override protected def getToken(dockerInfoContext: DockerInfoActor.DockerInfoContext)(implicit client: Client[IO]): IO[Option[String]] = {
-
+    logger.info("obtaining access token for '{}'", dockerInfoContext.dockerImageID.fullName)
     val eventualMaybeToken: Future[Option[String]] = Future(
       Option(ecrClient
         .getAuthorizationToken(GetAuthorizationTokenRequest.builder().build())

dockerHashing/src/main/scala/cromwell/docker/registryv2/flows/aws/EcrUtils.scala

@@ -2,7 +2,7 @@ package cromwell.docker.registryv2.flows.aws
 
 object EcrUtils {
 
-  case class EcrUnauthorized() extends Exception
+  case class EcrUnauthorized(msg: String) extends Exception
   case class EcrNotFound() extends Exception
   case class EcrForbidden() extends Exception
 

supportedBackends/aws/src/main/scala/cromwell/backend/impl/aws/AwsBatchJob.scala

@@ -390,43 +390,31 @@ final case class AwsBatchJob(jobDescriptor: BackendJobDescriptor, // WDL/CWL
 
 
       //check if there is already a suitable definition based on the calculated job definition name
-      val jobDefinitionName = jobDefinition.name
 
-      Log.debug(s"Checking for existence of job definition called: $jobDefinitionName")
+      Log.debug(s"Checking for existence of job definition called: ${jobDefinition.name}")
 
       val describeJobDefinitionRequest = DescribeJobDefinitionsRequest.builder()
-        .jobDefinitionName( jobDefinitionName )
+        .jobDefinitionName( jobDefinition.name )
         .status("ACTIVE")
         .build()
 
       val describeJobDefinitionResponse = batchClient.describeJobDefinitions(describeJobDefinitionRequest)
 
       if ( !describeJobDefinitionResponse.jobDefinitions.isEmpty ) {
         //sort the definitions so that the latest revision is at the head
-        val definitions = describeJobDefinitionResponse.jobDefinitions().asScala.toList.sortWith(_.revision > _.revision)
-
-        //return the arn of the job
-        definitions.head.jobDefinitionArn()
+        val existingDefinition = describeJobDefinitionResponse.jobDefinitions().asScala.toList.sortWith(_.revision > _.revision).head
+
+        //TODO test this
+        //if (existingDefinition.containerProperties().memory() != null || existingDefinition.containerProperties().vcpus() != null) {
+        //  Log.warn("the job definition '{}' has deprecated configuration for memory and vCPU and will be replaced", existingDefinition.jobDefinitionName())
+        //  registerJobDefinition(jobDefinition, jobDefinitionContext).jobDefinitionArn()
+        //} else {
+        existingDefinition.jobDefinitionArn()
+        //}
       } else {
-        Log.debug(s"No job definition found. Creating job definition: $jobDefinitionName")
-
-        // See:
-        //
-        // http://aws-java-sdk-javadoc.s3-website-us-west-2.amazonaws.com/latest/software/amazon/awssdk/services/batch/model/RegisterJobDefinitionRequest.Builder.html
-        var definitionRequest = RegisterJobDefinitionRequest.builder
-          .containerProperties(jobDefinition.containerProperties)
-          .jobDefinitionName(jobDefinitionName)
-          // See https://stackoverflow.com/questions/24349517/scala-method-named-type
-          .`type`(JobDefinitionType.CONTAINER)
-
-        if (jobDefinitionContext.runtimeAttributes.awsBatchRetryAttempts != 0){
-          definitionRequest = definitionRequest.retryStrategy(jobDefinition.retryStrategy)
-        }
-
-        Log.debug(s"Submitting definition request: $definitionRequest")
+        Log.debug(s"No job definition found. Creating job definition: ${jobDefinition.name}")
 
-        val response: RegisterJobDefinitionResponse = batchClient.registerJobDefinition(definitionRequest.build)
-        Log.info(s"Definition created: $response")
+        val response: RegisterJobDefinitionResponse = registerJobDefinition(jobDefinition, jobDefinitionContext)
         response.jobDefinitionArn()
       }
     })
@@ -455,6 +443,21 @@ final case class AwsBatchJob(jobDescriptor: BackendJobDescriptor, // WDL/CWL
     }
   }
 
+  def registerJobDefinition(jobDefinition: AwsBatchJobDefinition, jobDefinitionContext: AwsBatchJobDefinitionContext): RegisterJobDefinitionResponse = {
+    // See:
+    //
+    // http://aws-java-sdk-javadoc.s3-website-us-west-2.amazonaws.com/latest/software/amazon/awssdk/services/batch/model/RegisterJobDefinitionRequest.Builder.html
+    var definitionRequest = RegisterJobDefinitionRequest.builder
+      .containerProperties(jobDefinition.containerProperties)
+      .jobDefinitionName(jobDefinition.name)
+      // See https://stackoverflow.com/questions/24349517/scala-method-named-type
+      .`type`(JobDefinitionType.CONTAINER)
+
+    if (jobDefinitionContext.runtimeAttributes.awsBatchRetryAttempts != 0){
+      definitionRequest = definitionRequest.retryStrategy(jobDefinition.retryStrategy)
+    }
+    batchClient.registerJobDefinition(definitionRequest.build)
+  }
 
   /** Gets the status of a job by its Id, converted to a RunStatus
     *
@@ -476,10 +479,18 @@ final case class AwsBatchJob(jobDescriptor: BackendJobDescriptor, // WDL/CWL
     jobDetail
   }
 
-  def rc(detail: JobDetail): Integer = {
-    detail.container.exitCode
+  // code didn't get into the null block, so possibly not needed.
+  def rc(detail: JobDetail): Integer =  {
+    if (detail.container.exitCode == null) {
+        // if exitCode is not present, return failed ( exitCode == 127 for command not found)
+        Log.info("rc value missing. Setting to failed and sleeping for 30s...")
+        Thread.sleep(30000)
+        127
+    } else {
+        Log.info("rc value found. Setting to '{}'",detail.container.exitCode.toString())
+        detail.container.exitCode
+    }
   }
-
   def output(detail: JobDetail): String = {
     val events: Seq[OutputLogEvent] = cloudWatchLogsClient.getLogEvents(GetLogEventsRequest.builder
       // http://aws-java-sdk-javadoc.s3-website-us-west-2.amazonaws.com/latest/software/amazon/awssdk/services/batch/model/ContainerDetail.html#logStreamName--