Allow docker pull to be skipped for local docker actions. (#3052)

This commit adds a deployment flag which allows a docker action to be treated as a native image. A native image may eschew a docker pull. It is defined as one that has a prefix matching the docker prefix for managed images.

Also added some missing tids.

Author: rabbah

ansible/environments/docker-machine/group_vars/all

@@ -3,6 +3,7 @@ config_root_dir: /Users/Shared/wskconf
 whisk_logs_dir: /Users/Shared/wsklogs
 docker_registry: ""
 docker_dns: ""
+bypass_pull_for_local_images: true
 
 # The whisk_api_localhost_name is used to configure nginx to permit vanity URLs for web actions.
 # It is also used for the SSL certificate generation. For a local deployment, this is typically

ansible/environments/local/group_vars/all

@@ -3,6 +3,7 @@ config_root_dir: /tmp/wskconf
 whisk_logs_dir: /tmp/wsklogs
 docker_registry: ""
 docker_dns: ""
+bypass_pull_for_local_images: true
 
 db_prefix: whisk_local_
 

ansible/group_vars/all

@@ -29,10 +29,13 @@ whisk:
 #   defaultImageTag: the default image tag
 #   runtimes: set of language runtime families grouped by language (e.g., nodejs, python)
 #   blackboxes: list of pre-populated docker action images as "name" with optional "prefix" and "tag"
+#   bypassPullForLocalImages: optional, if true, allow images with a prefix that matches {{ docker.image.prefix }}
+#                             to skip docker pull in invoker even if the image is not part of the blackboxe set
 #
 runtimesManifest: "{{ runtimes_manifest | default(runtimesManifestDefault) }}"
 
 runtimesManifestDefault:
+  bypassPullForLocalImages: "{{ bypass_pull_for_local_images | default(false) }}"
   defaultImagePrefix: "openwhisk"
   defaultImageTag: "latest"
   runtimes:

common/scala/src/main/scala/whisk/common/TransactionId.scala

@@ -215,6 +215,7 @@ object TransactionId {
   val loadbalancer = TransactionId(-120) // Loadbalancer thread
   val invokerHealth = TransactionId(-121) // Invoker supervision
   val controller = TransactionId(-130) // Controller startup
+  val dbBatcher = TransactionId(-140) // Database batcher
 
   def apply(tid: BigDecimal): TransactionId = {
     Try {

common/scala/src/main/scala/whisk/core/database/CouchDbRestStore.scala

@@ -74,7 +74,7 @@ class CouchDbRestStore[DocumentAbstraction <: DocumentSerializer](dbProtocol: St
   private val maxOpenDbRequests = system.settings.config.getInt("akka.http.host-connection-pool.max-connections") / 2
 
   private val batcher: Batcher[JsObject, Either[ArtifactStoreException, DocInfo]] =
-    new Batcher(500, maxOpenDbRequests)(put(_)(TransactionId.unknown))
+    new Batcher(500, maxOpenDbRequests)(put(_)(TransactionId.dbBatcher))
 
   override protected[database] def put(d: DocumentAbstraction)(implicit transid: TransactionId): Future[DocInfo] = {
     val asJson = d.toDocumentRecord

common/scala/src/main/scala/whisk/core/entity/Exec.scala

@@ -266,7 +266,7 @@ protected[core] object Exec extends ArgNormalizer[Exec] with DefaultJsonProtocol
                 s"if defined, 'code' must a string defined in 'exec' for '${Exec.BLACKBOX}' actions")
             case None => None
           }
-          val native = execManifests.blackboxImages.contains(image)
+          val native = execManifests.skipDockerPull(image)
           BlackBoxExec(image, code, optMainField, native)
 
         case _ =>
@@ -384,8 +384,7 @@ protected[core] object ExecMetaDataBase extends ArgNormalizer[ExecMetaDataBase]
               throw new DeserializationException(
                 s"'image' must be a string defined in 'exec' for '${Exec.BLACKBOX}' actions")
           }
-
-          val native = execManifests.blackboxImages.contains(image)
+          val native = execManifests.skipDockerPull(image)
           BlackBoxExecMetaData(native)
 
         case _ =>

common/scala/src/main/scala/whisk/core/entity/ExecManifest.scala

@@ -17,7 +17,7 @@
 
 package whisk.core.entity
 
-import scala.util.{Failure, Success, Try}
+import scala.util.{Failure, Try}
 import spray.json._
 import spray.json.DefaultJsonProtocol._
 import whisk.core.WhiskConfig
@@ -41,15 +41,13 @@ protected[core] object ExecManifest {
    * singleton Runtime instance.
    *
    * @param config a valid configuration
-   * @param reinit re-initialize singleton iff true
-   * @return the manifest if initialized successfully, or if previously initialized
+   * @param localDockerImagePrefix optional local docker prefix, permitting images matching prefix to bypass docker pull
+   * @return the manifest if initialized successfully, or an failure
    */
-  protected[core] def initialize(config: WhiskConfig, reinit: Boolean = false): Try[Runtimes] = {
-    if (manifest.isEmpty || reinit) {
-      val mf = Try(config.runtimesManifest.parseJson.asJsObject).flatMap(runtimes(_))
-      mf.foreach(m => manifest = Some(m))
-      mf
-    } else Success(manifest.get)
+  protected[core] def initialize(config: WhiskConfig, localDockerImagePrefix: Option[String] = None): Try[Runtimes] = {
+    val mf = Try(config.runtimesManifest.parseJson.asJsObject).flatMap(runtimes(_, localDockerImagePrefix))
+    mf.foreach(m => manifest = Some(m))
+    mf
   }
 
   /**
@@ -71,26 +69,34 @@ protected[core] object ExecManifest {
    * @param config a configuration object as JSON
    * @return Runtimes instance
    */
-  protected[entity] def runtimes(config: JsObject): Try[Runtimes] = Try {
+  protected[entity] def runtimes(config: JsObject, localDockerImagePrefix: Option[String] = None): Try[Runtimes] = Try {
     val prefix = config.fields.get("defaultImagePrefix").map(_.convertTo[String])
     val tag = config.fields.get("defaultImageTag").map(_.convertTo[String])
-    val runtimes = config
-      .fields("runtimes")
-      .convertTo[Map[String, Set[RuntimeManifest]]]
-      .map {
+
+    val runtimes = config.fields
+      .get("runtimes")
+      .map(_.convertTo[Map[String, Set[RuntimeManifest]]].map {
         case (name, versions) =>
           RuntimeFamily(name, versions.map { mf =>
             val img = ImageName(mf.image.name, mf.image.prefix.orElse(prefix), mf.image.tag.orElse(tag))
             mf.copy(image = img)
           })
-      }
-      .toSet
+      }.toSet)
+
     val blackbox = config.fields
       .get("blackboxes")
       .map(_.convertTo[Set[ImageName]].map { image =>
         ImageName(image.name, image.prefix.orElse(prefix), image.tag.orElse(tag))
       })
-    Runtimes(runtimes, blackbox.getOrElse(Set.empty))
+
+    val bypassPullForLocalImages = config.fields
+      .get("bypassPullForLocalImages")
+      .map(_.convertTo[Boolean])
+      .filter(identity)
+      .flatMap(_ => localDockerImagePrefix)
+      .map(_.trim)
+
+    Runtimes(runtimes.getOrElse(Set.empty), blackbox.getOrElse(Set.empty), bypassPullForLocalImages)
   }
 
   /**
@@ -215,10 +221,17 @@ protected[core] object ExecManifest {
    *
    * @param set of supported runtime families
    */
-  protected[core] case class Runtimes(runtimes: Set[RuntimeFamily], blackboxImages: Set[ImageName]) {
+  protected[core] case class Runtimes(runtimes: Set[RuntimeFamily],
+                                      blackboxImages: Set[ImageName],
+                                      bypassPullForLocalImages: Option[String]) {
 
     val knownContainerRuntimes: Set[String] = runtimes.flatMap(_.versions.map(_.kind))
 
+    def skipDockerPull(image: ImageName): Boolean = {
+      blackboxImages.contains(image) ||
+      image.prefix.flatMap(p => bypassPullForLocalImages.map(_ == p)).getOrElse(false)
+    }
+
     def toJson: JsObject = {
       runtimes
         .map { family =>

core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala

@@ -25,6 +25,7 @@ import akka.actor.ActorRefFactory
 import akka.actor.Props
 import whisk.common.AkkaLogging
 
+import whisk.common.TransactionId
 import whisk.core.entity.ByteSize
 import whisk.core.entity.CodeExec
 import whisk.core.entity.EntityName
@@ -72,7 +73,7 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
   var prewarmedPool = immutable.Map.empty[ActorRef, ContainerData]
 
   prewarmConfig.foreach { config =>
-    logging.info(this, s"pre-warming ${config.count} ${config.exec.kind} containers")
+    logging.info(this, s"pre-warming ${config.count} ${config.exec.kind} containers")(TransactionId.invokerWarmup)
     (1 to config.count).foreach { _ =>
       prewarmContainer(config.exec, config.memoryLimit)
     }

core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala

@@ -101,7 +101,7 @@ object Invoker {
       abort("Bad configuration, cannot start.")
     }
 
-    val execManifest = ExecManifest.initialize(config)
+    val execManifest = ExecManifest.initialize(config, localDockerImagePrefix = Some(config.dockerImagePrefix))
     if (execManifest.isFailure) {
       logger.error(this, s"Invalid runtimes manifest: ${execManifest.failed.get}")
       abort("Bad configuration, cannot start.")

tests/src/test/scala/whisk/core/entity/test/ExecManifestTests.scala

@@ -18,20 +18,18 @@
 package whisk.core.entity.test
 
 import java.io.{BufferedWriter, File, FileWriter}
-import java.util.NoSuchElementException
 
-import scala.util.{Success}
+import common.{StreamLogging, WskActorSystem}
 import org.junit.runner.RunWith
-import org.scalatest.FlatSpec
-import org.scalatest.Matchers
+import org.scalatest.{FlatSpec, Matchers}
 import org.scalatest.junit.JUnitRunner
-import spray.json._
 import spray.json.DefaultJsonProtocol._
+import spray.json._
 import whisk.core.WhiskConfig
 import whisk.core.entity.ExecManifest
 import whisk.core.entity.ExecManifest._
-import common.StreamLogging
-import common.WskActorSystem
+
+import scala.util.Success
 
 @RunWith(classOf[JUnitRunner])
 class ExecManifestTests extends FlatSpec with WskActorSystem with StreamLogging with Matchers {
@@ -112,7 +110,10 @@ class ExecManifestTests extends FlatSpec with WskActorSystem with StreamLogging
 
     val mf = JsObject("runtimes" -> JsObject(), "blackboxes" -> imgs.toJson)
     val runtimes = ExecManifest.runtimes(mf).get
+
     runtimes.blackboxImages shouldBe imgs
+    imgs.foreach(img => runtimes.skipDockerPull(img) shouldBe true)
+    runtimes.skipDockerPull(ImageName("???", Some("bbb"))) shouldBe false
   }
 
   it should "read a valid configuration with blackbox images, default prefix and tag" in {
@@ -137,6 +138,8 @@ class ExecManifestTests extends FlatSpec with WskActorSystem with StreamLogging
         ImageName("???", Some("pre"), Some("ttt")))
     }
 
+    runtimes.skipDockerPull(ImageName("???", Some("pre"), Some("test"))) shouldBe true
+    runtimes.skipDockerPull(ImageName("???", Some("bbb"), Some("test"))) shouldBe false
   }
 
   it should "reject runtimes with multiple defaults" in {
@@ -175,21 +178,22 @@ class ExecManifestTests extends FlatSpec with WskActorSystem with StreamLogging
     }
   }
 
-  it should "throw an error when configured manifest is a valid JSON, but with a missing key" in {
-    val config_manifest = """{"nodejs":[{"kind":"nodejs:6","default":true,"image":{"name":"nodejs6action"}}]}"""
+  it should "indicate image is local if it matches deployment docker prefix" in {
+    val config_manifest = """{"bypassPullForLocalImages":true}"""
     val file = File.createTempFile("cxt", ".txt")
     file.deleteOnExit()
 
     val bw = new BufferedWriter(new FileWriter(file))
-    bw.write("runtimes.manifest=" + config_manifest + "\n")
+    bw.write(WhiskConfig.runtimesManifest + s"=$config_manifest\n")
     bw.close()
 
-    val result = ExecManifest.initialize(new WhiskConfig(Map("runtimes.manifest" -> null), Set(), file), true)
-
-    result should be a 'failure
+    val props = Map(WhiskConfig.runtimesManifest -> null)
+    val manifest =
+      ExecManifest.initialize(new WhiskConfig(props, Set(), file), localDockerImagePrefix = Some("localpre"))
+    manifest should be a 'success
 
-    the[NoSuchElementException] thrownBy {
-      result.get
-    } should have message ("key not found: runtimes")
+    manifest.get.skipDockerPull(ImageName(prefix = Some("x"), name = "y")) shouldBe false
+    manifest.get.skipDockerPull(ImageName(prefix = Some("localpre"), name = "y")) shouldBe true
   }
+
 }