Add FPC Load Balancer

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

@@ -37,7 +37,7 @@ class AverageRingBuffer(private val maxSize: Int) {
 
   def nonEmpty: Boolean = elements.nonEmpty
 
-  def  average: Double = {
+  def average: Double = {
     val size = elements.size
     if (size > 2) {
       (sum - max - min) / (size - 2)

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

@@ -61,6 +61,8 @@ abstract class CommonLoadBalancer(config: WhiskConfig,
   protected[loadBalancer] val activationPromises =
     TrieMap[ActivationId, Promise[Either[ActivationId, WhiskActivation]]]()
   protected val activationsPerNamespace = TrieMap[UUID, LongAdder]()
+  protected val activationsPerController = TrieMap[ControllerInstanceId, LongAdder]()
+  protected val activationsPerInvoker = TrieMap[InvokerInstanceId, LongAdder]()
   protected val totalActivations = new LongAdder()
   protected val totalBlackBoxActivationMemory = new LongAdder()
   protected val totalManagedActivationMemory = new LongAdder()
@@ -83,6 +85,14 @@ abstract class CommonLoadBalancer(config: WhiskConfig,
   override def activeActivationsFor(namespace: UUID): Future[Int] =
     Future.successful(activationsPerNamespace.get(namespace).map(_.intValue).getOrElse(0))
   override def totalActiveActivations: Future[Int] = Future.successful(totalActivations.intValue)
+  override def activeActivationsByController(controller: String): Future[Int] =
+    Future.successful(activationsPerController.get(ControllerInstanceId(controller)).map(_.intValue()).getOrElse(0))
+  override def activeActivationsByController: Future[List[ActivationId]] =
+    Future.successful(activationSlots.keySet.toList)
+  override def activeActivationsByInvoker(invoker: String): Future[Int] =
+    Future.successful(
+      activationsPerInvoker.get(InvokerInstanceId(invoker.toInt, userMemory = 0.MB)).map(_.intValue()).getOrElse(0))
+  override def close: Unit = activationFeed ! GracefulShutdown
 
   /**
    * Calculate the duration within which a completion ack must be received for an activation.
@@ -125,6 +135,10 @@ abstract class CommonLoadBalancer(config: WhiskConfig,
     totalActivationMemory.add(action.limits.memory.megabytes)
 
     activationsPerNamespace.getOrElseUpdate(msg.user.namespace.uuid, new LongAdder()).increment()
+    activationsPerController.getOrElseUpdate(controllerInstance, new LongAdder()).increment()
+    activationsPerInvoker
+      .getOrElseUpdate(InvokerInstanceId(instance.instance, userMemory = 0.MB), new LongAdder())
+      .increment()
 
     // Completion Ack must be received within the calculated time.
     val completionAckTimeout = calculateCompletionAckTimeout(action.limits.timeout.duration)
@@ -162,7 +176,8 @@ abstract class CommonLoadBalancer(config: WhiskConfig,
           action.fullyQualifiedName(true),
           timeoutHandler,
           isBlackboxInvocation,
-          msg.blocking)
+          msg.blocking,
+          controllerInstance)
       })
 
     resultPromise
@@ -287,6 +302,10 @@ abstract class CommonLoadBalancer(config: WhiskConfig,
           if (entry.isBlackbox) totalBlackBoxActivationMemory else totalManagedActivationMemory
         totalActivationMemory.add(entry.memoryLimit.toMB * (-1))
         activationsPerNamespace.get(entry.namespaceId).foreach(_.decrement())
+        activationsPerController.get(entry.controllerId).foreach(_.decrement())
+        activationsPerInvoker
+          .get(InvokerInstanceId(entry.invokerName.instance, userMemory = 0.MB))
+          .foreach(_.decrement())
 
         invoker.foreach(releaseInvoker(_, entry))