Config runtime

Sometimes, admin may want to reinitalize the runtime config depend on
the real requirements, e.g. increase some prewarm containers

Author: ningyougang

ansible/group_vars/all

@@ -105,6 +105,8 @@ controller:
   authentication:
     spi: "{{ controller_authentication_spi | default('') }}"
   loglevel: "{{ controller_loglevel | default(whisk_loglevel) | default('INFO') }}"
+  username: "{{ controller_username | default('controller.user') }}"
+  password: "{{ controller_password | default('controller.pass') }}"
   entitlement:
     spi: "{{ controller_entitlement_spi | default('') }}"
   protocol: "{{ controller_protocol | default('https') }}"
@@ -201,6 +203,8 @@ invoker:
     runcdir: "{{ invoker_runcdir | default('/run/docker/runtime-runc/moby') }}"
     volumes: "{{ invoker_docker_volumes | default([]) }}"
   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) }}

ansible/roles/controller/tasks/deploy.yml

@@ -203,6 +203,9 @@
       "CONFIG_whisk_db_activationsFilterDdoc": "{{ db_whisk_activations_filter_ddoc | default() }}"
       "CONFIG_whisk_userEvents_enabled": "{{ user_events | default(false) | lower }}"
 
+      "CONFIG_whisk_credentials_controller_username": "{{ controller.username }}"
+      "CONFIG_whisk_credentials_controller_password": "{{ controller.password }}"
+
       "LIMITS_ACTIONS_INVOKES_PERMINUTE": "{{ limits.invocationsPerMinute }}"
       "LIMITS_ACTIONS_INVOKES_CONCURRENT": "{{ limits.concurrentInvocations }}"
       "LIMITS_TRIGGERS_FIRES_PERMINUTE": "{{ limits.firesPerMinute }}"

ansible/roles/invoker/tasks/deploy.yml

@@ -266,6 +266,8 @@
       "CONFIG_whisk_timeLimit_min": "{{ limit_action_time_min | default() }}"
       "CONFIG_whisk_timeLimit_max": "{{ limit_action_time_max | default() }}"
       "CONFIG_whisk_timeLimit_std": "{{ limit_action_time_std | default() }}"
+      "CONFIG_whisk_credentials_invoker_username": "{{ invoker.username }}"
+      "CONFIG_whisk_credentials_invoker_password": "{{ invoker.password }}"
       "CONFIG_whisk_concurrencyLimit_min": "{{ limit_action_concurrency_min | default() }}"
       "CONFIG_whisk_concurrencyLimit_max": "{{ limit_action_concurrency_max | default() }}"
       "CONFIG_whisk_concurrencyLimit_std": "{{ limit_action_concurrency_std | default() }}"

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

@@ -0,0 +1,22 @@
+/*
+ * 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.common
+
+case class ControllerCredentials(username: String, password: String)
+
+case class InvokerCredentials(username: String, password: String)

common/scala/src/main/scala/org/apache/openwhisk/core/WhiskConfig.scala

@@ -266,4 +266,7 @@ object ConfigKeys {
   val swaggerUi = "whisk.swagger-ui"
 
   val apacheClientConfig = "whisk.apache-client"
+
+  val controllerCredentials = "whisk.credentials.controller"
+  val invokerCredentials = "whisk.credentials.invoker"
 }

common/scala/src/main/scala/org/apache/openwhisk/core/connector/Message.scala

@@ -429,3 +429,30 @@ object EventMessage extends DefaultJsonProtocol {
 
   def parse(msg: String) = Try(format.read(msg.parseJson))
 }
+
+case class RuntimeMessage(runtime: String) extends Message {
+  override def serialize = RuntimeMessage.serdes.write(this).compactPrint
+}
+
+object RuntimeMessage extends DefaultJsonProtocol {
+  def parse(msg: String) = Try(serdes.read(msg.parseJson))
+  implicit val serdes = jsonFormat(RuntimeMessage.apply _, "runtime")
+}
+
+case class PrewarmContainerData(kind: String, memory: Long, var number: Int) extends Message {
+  override def serialize: String = PrewarmContainerData.serdes.write(this).compactPrint
+}
+
+object PrewarmContainerData extends DefaultJsonProtocol {
+  implicit val serdes = jsonFormat(PrewarmContainerData.apply _, "kind", "memory", "number")
+}
+
+case class PrewarmContainerDataList(items: List[PrewarmContainerData])
+
+object PrewarmContainerDataProtocol extends DefaultJsonProtocol {
+  implicit val prewarmContainerDataFormat = jsonFormat(PrewarmContainerData.apply _, "kind", "memory", "number")
+  implicit object prewarmContainerDataListJsonFormat extends RootJsonFormat[PrewarmContainerDataList] {
+    def read(value: JsValue) = PrewarmContainerDataList(value.convertTo[List[PrewarmContainerData]])
+    def write(f: PrewarmContainerDataList) = ???
+  }
+}

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

@@ -55,6 +55,20 @@ protected[core] object ExecManifest {
     mf
   }
 
+  /**
+   * Reads runtimes manifest from runtime string
+   *
+   * @param runtime
+   * @return the manifest if initialized successfully, or an failure
+   */
+  protected[core] def initialize(runtime: String): Try[Runtimes] = {
+    val rmc = loadConfigOrThrow[RuntimeManifestConfig](ConfigKeys.runtimes)
+    val mf = Try(runtime.parseJson.asJsObject).flatMap(runtimes(_, rmc))
+    var manifest: Option[Runtimes] = None
+    mf.foreach(m => manifest = Some(m))
+    mf
+  }
+
   /**
    * Gets existing runtime manifests.
    *

core/controller/src/main/scala/org/apache/openwhisk/core/controller/Controller.scala

@@ -21,7 +21,8 @@ import akka.Done
 import akka.actor.{ActorSystem, CoordinatedShutdown}
 import akka.event.Logging.InfoLevel
 import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
-import akka.http.scaladsl.model.Uri
+import akka.http.scaladsl.model.{StatusCodes, Uri}
+import akka.http.scaladsl.model.headers.BasicHttpCredentials
 import akka.http.scaladsl.server.Route
 import akka.stream.ActorMaterializer
 import kamon.Kamon
@@ -30,8 +31,15 @@ import pureconfig.generic.auto._
 import spray.json.DefaultJsonProtocol._
 import spray.json._
 import org.apache.openwhisk.common.Https.HttpsConfig
-import org.apache.openwhisk.common.{AkkaLogging, ConfigMXBean, Logging, LoggingMarkers, TransactionId}
-import org.apache.openwhisk.core.WhiskConfig
+import org.apache.openwhisk.common.{
+  AkkaLogging,
+  ConfigMXBean,
+  ControllerCredentials,
+  Logging,
+  LoggingMarkers,
+  TransactionId
+}
+import org.apache.openwhisk.core.{ConfigKeys, WhiskConfig}
 import org.apache.openwhisk.core.connector.MessagingProvider
 import org.apache.openwhisk.core.containerpool.logging.LogStoreProvider
 import org.apache.openwhisk.core.database.{ActivationStoreProvider, CacheChangeNotification, RemoteCacheInvalidation}
@@ -96,7 +104,7 @@ class Controller(val instance: ControllerInstanceId,
       (pathEndOrSingleSlash & get) {
         complete(info)
       }
-    } ~ apiV1.routes ~ swagger.swaggerRoutes ~ internalInvokerHealth
+    } ~ apiV1.routes ~ swagger.swaggerRoutes ~ internalInvokerHealth ~ configRuntime
   }
 
   // initialize datastores
@@ -163,6 +171,56 @@ class Controller(val instance: ControllerInstanceId,
     LogLimit.config,
     runtimes,
     List(apiV1.basepath()))
+
+  private val controllerCredentials = loadConfigOrThrow[ControllerCredentials](ConfigKeys.controllerCredentials)
+
+  /**
+   * config runtime
+   */
+  private val configRuntime = {
+    implicit val executionContext = actorSystem.dispatcher
+    (path("config" / "runtime") & post) {
+      extractCredentials {
+        case Some(BasicHttpCredentials(username, password)) =>
+          if (username == controllerCredentials.username && password == controllerCredentials.password) {
+            entity(as[String]) { runtime =>
+              val execManifest = ExecManifest.initialize(runtime)
+              if (execManifest.isFailure) {
+                logging.error(this, s"Received invalid runtimes manifest")
+                complete(s"Received invalid runtimes manifest")
+              } else {
+                parameter('limit.?) { limit =>
+                  limit match {
+                    case Some(targetValue) =>
+                      val pattern = "\\d+:\\d"
+                      if (targetValue.matches(pattern)) {
+                        val invokerArray = targetValue.split(":")
+                        val beginIndex = invokerArray(0).toInt
+                        val finishIndex = invokerArray(1).toInt
+                        if (finishIndex < beginIndex) {
+                          complete(s"finishIndex can't be less than beginIndex")
+                        } else {
+                          val targetInvokers = (beginIndex to finishIndex).toList
+                          loadBalancer.sendRuntimeToInvokers(runtime, Some(targetInvokers))
+                          complete(s"config runtime request is already sent to target invokers")
+                        }
+                      } else {
+                        complete(s"limit value can't match [beginIndex:finishIndex]")
+                      }
+                    case None =>
+                      loadBalancer.sendRuntimeToInvokers(runtime, None)
+                      complete(s"config runtime request is already sent to all managed invokers")
+                  }
+                }
+              }
+            }
+          } else {
+            complete("username or password is wrong")
+          }
+        case _ => complete(StatusCodes.Unauthorized)
+      }
+    }
+  }
 }
 
 /**

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 runtime to invokers
+   *
+   * @param runtime
+   * @param targetInvokers
+   */
+  def sendRuntimeToInvokers(runtime: String, targetInvokers: Option[List[Int]]): Unit = {}
+
   /**
    * Returns a message indicating the health of the containers and/or container pool in general.
    *

core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/ShardingContainerPoolBalancer.scala

@@ -43,6 +43,7 @@ import org.apache.openwhisk.spi.SpiLoader
 import scala.annotation.tailrec
 import scala.concurrent.Future
 import scala.concurrent.duration.FiniteDuration
+import scala.util.{Failure, Success}
 
 /**
  * A loadbalancer that schedules workload based on a hashing-algorithm.
@@ -316,6 +317,22 @@ class ShardingContainerPoolBalancer(
       }
   }
 
+  /** send runtime to invokers*/
+  override def sendRuntimeToInvokers(runtime: String, targetInvokers: Option[List[Int]]): Unit = {
+    val runtimeMessage = RuntimeMessage(runtime)
+    schedulingState.managedInvokers.filter { manageInvoker =>
+      targetInvokers.getOrElse(schedulingState.managedInvokers.map(_.id.instance)).contains(manageInvoker.id.instance)
+    } foreach { invokerHealth =>
+      val topic = s"invoker${invokerHealth.id.toInt}"
+      messageProducer.send(topic, runtimeMessage).andThen {
+        case Success(_) =>
+          logging.info(this, s"Successfully posted runtime to topic $topic")
+        case Failure(_) =>
+          logging.error(this, s"Failed posted runtime to topic $topic")
+      }
+    }
+  }
+
   override val invokerPool =
     invokerPoolFactory.createInvokerPool(
       actorSystem,