This is an automated email from the ASF dual-hosted git repository.
bdoyle pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/openwhisk.git
The following commit(s) were added to refs/heads/master by this push:
new 0912c73a4 add per min throttling support to fpc (#5241)
0912c73a4 is described below
commit 0912c73a4bf00caa448d6aea726fbf255b7a187e
Author: Brendan Doyle <[email protected]>
AuthorDate: Tue May 24 12:26:27 2022 -0700
add per min throttling support to fpc (#5241)
Co-authored-by: Brendan Doyle <[email protected]>
---
.../org/apache/openwhisk/core/WhiskConfig.scala | 1 +
core/controller/src/main/resources/reference.conf | 4 ++++
.../openwhisk/core/entitlement/Entitlement.scala | 2 +-
.../openwhisk/core/entitlement/FPCEntitlement.scala | 21 +++++++++++++++++++--
.../core/loadBalancer/FPCPoolBalancer.scala | 2 ++
5 files changed, 27 insertions(+), 3 deletions(-)
diff --git
a/common/scala/src/main/scala/org/apache/openwhisk/core/WhiskConfig.scala
b/common/scala/src/main/scala/org/apache/openwhisk/core/WhiskConfig.scala
index 2547126f3..5836a9fff 100644
--- a/common/scala/src/main/scala/org/apache/openwhisk/core/WhiskConfig.scala
+++ b/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 fpcLoadBalancer = "whisk.loadbalancer.fpc"
val fraction = "whisk.fraction"
val buildInformation = "whisk.info"
diff --git a/core/controller/src/main/resources/reference.conf
b/core/controller/src/main/resources/reference.conf
index 70d2fae51..dfde945a0 100644
--- a/core/controller/src/main/resources/reference.conf
+++ b/core/controller/src/main/resources/reference.conf
@@ -29,6 +29,10 @@ whisk {
# extra time to increase the timeout for forced active acks
# default is 1.minute
timeout-addon = 1m
+
+ fpc {
+ use-perMin-throttles = false
+ }
}
controller {
protocol: http
diff --git
a/core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/Entitlement.scala
b/core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/Entitlement.scala
index 8092ad8f0..88086d20f 100644
---
a/core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/Entitlement.scala
+++
b/core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/Entitlement.scala
@@ -360,7 +360,7 @@ protected[core] abstract class EntitlementProvider(
* @param resources the set of resources must contain at least one resource
that can be activated else return None
* @return future completing successfully if user is below limits else
failing with a rejection
*/
- private def checkUserThrottle(user: Identity, right: Privilege, resources:
Set[Resource])(
+ protected[core] def checkUserThrottle(user: Identity, right: Privilege,
resources: Set[Resource])(
implicit transid: TransactionId): Future[Unit] = {
if (right == ACTIVATE) {
if (resources.exists(_.collection.path == Collection.ACTIONS)) {
diff --git
a/core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/FPCEntitlement.scala
b/core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/FPCEntitlement.scala
index 9e89db23d..3c5a32569 100644
---
a/core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/FPCEntitlement.scala
+++
b/core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/FPCEntitlement.scala
@@ -21,12 +21,16 @@ 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.{ConfigKeys, 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
+import pureconfig.loadConfigOrThrow
+import pureconfig.generic.auto._
+
+case class FPCEntitlementConfig(usePerMinThrottles: Boolean)
protected[core] class FPCEntitlementProvider(
private val config: WhiskConfig,
@@ -34,7 +38,21 @@ protected[core] class FPCEntitlementProvider(
private val controllerInstance: ControllerInstanceId)(implicit actorSystem:
ActorSystem, logging: Logging)
extends LocalEntitlementProvider(config, loadBalancer, controllerInstance)
{
+ private implicit val executionContext = actorSystem.dispatcher
+
+ private val fpcEntitlementConfig: FPCEntitlementConfig =
+ loadConfigOrThrow[FPCEntitlementConfig](ConfigKeys.fpcLoadBalancer)
+
override protected[core] def checkThrottles(user: Identity, right:
Privilege, resources: Set[Resource])(
+ implicit transid: TransactionId): Future[Unit] = {
+ if (fpcEntitlementConfig.usePerMinThrottles) {
+ checkUserThrottle(user, right, resources).flatMap(_ =>
checkFPCConcurrentThrottle(user, right, resources))
+ } else {
+ checkFPCConcurrentThrottle(user, right, resources)
+ }
+ }
+
+ private def checkFPCConcurrentThrottle(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 =>
@@ -55,7 +73,6 @@ protected[core] class FPCEntitlementProvider(
} else Future.successful(())
} else Future.successful(())
}
-
}
private object FPCEntitlementProvider extends EntitlementSpiProvider {
diff --git
a/core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/FPCPoolBalancer.scala
b/core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/FPCPoolBalancer.scala
index 5705099ab..bd72c8cb6 100644
---
a/core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/FPCPoolBalancer.scala
+++
b/core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/FPCPoolBalancer.scala
@@ -33,6 +33,8 @@ import scala.concurrent.{ExecutionContext,
ExecutionContextExecutor, Future, Pro
import scala.language.postfixOps
import scala.util.{Failure, Random, Success, Try}
+case class FPCPoolBalancerConfig(usePerMinThrottle: Boolean)
+
class FPCPoolBalancer(config: WhiskConfig,
controllerInstance: ControllerInstanceId,
etcdClient: EtcdClient,
