[New Scheduler] Manage container creation (#5074)

* Manage container creation

* Add ContainerManager Test

* Add dedicated namespace

* Remove namespace

* Apply scala fmt

* Add dedicatedNamespaces filter

* Add dedicatedNamespaces test

* Move InvokerState to common

* Unify InvokerHealth message

* Add configuration for test

* Add license header

* Remove compare InvokerResourceMessage

Author: KeonHee

ansible/group_vars/all

@@ -456,5 +456,6 @@ etcd_connect_string: "{% set ret = [] %}\
                       {{ ret | join(',') }}"
 
 scheduler:
+  protocol: "{{ scheduler_protocol | default('http') }}"
   dataManagementService:
     retryInterval: "{{ scheduler_dataManagementService_retryInterval | default('1 second') }}"

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

@@ -560,9 +560,11 @@ object LoggingMarkers {
       LogMarkerToken(kafka, "topic", start, Some("delay"), Map("topic" -> topic))(MeasurementUnit.time.milliseconds)
     else LogMarkerToken(kafka, topic, start, Some("delay"))(MeasurementUnit.time.milliseconds)
 
+  // Time that is needed to produce message in kafka
+  val SCHEDULER_KAFKA = LogMarkerToken(scheduler, kafka, start)(MeasurementUnit.time.milliseconds)
+
   def SCHEDULER_KEEP_ALIVE(leaseId: Long) =
     LogMarkerToken(scheduler, "keepAlive", counter, None, Map("leaseId" -> leaseId.toString))(MeasurementUnit.none)
-
   /*
    * General markers
    */

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

@@ -17,5 +17,45 @@
 
 package org.apache.openwhisk.common
 
+import org.apache.openwhisk.core.entity.InvokerInstanceId
+
 case object GracefulShutdown
 case object Enable
+
+// States an Invoker can be in
+sealed trait InvokerState {
+  val asString: String
+  val isUsable: Boolean
+}
+
+object InvokerState {
+  // Invokers in this state can be used to schedule workload to
+  sealed trait Usable extends InvokerState { val isUsable = true }
+  // No workload should be scheduled to invokers in this state
+  sealed trait Unusable extends InvokerState { val isUsable = false }
+
+  // A completely healthy invoker, pings arriving fine, no system errors
+  case object Healthy extends Usable { val asString = "up" }
+  // The invoker can not create a container
+  case object Unhealthy extends Unusable { val asString = "unhealthy" }
+  // Pings are arriving fine, the invoker does not respond with active-acks in the expected time though
+  case object Unresponsive extends Unusable { val asString = "unresponsive" }
+  // The invoker is down
+  case object Offline extends Unusable { val asString = "down" }
+}
+
+/**
+ * Describes an abstract invoker. An invoker is a local container pool manager that
+ * is in charge of the container life cycle management.
+ *
+ * @param id a unique instance identifier for the invoker
+ * @param status it status (healthy, unhealthy, unresponsive, offline)
+ */
+case class InvokerHealth(id: InvokerInstanceId, status: InvokerState) {
+  override def equals(obj: scala.Any): Boolean = obj match {
+    case that: InvokerHealth => that.id == this.id && that.status == this.status
+    case _                   => false
+  }
+
+  override def toString = s"InvokerHealth($id, $status)"
+}

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

@@ -232,11 +232,11 @@ object TransactionId {
 
   val systemPrefix = "sid_"
 
-  var containerCreation = TransactionId(systemPrefix + "containerCreation")
   val unknown = TransactionId(systemPrefix + "unknown")
   val testing = TransactionId(systemPrefix + "testing") // Common id for for unit testing
   val invoker = TransactionId(systemPrefix + "invoker") // Invoker startup/shutdown or GC activity
   val invokerHealthManager = TransactionId(systemPrefix + "invokerHealthManager") // Invoker startup/shutdown or GC activity
+  def invokerHealthActivation = TransactionId(systemPrefix + "invokerHealthActivation") // Invoker health activation
   val invokerWarmup = TransactionId(systemPrefix + "invokerWarmup") // Invoker warmup thread that makes stem-cell containers
   val invokerNanny = TransactionId(systemPrefix + "invokerNanny") // Invoker nanny thread
   val dispatcher = TransactionId(systemPrefix + "dispatcher") // Kafka message dispatcher
@@ -245,7 +245,9 @@ object TransactionId {
   val controller = TransactionId(systemPrefix + "controller") // Controller startup
   val dbBatcher = TransactionId(systemPrefix + "dbBatcher") // Database batcher
   val actionHealthPing = TransactionId(systemPrefix + "actionHealth")
-  def invokerHealthActivation = TransactionId(systemPrefix + "invokerHealthActivation") // Invoker health activation
+  var containerCreation = TransactionId(systemPrefix + "containerCreation")
+  var containerDeletion = TransactionId(systemPrefix + "containerDeletion")
+  val warmUp = TransactionId(systemPrefix + "warmUp")
 
   private val dict = ('A' to 'Z') ++ ('a' to 'z') ++ ('0' to '9')
 

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

@@ -0,0 +1,72 @@
+/*
+ * 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
+
+import org.apache.openwhisk.common.TransactionId
+import org.apache.openwhisk.core.connector.{ActivationMessage, ContainerCreationMessage}
+import org.apache.openwhisk.core.entity.ActivationId.ActivationIdGenerator
+import org.apache.openwhisk.core.entity._
+
+object WarmUp {
+  val warmUpActionIdentity = {
+    val whiskSystem = "whisk.system"
+    val uuid = UUID()
+    Identity(Subject(whiskSystem), Namespace(EntityName(whiskSystem), uuid), BasicAuthenticationAuthKey(uuid, Secret()))
+  }
+
+  private val actionName = "warmUp"
+
+  // this action doesn't need to be in database
+  val warmUpAction = FullyQualifiedEntityName(warmUpActionIdentity.namespace.name.toPath, EntityName(actionName))
+
+  def warmUpActivation(controller: ControllerInstanceId) = {
+    ActivationMessage(
+      transid = TransactionId.warmUp,
+      action = warmUpAction,
+      revision = DocRevision.empty,
+      user = warmUpActionIdentity,
+      activationId = new ActivationIdGenerator {}.make(),
+      rootControllerIndex = controller,
+      blocking = false,
+      content = None,
+      initArgs = Set.empty)
+  }
+
+  def warmUpContainerCreationMessage(scheduler: SchedulerInstanceId) =
+    ExecManifest.runtimesManifest
+      .resolveDefaultRuntime("nodejs:default")
+      .map { manifest =>
+        val metadata = WhiskActionMetaData(
+          warmUpAction.path,
+          warmUpAction.name,
+          CodeExecMetaDataAsString(manifest, false, entryPoint = None))
+        ContainerCreationMessage(
+          TransactionId.warmUp,
+          warmUpActionIdentity.namespace.name.toString,
+          warmUpAction,
+          DocRevision.empty,
+          metadata,
+          scheduler,
+          "",
+          0)
+      }
+
+  def isWarmUpAction(fqn: FullyQualifiedEntityName): Boolean = {
+    fqn == warmUpAction
+  }
+}

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

@@ -203,6 +203,7 @@ object WhiskConfig {
 object ConfigKeys {
   val cluster = "whisk.cluster"
   val loadbalancer = "whisk.loadbalancer"
+  val fraction = "whisk.fraction"
   val buildInformation = "whisk.info"
 
   val couchdb = "whisk.couchdb"

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

@@ -24,4 +24,6 @@ object Annotations {
   val RawHttpAnnotationName = "raw-http"
   val RequireWhiskAuthAnnotation = "require-whisk-auth"
   val ProvideApiKeyAnnotationName = "provide-api-key"
+  val InvokerResourcesAnnotationName = "invoker-resources"
+  val InvokerResourcesStrictPolicyAnnotationName = "invoker-resources-strict-policy"
 }

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

@@ -26,12 +26,8 @@ import akka.http.scaladsl.model.Uri
 import akka.http.scaladsl.server.Route
 import akka.stream.ActorMaterializer
 import kamon.Kamon
-import pureconfig._
-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.common._
 import org.apache.openwhisk.core.{ConfigKeys, WhiskConfig}
 import org.apache.openwhisk.core.connector.MessagingProvider
 import org.apache.openwhisk.core.containerpool.logging.LogStoreProvider
@@ -40,13 +36,17 @@ import org.apache.openwhisk.core.entitlement._
 import org.apache.openwhisk.core.entity.ActivationId.ActivationIdGenerator
 import org.apache.openwhisk.core.entity.ExecManifest.Runtimes
 import org.apache.openwhisk.core.entity._
-import org.apache.openwhisk.core.loadBalancer.{InvokerState, LoadBalancerProvider}
+import org.apache.openwhisk.core.loadBalancer.LoadBalancerProvider
 import org.apache.openwhisk.http.{BasicHttpService, BasicRasService}
 import org.apache.openwhisk.spi.SpiLoader
+import pureconfig._
+import spray.json.DefaultJsonProtocol._
+import spray.json._
+import pureconfig.generic.auto._
 
+import scala.concurrent.Await
 import scala.concurrent.ExecutionContext.Implicits
 import scala.concurrent.duration.DurationInt
-import scala.concurrent.Await
 import scala.util.{Failure, Success}
 
 /**

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

@@ -43,28 +43,6 @@ case object GetStatus
 
 case object Tick
 
-// States an Invoker can be in
-sealed trait InvokerState {
-  val asString: String
-  val isUsable: Boolean
-}
-
-object InvokerState {
-  // Invokers in this state can be used to schedule workload to
-  sealed trait Usable extends InvokerState { val isUsable = true }
-  // No workload should be scheduled to invokers in this state
-  sealed trait Unusable extends InvokerState { val isUsable = false }
-
-  // A completely healthy invoker, pings arriving fine, no system errors
-  case object Healthy extends Usable { val asString = "up" }
-  // Pings are arriving fine, the invoker returns system errors though
-  case object Unhealthy extends Unusable { val asString = "unhealthy" }
-  // Pings are arriving fine, the invoker does not respond with active-acks in the expected time though
-  case object Unresponsive extends Unusable { val asString = "unresponsive" }
-  // Pings are not arriving for this invoker
-  case object Offline extends Unusable { val asString = "down" }
-}
-
 // Possible answers of an activation
 sealed trait InvocationFinishedResult
 object InvocationFinishedResult {

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

@@ -17,34 +17,18 @@
 
 package org.apache.openwhisk.core.loadBalancer
 
-import scala.concurrent.Future
 import akka.actor.{ActorRefFactory, ActorSystem, Props}
 import akka.stream.ActorMaterializer
-import org.apache.openwhisk.common.{Logging, TransactionId}
+import org.apache.openwhisk.common.{InvokerHealth, Logging, TransactionId}
 import org.apache.openwhisk.core.WhiskConfig
 import org.apache.openwhisk.core.connector._
 import org.apache.openwhisk.core.controller.Controller
 import org.apache.openwhisk.core.entity._
 import org.apache.openwhisk.spi.Spi
 
+import scala.concurrent.Future
 import scala.concurrent.duration._
 
-/**
- * Describes an abstract invoker. An invoker is a local container pool manager that
- * is in charge of the container life cycle management.
- *
- * @param id a unique instance identifier for the invoker
- * @param status it status (healthy, unhealthy, offline)
- */
-class InvokerHealth(val id: InvokerInstanceId, val status: InvokerState) {
-  override def equals(obj: scala.Any): Boolean = obj match {
-    case that: InvokerHealth => that.id == this.id && that.status == this.status
-    case _                   => false
-  }
-
-  override def toString = s"InvokerHealth($id, $status)"
-}
-
 trait LoadBalancer {
 
   /**