AN-381 Enable GCR mirroring of Dockerhub (#7740)

Co-authored-by: Adam Nichols <aednichols@gmail.com>

Author: jgainerdewar

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Cromwell Change Log
 
+## 90 Release Notes
+
+### GCP Batch
+ * Cromwell now supports automatic use of the [GAR Dockerhub mirror](https://cloud.google.com/artifact-registry/docs/pull-cached-dockerhub-images), see [ReadTheDocs](https://cromwell.readthedocs.io/en/develop/backends/GCPBatch/) for details.
+
 ## 89 Release Notes
 
 ### Improvements

backend/src/main/scala/cromwell/backend/BackendLifecycleActorFactory.scala

@@ -8,6 +8,7 @@ import cromwell.core.CallOutputs
 import cromwell.core.JobToken.JobTokenType
 import cromwell.core.path.Path
 import cromwell.core.path.PathFactory.PathBuilders
+import cromwell.docker.DockerMirroring
 import net.ceedubs.ficus.Ficus._
 import wom.expression.{IoFunctionSet, NoIoFunctionSet}
 import wom.graph.CommandCallNode
@@ -164,6 +165,11 @@ trait BackendLifecycleActorFactory extends PlatformSpecific {
                             initializationDataOption: Option[BackendInitializationData]
   ): List[Any] = List.empty
 
+  /**
+   * Returns a DockerMirror built based on backend configuration
+   */
+  val dockerMirroring: Option[DockerMirroring] = None
+
   /**
     * Allows Cromwell to self-identify which cloud it's running on for runtime attribute purposes
     */

centaur/src/main/resources/standardTestCases/hello_dockermirror_gcpbatch.test

@@ -0,0 +1,16 @@
+name: hello_dockermirror_gcpbatch
+testFormat: workflowsuccess
+# needs an alt if we're going to keep Docker image cache tests which I'm not sure we are
+backends: [GCPBATCHDockerMirror]
+
+files {
+  workflow: hello_dockermirror_gcpbatch/hello.wdl
+  inputs: hello_dockermirror_gcpbatch/hello.inputs
+}
+
+metadata {
+  workflowName: wf_hello
+  status: Succeeded
+  "calls.wf_hello.hello.runtimeAttributes.docker": "mirror.gcr.io/ubuntu@sha256:71cd81252a3563a03ad8daee81047b62ab5d892ebbfbf71cf53415f29c130950"
+  "calls.wf_hello.hello.dockerImageUsed": "mirror.gcr.io/ubuntu@sha256:71cd81252a3563a03ad8daee81047b62ab5d892ebbfbf71cf53415f29c130950"
+}

centaur/src/main/resources/standardTestCases/hello_dockermirror_gcpbatch/hello.inputs

@@ -0,0 +1,4 @@
+{
+  "wf_hello.hello.addressee": "m'Lord"
+}
+

centaur/src/main/resources/standardTestCases/hello_dockermirror_gcpbatch/hello.wdl

@@ -0,0 +1,20 @@
+task hello {
+  String addressee
+  command {
+    echo "Hello ${addressee}!"
+  }
+  output {
+    String salutation = read_string(stdout())
+  }
+  runtime {
+    backend: "GCPBATCHDockerMirror"
+    docker: "ubuntu@sha256:71cd81252a3563a03ad8daee81047b62ab5d892ebbfbf71cf53415f29c130950"
+  }
+}
+
+workflow wf_hello {
+  call hello
+  output {
+     hello.salutation
+  }
+}

centaur/src/main/resources/standardTestCases/hello_dockermirror_papiv2.test

@@ -0,0 +1,16 @@
+name: hello_dockermirror_papiv2
+testFormat: workflowsuccess
+# needs an alt if we're going to keep Docker image cache tests which I'm not sure we are
+backends: [Papiv2DockerMirror]
+
+files {
+  workflow: hello_dockermirror_papiv2/hello.wdl
+  inputs: hello_dockermirror_papiv2/hello.inputs
+}
+
+metadata {
+  workflowName: wf_hello
+  status: Succeeded
+  "calls.wf_hello.hello.runtimeAttributes.docker": "mirror.gcr.io/ubuntu@sha256:71cd81252a3563a03ad8daee81047b62ab5d892ebbfbf71cf53415f29c130950"
+  "calls.wf_hello.hello.dockerImageUsed": "mirror.gcr.io/ubuntu@sha256:71cd81252a3563a03ad8daee81047b62ab5d892ebbfbf71cf53415f29c130950"
+}

centaur/src/main/resources/standardTestCases/hello_dockermirror_papiv2/hello.inputs

@@ -0,0 +1,4 @@
+{
+  "wf_hello.hello.addressee": "m'Lord"
+}
+

centaur/src/main/resources/standardTestCases/hello_dockermirror_papiv2/hello.wdl

@@ -0,0 +1,20 @@
+task hello {
+  String addressee
+  command {
+    echo "Hello ${addressee}!"
+  }
+  output {
+    String salutation = read_string(stdout())
+  }
+  runtime {
+    backend: "Papiv2DockerMirror"
+    docker: "ubuntu@sha256:71cd81252a3563a03ad8daee81047b62ab5d892ebbfbf71cf53415f29c130950"
+  }
+}
+
+workflow wf_hello {
+  call hello
+  output {
+     hello.salutation
+  }
+}

dockerHashing/src/main/scala/cromwell/docker/DockerImageIdentifier.scala

@@ -11,6 +11,7 @@ sealed trait DockerImageIdentifier {
   def reference: String
 
   def swapReference(newReference: String): DockerImageIdentifier
+  def swapHost(newHost: String): DockerImageIdentifier
 
   // The name of the image with a repository prefix iff a repository was explicitly specified.
   lazy val name = repository map { r => s"$r/$image" } getOrElse image
@@ -35,6 +36,7 @@ case class DockerImageIdentifierWithoutHash(host: Option[String],
 ) extends DockerImageIdentifier {
   def withHash(hash: DockerHashResult) = DockerImageIdentifierWithHash(host, repository, image, reference, hash)
   override def swapReference(newReference: String) = this.copy(reference = newReference)
+  override def swapHost(newHost: String) = this.copy(host = Option(newHost))
 }
 
 case class DockerImageIdentifierWithHash(host: Option[String],
@@ -45,6 +47,7 @@ case class DockerImageIdentifierWithHash(host: Option[String],
 ) extends DockerImageIdentifier {
   override lazy val fullName: String = s"$hostAsString$name@${hash.algorithmAndHash}"
   override def swapReference(newReference: String) = this.copy(reference = newReference)
+  override def swapHost(newHost: String) = this.copy(host = Option(newHost))
 }
 
 object DockerImageIdentifier {

dockerHashing/src/main/scala/cromwell/docker/DockerMirroring.scala

@@ -0,0 +1,52 @@
+package cromwell.docker
+
+import com.typesafe.config.Config
+import com.typesafe.scalalogging.LazyLogging
+import cromwell.docker.registryv2.flows.dockerhub.DockerHub
+import net.ceedubs.ficus.Ficus._
+import net.ceedubs.ficus.readers.ValueReader
+
+case class DockerMirroring(mirrors: List[DockerMirror]) {
+  def mirrorImage(image: DockerImageIdentifier): Option[DockerImageIdentifier] =
+    mirrors.flatMap(_.mirrorImage.lift(image)).headOption
+}
+
+object DockerMirroring {
+  def fromConfig(config: Config): Option[DockerMirroring] = {
+    val mirrors = config
+      .getAs[Config]("docker-mirror")
+      .map { mirrorConfig =>
+        val dockerhubMirror = mirrorConfig.getAs[DockerHubMirror]("dockerhub")
+        // Add support for additional repositories here
+
+        dockerhubMirror.toList
+      }
+      .getOrElse(List[DockerMirror]())
+    Option.when(mirrors.nonEmpty)(DockerMirroring(mirrors))
+  }
+}
+
+sealed trait DockerMirror {
+  val mirrorImage: PartialFunction[DockerImageIdentifier, DockerImageIdentifier]
+}
+
+case class DockerHubMirror(address: String) extends DockerMirror {
+  val mirrorImage: PartialFunction[DockerImageIdentifier, DockerImageIdentifier] = {
+    case i if DockerHub.isValidDockerHubHost(i.host) => i.swapHost(address)
+  }
+}
+
+object DockerHubMirror extends LazyLogging {
+  implicit val dockerHubMirrorOptionValueReader: ValueReader[Option[DockerHubMirror]] =
+    (config: Config, path: String) =>
+      config.getAs[Config](path) flatMap { dockerMirrorConfig =>
+        val enabled = dockerMirrorConfig.as[Boolean]("enabled")
+        val address = dockerMirrorConfig.as[Option[String]]("address").getOrElse("")
+        if (address.isEmpty)
+          logger.warn(
+            "Potential misconfiguration: docker-mirror.dockerhub.enabled=true with no address provided. " +
+              "Mirroring will be disabled."
+          )
+        Option.when(enabled && address.nonEmpty)(DockerHubMirror(address))
+      }
+}

dockerHashing/src/test/scala/cromwell/docker/DockerMirroringSpec.scala

@@ -0,0 +1,88 @@
+package cromwell.docker
+
+import com.typesafe.config.ConfigFactory
+import cromwell.core.TestKitSuite
+import org.scalatest.flatspec.AnyFlatSpecLike
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.prop.TableDrivenPropertyChecks._
+
+class DockerMirroringSpec extends TestKitSuite with AnyFlatSpecLike with Matchers {
+
+  behavior of "DockerMirroring config parsing"
+
+  it should "parse empty config correctly" in {
+    val config = ConfigFactory.parseString("")
+    DockerMirroring.fromConfig(config) shouldBe None
+  }
+
+  it should "parse dockerhub config correctly" in {
+    val config = ConfigFactory.parseString("""
+                                             |docker-mirror {
+                                             |  dockerhub {
+                                             |    enabled: true
+                                             |    address: "foo.bar"
+                                             |  }
+                                             |}
+                                             |""".stripMargin)
+    DockerMirroring.fromConfig(config) shouldBe Some(DockerMirroring(mirrors = List(DockerHubMirror("foo.bar"))))
+  }
+
+  it should "parse no-address dockerhub config correctly" in {
+    val config = ConfigFactory.parseString("""
+                                             |docker-mirror {
+                                             |  dockerhub {
+                                             |    enabled: true
+                                             |  }
+                                             |}
+                                             |""".stripMargin)
+    DockerMirroring.fromConfig(config) shouldBe None
+  }
+
+  it should "parse disabled config correctly" in {
+    val config = ConfigFactory.parseString("""
+                                             |docker-mirror {
+                                             |  dockerhub {
+                                             |    enabled: false
+                                             |    address: "foo.bar"
+                                             |  }
+                                             |}
+                                             |""".stripMargin)
+    DockerMirroring.fromConfig(config) shouldBe None
+  }
+
+  it should "parse non-dockerhub config (unsupported) correctly" in {
+    val config = ConfigFactory.parseString("""
+                                             |docker-mirror {
+                                             |  quay {
+                                             |    enabled: true
+                                             |    address: "foo.bar"
+                                             |  }
+                                             |}
+                                             |""".stripMargin)
+    DockerMirroring.fromConfig(config) shouldBe None
+  }
+
+  behavior of "DockerMirroring"
+
+  val mirroring = DockerMirroring(mirrors = List(DockerHubMirror("my.mirror.io")))
+
+  val dockerMirrorInputs = Table(
+    ("testName", "origImg", "mirrorResult"),
+    ("mirror a Dockerhub image", "ubuntu:latest", Some("my.mirror.io/ubuntu:latest")),
+    ("mirror a docker.io Dockerhub image",
+     "docker.io/broadinstitute/cromwell",
+     Some("my.mirror.io/broadinstitute/cromwell:latest")
+    ),
+    ("not mirror a GCR image", "gcr.io/broad-dsde-cromwell-dev/cromwell-drs-localizer", None)
+  )
+
+  forAll(dockerMirrorInputs) { (testName, origImg, mirrorResult) =>
+    it should testName in {
+      val imgOpt = DockerImageIdentifier.fromString(origImg)
+      val img = imgOpt.get
+      val m = mirroring.mirrorImage(img)
+      val name = m.map(_.fullName)
+      name shouldBe mirrorResult
+    }
+  }
+}

docs/backends/GCPBatch.md

@@ -1,6 +1,5 @@
-**Google Cloud Batch Backend (alpha)**
+**Google Cloud Batch Backend**
 
-[//]:
 Google Cloud Batch is a fully managed service that lets you schedule, queue, and execute batch processing workloads on Google Cloud resources. Batch provisions resources and manages capacity on your behalf, allowing your batch workloads to run at scale.
 
 This section offers detailed configuration instructions for using Cromwell with the Google Cloud Batch in all supported
@@ -9,8 +8,6 @@ authentication modes. Before reading further in this section please see the
 and detailed instructions for the application default authentication scheme in particular.
 The instructions below assume you have created a Google Cloud Storage bucket and a Google project enabled for the appropriate APIs.
 
-*NOTE*: Google Cloud Batch is still in alpha version, this means that there could be breaking changes, be sure to review the [GCP Batch CHANGELOG](https://github.com/broadinstitute/cromwell/blob/develop/CHANGELOG.md#gcp-batch) carefully before upgrading.
-
 **Configuring Authentication**
 
 The `google` stanza in the Cromwell configuration file defines how to authenticate to Google.  There are four different
@@ -164,6 +161,31 @@ backend {
 Note that as per the Google Secret Manager docs, the compute service account for the project in which the GCP Batch
 jobs will run will need to be assigned the `Secret Manager Secret Accessor` IAM role.
 
+***Dockerhub Mirroring***
+
+Cromwell supports automatic use of [GAR's Dockerhub mirror](https://cloud.google.com/artifact-registry/docs/pull-cached-dockerhub-images) 
+in the Batch backend. When enabled, Dockerhub images will be pulled through this mirror rather than directly from Dockerhub. 
+
+To use, include the below `docker-mirror` config in your backend configuration:
+```
+backend {
+  default = GCPBATCH
+  providers {
+    GCPBATCH {
+      actor-factory = "cromwell.backend.google.batch.GcpBatchBackendLifecycleActorFactory"
+      config {
+        docker-mirror {
+          dockerhub {
+            enabled: true
+            address: "mirror.gcr.io"
+          }
+        }
+      }
+    }
+  }
+}
+```
+
 **Monitoring**
 
 In order to monitor metrics (CPU, Memory, Disk usage...) about the VM during Call Runtime, a workflow option can be used to specify the path to a script that will run in the background and write its output to a log file.

engine/src/main/scala/cromwell/engine/workflow/lifecycle/execution/job/preparation/JobPreparationActor.scala

@@ -79,6 +79,8 @@ class JobPreparationActor(workflowDescriptor: EngineWorkflowDescriptor,
   private[preparation] lazy val runtimeAttributeDefinitions = factory.runtimeAttributeDefinitions(initializationData)
   private[preparation] lazy val hasDockerDefinition =
     runtimeAttributeDefinitions.exists(_.name == DockerValidation.instance.key)
+  private[preparation] lazy val dockerMirroring = factory.dockerMirroring
+  private[preparation] lazy val platform = factory.platform
 
   startWith(Idle, JobPreparationActorNoData)
 
@@ -146,7 +148,7 @@ class JobPreparationActor(workflowDescriptor: EngineWorkflowDescriptor,
   private[preparation] def evaluateInputsAndAttributes(
     valueStore: ValueStore
   ): ErrorOr[(WomEvaluatedCallInputs, Map[LocallyQualifiedName, WomValue])] = {
-    import common.validation.ErrorOr.{ShortCircuitingFlatMap, NestedErrorOr}
+    import common.validation.ErrorOr.{NestedErrorOr, ShortCircuitingFlatMap}
     for {
       evaluatedInputs <- ErrorOr(resolveAndEvaluateInputs(jobKey, expressionLanguageFunctions, valueStore)).flatten
       runtimeAttributes <- prepareRuntimeAttributes(evaluatedInputs)
@@ -331,12 +333,31 @@ class JobPreparationActor(workflowDescriptor: EngineWorkflowDescriptor,
     import RuntimeAttributeDefinition.{addDefaultsToAttributes, evaluateRuntimeAttributes}
     val curriedAddDefaultsToAttributes =
       addDefaultsToAttributes(runtimeAttributeDefinitions, workflowDescriptor.backendDescriptor.workflowOptions) _
+
+    def applyDockerMirroring(attributes: Map[LocallyQualifiedName, WomValue]): Map[LocallyQualifiedName, WomValue] = {
+      val mirroredImageAttribute = for {
+        mirror <- dockerMirroring
+        origDockerStr <- attributes.get(RuntimeAttributesKeys.DockerKey).map(_.valueString)
+        origDockerImg <- DockerImageIdentifier.fromString(origDockerStr) match {
+          case Success(i) => Some(i)
+          case Failure(e) =>
+            workflowLogger.warn(s"Failed to attempt mirroring image ${origDockerStr} due to ${e.toString}")
+            None
+        }
+        mirroredImage <- mirror.mirrorImage(origDockerImg)
+        newDockerImageAttribute = (RuntimeAttributesKeys.DockerKey, WomString(mirroredImage.fullName))
+      } yield newDockerImageAttribute
+
+      attributes ++ mirroredImageAttribute.toList
+    }
+
     val unevaluatedRuntimeAttributes = jobKey.call.callable.runtimeAttributes
     evaluateRuntimeAttributes(unevaluatedRuntimeAttributes,
                               expressionLanguageFunctions,
                               inputEvaluation,
-                              factory.platform
-    ) map curriedAddDefaultsToAttributes
+                              platform
+    ) map curriedAddDefaultsToAttributes map applyDockerMirroring
+
   }
 }