[GitHub] [openwhisk] style95 commented on a change in pull request #4917: Adjust user memory via api

[GitBox <gi...@apache.org>]

style95 commented on a change in pull request #4917:
URL: https://github.com/apache/openwhisk/pull/4917#discussion_r523418447



##########
File path: core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/LoadBalancer.scala
##########
@@ -60,6 +60,14 @@ trait LoadBalancer {
   def publish(action: ExecutableWhiskActionMetaData, msg: ActivationMessage)(
     implicit transid: TransactionId): Future[Future[Either[ActivationId, WhiskActivation]]]
 
+  /**
+   * send user memory to invokers
+   *
+   * @param userMemory
+   * @param targetInvokers
+   */
+  def sendUserMemoryToInvoker(userMemoryMessage: UserMemoryMessage, targetInvoker: Int): Unit = {}

Review comment:
       How about changing the name like this?
   
   ```suggestion
     def sendChangeRequestToInvoker(userMemoryMessage: UserMemoryMessage, targetInvoker: Int): Unit = {}
   ```

##########
File path: core/controller/src/main/scala/org/apache/openwhisk/core/controller/Controller.scala
##########
@@ -176,6 +184,43 @@ class Controller(val instance: ControllerInstanceId,
     LogLimit.config,
     runtimes,
     List(apiV1.basepath()))
+
+  private val controllerCredentials = loadConfigOrThrow[ControllerCredentials](ConfigKeys.controllerCredentials)
+
+  /**
+   * config user memory of ContainerPool
+   */
+  import org.apache.openwhisk.core.connector.ConfigMemoryProtocol._
+  private val configMemory = {
+    implicit val executionContext = actorSystem.dispatcher
+    (path("config" / "memory") & post) {
+      extractCredentials {
+        case Some(BasicHttpCredentials(username, password)) =>
+          if (username == controllerCredentials.username && password == controllerCredentials.password) {
+            entity(as[String]) { memory =>

Review comment:
       Can't we directly convert this to `ConfigMemoryList`?
   

##########
File path: core/controller/src/main/scala/org/apache/openwhisk/core/controller/Controller.scala
##########
@@ -176,6 +184,43 @@ class Controller(val instance: ControllerInstanceId,
     LogLimit.config,
     runtimes,
     List(apiV1.basepath()))
+
+  private val controllerCredentials = loadConfigOrThrow[ControllerCredentials](ConfigKeys.controllerCredentials)
+
+  /**
+   * config user memory of ContainerPool
+   */
+  import org.apache.openwhisk.core.connector.ConfigMemoryProtocol._
+  private val configMemory = {
+    implicit val executionContext = actorSystem.dispatcher
+    (path("config" / "memory") & post) {
+      extractCredentials {
+        case Some(BasicHttpCredentials(username, password)) =>
+          if (username == controllerCredentials.username && password == controllerCredentials.password) {
+            entity(as[String]) { memory =>
+              val configMemoryList = memory.parseJson.convertTo[ConfigMemoryList]
+
+              val existIllegalUserMemory = configMemoryList.items.exists { configMemory =>

Review comment:
       Scalaism. I think we can change this something like this:
   
   configMemoryList.items.find(MemoryLimit.MIN_MEMORY.compare(_.memory) > 0)) match {
    case Some(_) => 
      // reject the request
    case None =>
   }

##########
File path: core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerPool.scala
##########
@@ -74,6 +76,7 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
   var busyPool = immutable.Map.empty[ActorRef, ContainerData]
   var prewarmedPool = immutable.Map.empty[ActorRef, PreWarmedData]
   var prewarmStartingPool = immutable.Map.empty[ActorRef, (String, ByteSize)]
+  var latestUserMemory = poolConfig.userMemory

Review comment:
       I think we can take advantage of `Option`.
   

##########
File path: common/scala/src/main/scala/org/apache/openwhisk/core/connector/Message.scala
##########
@@ -427,3 +427,43 @@ object EventMessage extends DefaultJsonProtocol {
 
   def parse(msg: String) = Try(format.read(msg.parseJson))
 }
+
+case class UserMemoryMessage(userMemory: ByteSize) extends Message {
+  override def serialize = UserMemoryMessage.serdes.write(this).compactPrint
+}
+
+object UserMemoryMessage extends DefaultJsonProtocol {
+  implicit val serdes = new RootJsonFormat[UserMemoryMessage] {
+    override def write(message: UserMemoryMessage): JsValue = {
+      JsObject("userMemory" -> JsString(message.userMemory.toString))
+    }
+
+    override def read(json: JsValue): UserMemoryMessage = {
+      val userMemory = fromField[String](json, "userMemory")
+      new UserMemoryMessage(ByteSize.fromString(userMemory))
+    }
+  }
+
+  def parse(msg: String) = Try(serdes.read(msg.parseJson))
+}
+
+case class ConfigMemory(invoker: Int, memory: ByteSize)
+case class ConfigMemoryList(items: List[ConfigMemory])

Review comment:
       I believe we can directly unmarshal a list of `ConfigMemory` rather than haveing this case class.

##########
File path: core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/Invoker.scala
##########
@@ -220,9 +223,3 @@ trait InvokerServerProvider extends Spi {
   def instance(
     invoker: InvokerCore)(implicit ec: ExecutionContext, actorSystem: ActorSystem, logger: Logging): BasicRasService
 }
-
-object DefaultInvokerServer extends InvokerServerProvider {

Review comment:
       Why is this removed?
   This is being used.
   https://github.com/apache/openwhisk/blob/master/common/scala/src/main/resources/reference.conf#L30
   

##########
File path: common/scala/src/main/scala/org/apache/openwhisk/core/connector/Message.scala
##########
@@ -427,3 +427,43 @@ object EventMessage extends DefaultJsonProtocol {
 
   def parse(msg: String) = Try(format.read(msg.parseJson))
 }
+
+case class UserMemoryMessage(userMemory: ByteSize) extends Message {
+  override def serialize = UserMemoryMessage.serdes.write(this).compactPrint
+}
+
+object UserMemoryMessage extends DefaultJsonProtocol {
+  implicit val serdes = new RootJsonFormat[UserMemoryMessage] {
+    override def write(message: UserMemoryMessage): JsValue = {
+      JsObject("userMemory" -> JsString(message.userMemory.toString))
+    }
+
+    override def read(json: JsValue): UserMemoryMessage = {
+      val userMemory = fromField[String](json, "userMemory")
+      new UserMemoryMessage(ByteSize.fromString(userMemory))
+    }
+  }
+
+  def parse(msg: String) = Try(serdes.read(msg.parseJson))
+}
+
+case class ConfigMemory(invoker: Int, memory: ByteSize)

Review comment:
       How about changing the name to `InvokerConfiguration`.
   We might extend this to other invoker configurations as well in the future.
   

##########
File path: core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerPool.scala
##########
@@ -305,6 +308,13 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
     case RescheduleJob =>
       freePool = freePool - sender()
       busyPool = busyPool - sender()
+    case message: ByteSizeMessage =>
+      logging.info(
+        this,
+        s"user memory is reconfigured from ${latestUserMemory.toString} to ${message.userMemory.toString}")
+      latestUserMemory = message.userMemory
+    case UserMemoryQuery =>
+      sender() ! latestUserMemory.toString

Review comment:
       Ok since this is just a `ByteSize` I think we can take this approach.
   




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org