Change prewarm container

ningyougang · ningyougang · commit fbbaaf39662c · 2019-01-21T13:06:42.000+08:00
- Add prewarm container
- Delete prewarm container
diff --git a/ansible/group_vars/all b/ansible/group_vars/all
@@ -181,6 +181,8 @@ invoker:
   docker:
     become: "{{ invoker_docker_become | default(false) }}"
   loglevel: "{{ invoker_loglevel | default(whisk_loglevel) | default('INFO') }}"
+  username: "{{ invoker_username | default('invoker.user') }}"
+  password: "{{ invoker_password | default('invoker.pass') }}"
   jmxremote:
     jvmArgs: "{% if inventory_hostname in groups['invokers'] %}
     {{ jmx.jvmCommonArgs }} -Djava.rmi.server.hostname={{ invokerHostname }} -Dcom.sun.management.jmxremote.rmi.port={{ jmx.rmiBasePortInvoker + groups['invokers'].index(inventory_hostname) }} -Dcom.sun.management.jmxremote.port={{ jmx.basePortInvoker + groups['invokers'].index(inventory_hostname) }}
diff --git a/common/scala/src/main/scala/org/apache/openwhisk/core/entity/ExecManifest.scala b/common/scala/src/main/scala/org/apache/openwhisk/core/entity/ExecManifest.scala
@@ -53,6 +53,14 @@ protected[core] object ExecManifest {
     mf
   }
 
+  protected[core] def initializePrewarm(prewarmRuntime: String): Try[Runtimes] = {
+    val rmc = loadConfigOrThrow[RuntimeManifestConfig](ConfigKeys.runtimes)
+    val mf = Try(prewarmRuntime.parseJson.asJsObject).flatMap(runtimes(_, rmc))
+    var manifest: Option[Runtimes] = None
+    mf.foreach(m => manifest = Some(m))
+    mf
+  }
+
   /**
    * Gets existing runtime manifests.
    *
diff --git a/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerPool.scala b/core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerPool.scala
@@ -22,6 +22,7 @@ import org.apache.openwhisk.common.{AkkaLogging, LoggingMarkers, TransactionId}
 import org.apache.openwhisk.core.connector.MessageFeed
 import org.apache.openwhisk.core.entity._
 import org.apache.openwhisk.core.entity.size._
+import spray.json.{JsNumber, JsObject}
 
 import scala.collection.immutable
 import scala.concurrent.duration._
@@ -32,6 +33,9 @@ case object Busy extends WorkerState
 case object Free extends WorkerState
 
 case class WorkerData(data: ContainerData, state: WorkerState)
+case class AddPreWarmConfigList(list: List[PrewarmingConfig])
+case class DeletePreWarmConfigList(list: List[PrewarmingConfig])
+case class PreWarmConfig(kind: String, memory: ByteSize)
 
 /**
  * A pool managing containers to run actions on.
@@ -93,6 +97,30 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
   }
 
   def receive: Receive = {
+    case prewarmConfigList: AddPreWarmConfigList =>
+      prewarmConfigList.list foreach { config =>
+        logging.info(this, s"add extra pre-warming ${config.count} ${config.exec.kind} ${config.memoryLimit.toString}")(
+          TransactionId.invokerWarmup)
+        (1 to config.count).foreach { _ =>
+          prewarmContainer(config.exec, config.memoryLimit)
+        }
+      }
+
+    case prewarmConfigList: DeletePreWarmConfigList =>
+      prewarmConfigList.list foreach { config =>
+        logging.info(this, s"delete pre-warming ${config.count} ${config.exec.kind} ${config.memoryLimit.toString}")(
+          TransactionId.invokerWarmup)
+        (1 to config.count).foreach { _ =>
+          deletePrewarmContainer(config.exec.kind, config.memoryLimit)
+        }
+      }
+
+    case prewarmConfig: PreWarmConfig =>
+      val numberResponse = {
+        JsObject("number" -> JsNumber(getPrewarmContainerNumber(prewarmConfig.kind, prewarmConfig.memory)))
+      }
+      sender() ! numberResponse
+
     // A job to run on a container
     //
     // Run messages are received either via the feed or from child containers which cannot process
@@ -293,6 +321,38 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
       }
   }
 
+  /**
+   * Delete the prewarm container
+   * @param kind
+   * @param memory
+   * @return
+   */
+  def deletePrewarmContainer(kind: String, memory: ByteSize) = {
+    prewarmedPool
+      .find {
+        case (_, PreWarmedData(_, `kind`, `memory`, _)) => true
+        case _                                          => false
+      }
+      .map {
+        case (ref, data) =>
+          ref ! Remove
+          prewarmedPool = prewarmedPool - ref
+      }
+  }
+
+  /**
+   * get the prewarm container number
+   * @param kind
+   * @param memory
+   * @return
+   */
+  def getPrewarmContainerNumber(kind: String, memory: ByteSize) = {
+    prewarmedPool.filter {
+      case (_, PreWarmedData(_, `kind`, `memory`, _)) => true
+      case _                                          => false
+    }.size
+  }
+
   /** Removes a container and updates state accordingly. */
   def removeContainer(toDelete: ActorRef) = {
     toDelete ! Remove
diff --git a/core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/Invoker.scala b/core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/Invoker.scala
@@ -32,7 +32,7 @@ import org.apache.openwhisk.core.containerpool.ContainerPoolConfig
 import org.apache.openwhisk.core.entity.{ExecManifest, InvokerInstanceId}
 import org.apache.openwhisk.core.entity.ActivationEntityLimit
 import org.apache.openwhisk.core.entity.size._
-import org.apache.openwhisk.http.{BasicHttpService, BasicRasService}
+import org.apache.openwhisk.http.BasicHttpService
 import org.apache.openwhisk.spi.SpiLoader
 import org.apache.openwhisk.utils.ExecutionContextFactory
 
@@ -46,6 +46,8 @@ object Invoker {
 
   protected val protocol = loadConfigOrThrow[String]("whisk.invoker.protocol")
 
+  var invokerReactive: Option[InvokerReactive] = None
+
   /**
    * An object which records the environment variables required for this component to run.
    */
@@ -157,7 +159,7 @@ object Invoker {
     }
     val producer = msgProvider.getProducer(config, Some(ActivationEntityLimit.MAX_ACTIVATION_LIMIT))
     val invoker = try {
-      new InvokerReactive(config, invokerInstance, producer, poolConfig)
+      invokerReactive = Some(new InvokerReactive(config, invokerInstance, producer, poolConfig))
     } catch {
       case e: Exception => abort(s"Failed to initialize reactive invoker: ${e.getMessage}")
     }
@@ -172,7 +174,7 @@ object Invoker {
     val httpsConfig =
       if (Invoker.protocol == "https") Some(loadConfigOrThrow[HttpsConfig]("whisk.invoker.https")) else None
 
-    BasicHttpService.startHttpService(new BasicRasService {}.route, port, httpsConfig)(
+    BasicHttpService.startHttpService(new InvokerServer().route, port, httpsConfig)(
       actorSystem,
       ActorMaterializer.create(actorSystem))
   }
diff --git a/core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/InvokerReactive.scala b/core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/InvokerReactive.scala
@@ -20,7 +20,7 @@ package org.apache.openwhisk.core.invoker
 import java.nio.charset.StandardCharsets
 import java.time.Instant
 
-import akka.actor.{ActorRefFactory, ActorSystem, Props}
+import akka.actor.{ActorRef, ActorRefFactory, ActorSystem, Props}
 import akka.event.Logging.InfoLevel
 import akka.stream.ActorMaterializer
 import org.apache.kafka.common.errors.RecordTooLargeException
@@ -195,6 +195,10 @@ class InvokerReactive(
   private val pool =
     actorSystem.actorOf(ContainerPool.props(childFactory, poolConfig, activationFeed, prewarmingConfigs))
 
+  def getContainerPoolInstance: ActorRef = {
+    pool
+  }
+
   /** Is called when an ActivationMessage is read from Kafka */
   def processActivationMessage(bytes: Array[Byte]): Future[Unit] = {
     Future(ActivationMessage.parse(new String(bytes, StandardCharsets.UTF_8)))
diff --git a/core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/InvokerServer.scala b/core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/InvokerServer.scala
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.openwhisk.core.invoker
+
+import akka.actor.ActorSystem
+import akka.http.scaladsl.model.StatusCodes
+import akka.http.scaladsl.model.headers.BasicHttpCredentials
+import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
+import akka.http.scaladsl.server.Route
+import akka.pattern.ask
+import akka.util.Timeout
+
+import org.apache.openwhisk.common.{Logging, TransactionId}
+import org.apache.openwhisk.core.containerpool._
+import org.apache.openwhisk.core.entity.{CodeExecAsString, ExecManifest}
+import org.apache.openwhisk.core.entity.size._
+import org.apache.openwhisk.http.BasicRasService
+
+import spray.json.JsObject
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration._
+
+/**
+ * Implements web server to handle certain REST API calls.
+ * Currently provides a health ping route, only.
+ */
+class InvokerServer(implicit val ec: ExecutionContext,
+                    implicit val actorSystem: ActorSystem,
+                    implicit val logger: Logging)
+    extends BasicRasService {
+
+  val invokerUsername = {
+    val source = scala.io.Source.fromFile("/conf/invokerauth.username");
+    try source.mkString.replaceAll("\r|\n", "")
+    finally source.close()
+  }
+  val invokerPassword = {
+    val source = scala.io.Source.fromFile("/conf/invokerauth.password");
+    try source.mkString.replaceAll("\r|\n", "")
+    finally source.close()
+  }
+
+  override def routes(implicit transid: TransactionId): Route = {
+    super.routes ~ {
+      (path("prewarmContainer") & post) {
+        extractCredentials {
+          case Some(BasicHttpCredentials(username, password)) =>
+            if (username == invokerUsername && password == invokerPassword) {
+              entity(as[String]) { prewarmRuntime =>
+                logger.info(this, s"add prewarmRuntime:${prewarmRuntime}")
+                val execManifest = ExecManifest.initializePrewarm(prewarmRuntime)
+                if (execManifest.isFailure) {
+                  logger.error(this, s"Invalid prewarm runtimes manifest: ${execManifest.failed.get}")
+                  complete("Invalid pass prewarm runtimes manifest")
+                } else {
+                  val prewarmingConfigs: List[PrewarmingConfig] = execManifest.get.stemcells.flatMap {
+                    case (mf, cells) =>
+                      cells.map { cell =>
+                        PrewarmingConfig(cell.count, new CodeExecAsString(mf, "", None), cell.memory)
+                      }
+                  }.toList
+                  Invoker.invokerReactive.get.getContainerPoolInstance ! AddPreWarmConfigList(prewarmingConfigs)
+                  complete("add prewarm container request is handling")
+                }
+              }
+            } else {
+              complete("username or password is wrong")
+            }
+          case _ => complete(StatusCodes.Unauthorized)
+        }
+      }
+    } ~ {
+      (path("prewarmContainer") & delete) {
+        extractCredentials {
+          case Some(BasicHttpCredentials(username, password)) =>
+            if (username == invokerUsername && password == invokerPassword) {
+              entity(as[String]) { prewarmRuntime =>
+                logger.info(this, s"delete prewarmRuntime:${prewarmRuntime}")
+                val execManifest = ExecManifest.initializePrewarm(prewarmRuntime)
+                if (execManifest.isFailure) {
+                  logger.error(this, s"Invalid prewarm runtimes manifest: ${execManifest.failed.get}")
+                  complete("Invalid pass prewarm runtimes manifest")
+                } else {
+                  val prewarmingConfigs: List[PrewarmingConfig] = execManifest.get.stemcells.flatMap {
+                    case (mf, cells) =>
+                      cells.map { cell =>
+                        PrewarmingConfig(cell.count, new CodeExecAsString(mf, "", None), cell.memory)
+                      }
+                  }.toList
+                  Invoker.invokerReactive.get.getContainerPoolInstance ! DeletePreWarmConfigList(prewarmingConfigs)
+                  complete("delete prewarm container request is handling")
+                }
+              }
+            } else {
+              complete("username or password is wrong")
+            }
+          case _ => complete(StatusCodes.Unauthorized)
+        }
+      }
+    } ~ {
+      (path("prewarmContainerNumber") & get) {
+        parameter('kind.as[String], 'memory.as[Int]) { (kind, memory) =>
+          implicit val timeout = Timeout(5.seconds)
+          val numberFuture = Future {
+            Invoker.invokerReactive.get.getContainerPoolInstance
+              .ask(PreWarmConfig(kind, memory.MB))
+              .mapTo[JsObject]
+          }.flatten
+          complete(numberFuture)
+        }
+      }
+    }
+  }
+}
diff --git a/docs/operation.md b/docs/operation.md
@@ -0,0 +1,78 @@
+<!--
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+-->
+
+# Prewarm container operation
+## Add prewarm container on assigned invoker, e.g:
+```
+curl -u ${username}:${password} -X POST http://${invokerAddress}:${invokerPort}/prewarmContainer -d '
+{
+ "runtimes": {
+  "nodejs": [{
+   "kind": "nodejs:6",
+   "default": true,
+   "image": {
+    "prefix": "openwhisk",
+    "name": "nodejs6action",
+    "tag": "latest"
+   },
+   "deprecated": false,
+   "attached": {
+    "attachmentName": "codefile",
+    "attachmentType": "text/plain"
+   },
+   "stemCells": [{
+    "count": 2,
+    "memory": "128 MB"
+   }]
+  }]
+ }
+}
+'
+```
+## Delete prewarm container
+```
+curl -u ${username}:${password} -X DELETE http://${invokerAddress}:${invokerPort}/prewarmContainer -d '
+{
+ "runtimes": {
+  "nodejs": [{
+   "kind": "nodejs:6",
+   "default": true,
+   "image": {
+    "prefix": "openwhisk",
+    "name": "nodejs6action",
+    "tag": "latest"
+   },
+   "deprecated": false,
+   "attached": {
+    "attachmentName": "codefile",
+    "attachmentType": "text/plain"
+   },
+   "stemCells": [{
+    "count": 2,
+    "memory": "128 MB"
+   }]
+  }]
+ }
+}
+'
+```
+## Get prewarm container
+```
+curl  -X GET 'http://${invokerAddress}:${invokerPort}/prewarmContainerNumber?kind=nodejs:6&memory=128'
+```
diff --git a/tests/src/test/scala/common/WhiskProperties.java b/tests/src/test/scala/common/WhiskProperties.java
@@ -263,6 +263,18 @@ public static String getBaseControllerAddress() {
         return getBaseControllerHost() + ":" + getControllerBasePort();
     }
 
+    public static String getBaseInvokerAddress(){
+        return getInvokerHosts()[0] + ":" + whiskProperties.getProperty("invoker.hosts.basePort");
+    }
+
+    public static String getInvokerUsername() {
+        return whiskProperties.getProperty("invoker.username");
+    }
+
+    public static String getInvokerPassword() {
+        return whiskProperties.getProperty("invoker.password");
+    }
+
     public static int getMaxActionInvokesPerMinute() {
         String valStr = whiskProperties.getProperty("limits.actions.invokes.perMinute");
         return Integer.parseInt(valStr);

Original file line number	Diff line number	Diff line change
`@@ -53,6 +53,14 @@ protected[core] object ExecManifest {`
`53`	`53`	`mf`
`54`	`54`	`}`
`55`	`55`
	`56`	`+ protected[core] def initializePrewarm(prewarmRuntime: String): Try[Runtimes] = {`
	`57`	`+ val rmc = loadConfigOrThrow[RuntimeManifestConfig](ConfigKeys.runtimes)`
	`58`	`+ val mf = Try(prewarmRuntime.parseJson.asJsObject).flatMap(runtimes(_, rmc))`
	`59`	`+ var manifest: Option[Runtimes] = None`
	`60`	`+ mf.foreach(m => manifest = Some(m))`
	`61`	`+ mf`
	`62`	`+ }`
	`63`	`+`
`56`	`64`	`/**`
`57`	`65`	`* Gets existing runtime manifests.`
`58`	`66`	`*`