Implement FPCInvokerReactive

Author: ningyougang

ansible/group_vars/all

@@ -465,6 +465,8 @@ etcd_connect_string: "{% set ret = [] %}\
 
 scheduler:
   protocol: "{{ scheduler_protocol | default('http') }}"
+  grpc:
+    tls: "{{ scheduler_grpc_tls | default(false) }}"
   maxPeek: "{{ scheduler_max_peek | default(128) }}"
   queueManager:
     maxSchedulingTime: "{{ scheduler_maxSchedulingTime | default('20 second') }}"

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

@@ -494,6 +494,8 @@ object LoggingMarkers {
         "initiator" -> invocationNamespace,
         "namespace" -> namespace,
         "action" -> action))(MeasurementUnit.none)
+  def INVOKER_CONTAINER_CREATE(action: String, state: String) =
+    LogMarkerToken(invoker, "creation", counter, None, Map("action" -> action, "state" -> state))(MeasurementUnit.none)
   val INVOKER_CONTAINER_HEALTH = LogMarkerToken(invoker, "containerHealth", start)(MeasurementUnit.time.milliseconds)
   val INVOKER_CONTAINER_HEALTH_FAILED_WARM =
     LogMarkerToken(invoker, "containerHealthFailed", counter, Some("warm"), Map("containerState" -> "warm"))(

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

@@ -243,6 +243,7 @@ object ConfigKeys {
   val runtimesRegistry = s"$containerFactory.runtimes-registry"
   val userImagesRegistry = s"$containerFactory.user-images-registry"
   val containerPool = "whisk.container-pool"
+  val containerCreationMaxPeek = "whisk.invoker.container-creation.max-peek"
   val blacklist = "whisk.blacklist"
 
   val kubernetes = "whisk.kubernetes"
@@ -295,6 +296,7 @@ object ConfigKeys {
 
   val azBlob = "whisk.azure-blob"
 
+  val schedulerGrpcService = "whisk.scheduler.grpc"
   val schedulerMaxPeek = "whisk.scheduler.max-peek"
   val schedulerQueueManager = "whisk.scheduler.queue-manager"
   val schedulerInProgressJobRetentionSecond = "whisk.scheduler.in-progress-job-retention"

common/scala/src/main/scala/org/apache/openwhisk/core/ack/HealthActionAck.scala

@@ -0,0 +1,45 @@
+/*
+ * 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.ack
+
+import org.apache.openwhisk.common.{Logging, TransactionId}
+import org.apache.openwhisk.core.connector.{AcknowledegmentMessage, MessageProducer}
+import org.apache.openwhisk.core.entity.{ControllerInstanceId, UUID, WhiskActivation}
+import spray.json.DefaultJsonProtocol._
+
+import scala.concurrent.{ExecutionContext, Future}
+
+class HealthActionAck(producer: MessageProducer)(implicit logging: Logging, ec: ExecutionContext) extends ActiveAck {
+  override def apply(tid: TransactionId,
+                     activationResult: WhiskActivation,
+                     blockingInvoke: Boolean,
+                     controllerInstance: ControllerInstanceId,
+                     userId: UUID,
+                     acknowledegment: AcknowledegmentMessage): Future[Any] = {
+    implicit val transid: TransactionId = tid
+
+    logging.debug(this, s"health action is successfully invoked")
+    if (activationResult.response.isContainerError || activationResult.response.isWhiskError) {
+      val actionPath =
+        activationResult.annotations.getAs[String](WhiskActivation.pathAnnotation).getOrElse("unknown_path")
+      logging.error(this, s"Failed to invoke action $actionPath, error: ${activationResult.response.toString}")
+    }
+
+    Future.successful({})
+  }
+}

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

@@ -35,12 +35,14 @@ case class UserLimits(invocationsPerMinute: Option[Int] = None,
                       concurrentInvocations: Option[Int] = None,
                       firesPerMinute: Option[Int] = None,
                       allowedKinds: Option[Set[String]] = None,
-                      storeActivations: Option[Boolean] = None)
+                      storeActivations: Option[Boolean] = None,
+                      warmedContainerKeepingCount: Option[Int] = None,
+                      warmedContainerKeepingTimeout: Option[String] = None)
 
 object UserLimits extends DefaultJsonProtocol {
   val standardUserLimits = UserLimits()
 
-  implicit val serdes = jsonFormat5(UserLimits.apply)
+  implicit val serdes = jsonFormat7(UserLimits.apply)
 }
 
 protected[core] case class Namespace(name: EntityName, uuid: UUID)

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

@@ -155,6 +155,7 @@ whisk {
       #aka 'How long should a container sit idle until we kill it?'
       idle-container = 10 minutes
       pause-grace = 50 milliseconds
+      keeping-duration = 60 minutes
     }
     action-health-check {
       enabled = false # if true, prewarm containers will be pinged periodically and warm containers will be pinged once after resumed

core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerProxy.scala

@@ -961,7 +961,9 @@ class ContainerProxy(factory: (TransactionId,
   }
 }
 
-final case class ContainerProxyTimeoutConfig(idleContainer: FiniteDuration, pauseGrace: FiniteDuration)
+final case class ContainerProxyTimeoutConfig(idleContainer: FiniteDuration,
+                                             pauseGrace: FiniteDuration,
+                                             keepingDuration: FiniteDuration)
 final case class ContainerProxyHealthCheckConfig(enabled: Boolean, checkPeriod: FiniteDuration, maxFails: Int)
 final case class ContainerProxyActivationErrorLogConfig(applicationErrors: Boolean,
                                                         developerErrors: Boolean,