[New Scheduler]Implement FPCEntitlementProvider (#5029)

* Implement FPCEntitlementProvider

* Add throttling metric

Author: jiangpengcheng

core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/Entitlement.scala

@@ -153,7 +153,7 @@ protected[core] abstract class EntitlementProvider(
       activationThrottleCalculator(config.actionInvokeConcurrentLimit.toInt, _.limits.concurrentInvocations))
 
   private val messagingProvider = SpiLoader.get[MessagingProvider]
-  private val eventProducer = messagingProvider.getProducer(this.config)
+  protected val eventProducer = messagingProvider.getProducer(this.config)
 
   /**
    * Grants a subject the right to access a resources.
@@ -201,6 +201,19 @@ protected[core] abstract class EntitlementProvider(
       .flatMap(_ => checkThrottleOverload(concurrentInvokeThrottler.check(user), user))
   }
 
+  /**
+   * Checks action activation rate throttles for an identity.
+   *
+   * @param user      the identity to check rate throttles for
+   * @param right     the privilege the subject is requesting
+   * @param resources the set of resource the subject requests access to
+   * @return a promise that completes with success iff the user is within their activation quota
+   */
+  protected[core] def checkThrottles(user: Identity, right: Privilege, resources: Set[Resource])(
+    implicit transid: TransactionId): Future[Unit] = {
+    checkUserThrottle(user, right, resources).flatMap(_ => checkConcurrentUserThrottle(user, right, resources))
+  }
+
   private val kindRestrictor = {
     import pureconfig._
     import pureconfig.generic.auto._
@@ -284,11 +297,7 @@ protected[core] abstract class EntitlementProvider(
     val entitlementCheck: Future[Unit] = if (user.rights.contains(right)) {
       if (resources.nonEmpty) {
         logging.debug(this, s"checking user '$subject' has privilege '$right' for '${resources.mkString(", ")}'")
-        val throttleCheck =
-          if (noThrottle) Future.successful(())
-          else
-            checkUserThrottle(user, right, resources)
-              .flatMap(_ => checkConcurrentUserThrottle(user, right, resources))
+        val throttleCheck = if (noThrottle) Future.successful(()) else checkThrottles(user, right, resources)
         throttleCheck
           .flatMap(_ => checkPrivilege(user, right, resources))
           .flatMap(checkedResources => {

core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/FPCEntitlement.scala

@@ -0,0 +1,67 @@
+/*
+ * 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.entitlement
+
+import scala.concurrent.Future
+import akka.actor.ActorSystem
+import akka.http.scaladsl.model.StatusCodes.TooManyRequests
+import org.apache.openwhisk.common.{Logging, TransactionId, UserEvents}
+import org.apache.openwhisk.core.WhiskConfig
+import org.apache.openwhisk.core.connector.{EventMessage, Metric}
+import org.apache.openwhisk.core.controller.RejectRequest
+import org.apache.openwhisk.core.entitlement.Privilege.ACTIVATE
+import org.apache.openwhisk.core.entity.{ControllerInstanceId, Identity}
+import org.apache.openwhisk.core.loadBalancer.LoadBalancer
+
+protected[core] class FPCEntitlementProvider(
+  private val config: WhiskConfig,
+  private val loadBalancer: LoadBalancer,
+  private val controllerInstance: ControllerInstanceId)(implicit actorSystem: ActorSystem, logging: Logging)
+    extends LocalEntitlementProvider(config, loadBalancer, controllerInstance) {
+
+  override protected[core] def checkThrottles(user: Identity, right: Privilege, resources: Set[Resource])(
+    implicit transid: TransactionId): Future[Unit] = {
+    if (right == ACTIVATE) {
+      val checks = resources.filter(_.collection.path == Collection.ACTIONS).map { res =>
+        loadBalancer.checkThrottle(user.namespace.name.toPath, res.fqname)
+      }
+      if (checks.contains(true)) {
+        val metric = Metric("ConcurrentRateLimit", 1)
+        UserEvents.send(
+          eventProducer,
+          EventMessage(
+            s"controller${controllerInstance.asString}",
+            metric,
+            user.subject,
+            user.namespace.name.toString,
+            user.namespace.uuid,
+            metric.typeName))
+        Future.failed(RejectRequest(TooManyRequests, "Too many requests"))
+      } else Future.successful(())
+    } else Future.successful(())
+  }
+
+}
+
+private object FPCEntitlementProvider extends EntitlementSpiProvider {
+
+  override def instance(config: WhiskConfig, loadBalancer: LoadBalancer, instance: ControllerInstanceId)(
+    implicit actorSystem: ActorSystem,
+    logging: Logging) =
+    new FPCEntitlementProvider(config: WhiskConfig, loadBalancer: LoadBalancer, instance)
+}

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

@@ -75,6 +75,9 @@ trait LoadBalancer {
 
   /** Gets the size of the cluster all loadbalancers are acting in */
   def clusterSize: Int = 1
+
+  /** Gets the throttling for given action. */
+  def checkThrottle(namespace: EntityPath, action: String): Boolean = false
 }
 
 /**