Adjust prewarm container dynamically (#4871)

* Adjust prewarm container dynamically

Co-authored-by: ning.yougang <[email protected]>

Author: ningyougang

ansible/files/runtimes.json

@@ -57,8 +57,15 @@
                 },
                 "stemCells": [
                     {
-                        "count": 2,
-                        "memory": "256 MB"
+                        "initialCount": 2,
+                        "memory": "256 MB",
+                        "reactive": {
+                            "minCount": 1,
+                            "maxCount": 4,
+                            "ttl": "2 minutes",
+                            "threshold": 1,
+                            "increment": 1
+                        }
                     }
                 ]
             },

ansible/roles/invoker/tasks/deploy.yml

@@ -278,6 +278,7 @@
       "CONFIG_whisk_invoker_https_keystorePassword": "{{ invoker.ssl.keystore.password }}"
       "CONFIG_whisk_invoker_https_keystoreFlavor": "{{ invoker.ssl.storeFlavor }}"
       "CONFIG_whisk_invoker_https_clientAuth": "{{ invoker.ssl.clientAuth }}"
+      "CONFIG_whisk_containerPool_prewarmExpirationCheckInterval": "{{ container_pool_prewarm_expirationCheckInterval | default('1 minute') }}"
 
 - name: extend invoker dns env
   set_fact:

common/scala/src/main/scala/org/apache/openwhisk/common/Logging.scala

@@ -511,6 +511,12 @@ object LoggingMarkers {
     LogMarkerToken(containerPool, "prewarmSize", counter)(MeasurementUnit.information.megabytes)
   val CONTAINER_POOL_IDLES_COUNT =
     LogMarkerToken(containerPool, "idlesCount", counter)(MeasurementUnit.none)
+  def CONTAINER_POOL_PREWARM_COLDSTART(memory: String, kind: String) =
+    LogMarkerToken(containerPool, "prewarmColdstart", counter, None, Map("memory" -> memory, "kind" -> kind))(
+      MeasurementUnit.none)
+  def CONTAINER_POOL_PREWARM_EXPIRED(memory: String, kind: String) =
+    LogMarkerToken(containerPool, "prewarmExpired", counter, None, Map("memory" -> memory, "kind" -> kind))(
+      MeasurementUnit.none)
   val CONTAINER_POOL_IDLES_SIZE =
     LogMarkerToken(containerPool, "idlesSize", counter)(MeasurementUnit.information.megabytes)
 

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerFactory.scala

@@ -24,6 +24,7 @@ import org.apache.openwhisk.core.entity.{ByteSize, ExecManifest, ExecutableWhisk
 import org.apache.openwhisk.spi.Spi
 
 import scala.concurrent.Future
+import scala.concurrent.duration.FiniteDuration
 import scala.math.max
 
 case class ContainerArgsConfig(network: String,
@@ -43,11 +44,16 @@ case class ContainerArgsConfig(network: String,
     }.toMap
 }
 
-case class ContainerPoolConfig(userMemory: ByteSize, concurrentPeekFactor: Double, akkaClient: Boolean) {
+case class ContainerPoolConfig(userMemory: ByteSize,
+                               concurrentPeekFactor: Double,
+                               akkaClient: Boolean,
+                               prewarmExpirationCheckInterval: FiniteDuration) {
   require(
     concurrentPeekFactor > 0 && concurrentPeekFactor <= 1.0,
     s"concurrentPeekFactor must be > 0 and <= 1.0; was $concurrentPeekFactor")
 
+  require(prewarmExpirationCheckInterval.toSeconds > 0, "prewarmExpirationCheckInterval must be > 0")
+
   /**
    * The shareFactor indicates the number of containers that would share a single core, on average.
    * cpuShare is a docker option (-c) whereby a container's CPU access is limited.

common/scala/src/main/scala/org/apache/openwhisk/core/entity/ExecManifest.scala

@@ -26,7 +26,10 @@ import spray.json.DefaultJsonProtocol._
 import org.apache.openwhisk.core.{ConfigKeys, WhiskConfig}
 import org.apache.openwhisk.core.entity.Attachments._
 import org.apache.openwhisk.core.entity.Attachments.Attached._
-import fastparse._, NoWhitespace._
+import fastparse._
+import NoWhitespace._
+
+import scala.concurrent.duration.{Duration, FiniteDuration}
 
 /**
  * Reads manifest of supported runtimes from configuration file and stores
@@ -135,11 +138,37 @@ protected[core] object ExecManifest {
   /**
    * A stemcell configuration read from the manifest for a container image to be initialized by the container pool.
    *
-   * @param count  the number of stemcell containers to create
+   * @param initialCount  the initial number of stemcell containers to create
    * @param memory the max memory this stemcell will allocate
+   * @param reactive the reactive prewarming prewarmed config, which is disabled by default
    */
-  protected[entity] case class StemCell(count: Int, memory: ByteSize) {
-    require(count > 0, "count must be positive")
+  protected[entity] case class StemCell(initialCount: Int,
+                                        memory: ByteSize,
+                                        reactive: Option[ReactivePrewarmingConfig] = None) {
+    require(initialCount > 0, "initialCount must be positive")
+  }
+
+  /**
+   * A stemcell's ReactivePrewarmingConfig configuration
+   *
+   * @param minCount the max number of stemcell containers to exist
+   * @param maxCount  the max number of stemcell containers to create
+   * @param ttl time to live of the prewarmed container
+   * @param threshold the executed activation number of cold start in previous one minute
+   * @param increment increase per increment prewarmed number under per threshold activations
+   */
+  protected[core] case class ReactivePrewarmingConfig(minCount: Int,
+                                                      maxCount: Int,
+                                                      ttl: FiniteDuration,
+                                                      threshold: Int,
+                                                      increment: Int) {
+    require(
+      minCount >= 0 && minCount <= maxCount,
+      "minCount must be be greater than 0 and less than or equal to maxCount")
+    require(maxCount > 0, "maxCount must be positive")
+    require(ttl.toMillis > 0, "ttl must be positive")
+    require(threshold > 0, "threshold must be positive")
+    require(increment > 0 && increment <= maxCount, "increment must be positive and less than or equal to maxCount")
   }
 
   /**
@@ -344,9 +373,45 @@ protected[core] object ExecManifest {
 
   protected[entity] implicit val imageNameSerdes: RootJsonFormat[ImageName] = jsonFormat4(ImageName.apply)
 
-  protected[entity] implicit val stemCellSerdes: RootJsonFormat[StemCell] = {
+  protected[entity] implicit val ttlSerdes: RootJsonFormat[FiniteDuration] = new RootJsonFormat[FiniteDuration] {
+    override def write(finiteDuration: FiniteDuration): JsValue = JsString(finiteDuration.toString)
+
+    override def read(value: JsValue): FiniteDuration = value match {
+      case JsString(s) =>
+        val duration = Duration(s)
+        FiniteDuration(duration.length, duration.unit)
+      case _ =>
+        deserializationError("time unit not supported. Only milliseconds, seconds, minutes, hours, days are supported")
+    }
+  }
+
+  protected[entity] implicit val reactivePrewarmingConfigSerdes: RootJsonFormat[ReactivePrewarmingConfig] = jsonFormat5(
+    ReactivePrewarmingConfig.apply)
+
+  protected[entity] implicit val stemCellSerdes = new RootJsonFormat[StemCell] {
     import org.apache.openwhisk.core.entity.size.serdes
-    jsonFormat2(StemCell.apply)
+    val defaultSerdes = jsonFormat3(StemCell.apply)
+    override def read(value: JsValue): StemCell = {
+      val fields = value.asJsObject.fields
+      val initialCount: Option[Int] =
+        fields
+          .get("initialCount")
+          .orElse(fields.get("count"))
+          .map(_.convertTo[Int])
+      val memory: Option[ByteSize] = fields.get("memory").map(_.convertTo[ByteSize])
+      val config = fields.get("reactive").map(_.convertTo[ReactivePrewarmingConfig])
+
+      (initialCount, memory) match {
+        case (Some(c), Some(m)) => StemCell(c, m, config)
+        case (Some(c), None) =>
+          throw new IllegalArgumentException(s"memory is required, just provide initialCount: ${c}")
+        case (None, Some(m)) =>
+          throw new IllegalArgumentException(s"initialCount is required, just provide memory: ${m.toString}")
+        case _ => throw new IllegalArgumentException("both initialCount and memory are required")
+      }
+    }
+
+    override def write(s: StemCell) = defaultSerdes.write(s)
   }
 
   protected[entity] implicit val runtimeManifestSerdes: RootJsonFormat[RuntimeManifest] = jsonFormat8(RuntimeManifest)

core/invoker/src/main/resources/application.conf

@@ -60,6 +60,7 @@ whisk {
     user-memory: 1024 m
     concurrent-peek-factor: 0.5 #factor used to limit message peeking: 0 < factor <= 1.0 - larger number improves concurrent processing, but increases risk of message loss during invoker crash
     akka-client:  false # if true, use PoolingContainerClient for HTTP from invoker to action container (otherwise use ApacheBlockingContainerClient)
+    prewarm-expiration-check-interval: 1 minute
   }
 
   kubernetes {