DEVX-533: Implement LoadBalancer Strategy based on annotation

- addressed code review comments

Author: vrann

core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/KindBasedLoadBalancer.scala renamed to core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/MuxBalancer.scala

@@ -12,13 +12,12 @@ import spray.json._
 
 import scala.concurrent.Future
 
-class KindBasedLoadBalancer(config: WhiskConfig,
-                            feedFactory: FeedFactory,
-                            controllerInstance: ControllerInstanceId,
-                            implicit val messagingProvider: MessagingProvider = SpiLoader.get[MessagingProvider])(
+class MuxBalancer(config: WhiskConfig,
+                  feedFactory: FeedFactory,
+                  controllerInstance: ControllerInstanceId,
+                  implicit val messagingProvider: MessagingProvider = SpiLoader.get[MessagingProvider])(
   implicit actorSystem: ActorSystem,
-  logging: Logging,
-  materializer: ActorMaterializer)
+  logging: Logging)
     extends CommonLoadBalancer(config, feedFactory, controllerInstance) {
 
   private val balancers = lbConfig.strategy.foldLeft(Map.empty[String, LoadBalancer]) {
@@ -37,10 +36,13 @@ class KindBasedLoadBalancer(config: WhiskConfig,
   }
   override protected val invokerPool: ActorRef = actorSystem.actorOf(Props.empty)
 
-  /** 1. Publish a message to the loadbalancer */
+  /**
+    * Publish a message to the loadbalancer
+    *
+    * Select the LoadBalancer based on the annotation, if available, otherwise use the default one
+    **/
   override def publish(action: ExecutableWhiskActionMetaData, msg: ActivationMessage)(
     implicit transid: TransactionId): Future[Future[Either[ActivationId, WhiskActivation]]] = {
-    logging.info(this, "We are here")
     action.annotations.get("activationStrategy") match {
       case None =>
         balancers("default").publish(action, msg)
@@ -51,26 +53,18 @@ class KindBasedLoadBalancer(config: WhiskConfig,
           balancers("default").publish(action, msg)
         }
       }
-      case (Some(_))  =>  balancers("default").publish(action, msg)
+      case Some(_)  =>  balancers("default").publish(action, msg)
     }
-
-//    if (action.annotations.get("activationStrategy").isDefined && balancers.contains(action.annotations.get("activationStrategy").get.toString.)
-//    {
-//      balancers().publish(action, msg)
-//    } else {
-//      balancers("default").publish(action, msg)
-//    }
   }
 }
 
-object KindBasedLoadBalancer extends LoadBalancerProvider {
+object MuxBalancer extends LoadBalancerProvider {
 
   override def instance(whiskConfig: WhiskConfig, instance: ControllerInstanceId)(
     implicit actorSystem: ActorSystem,
-    logging: Logging,
-    materializer: ActorMaterializer): LoadBalancer = {
+    logging: Logging): LoadBalancer = {
 
-    new KindBasedLoadBalancer(whiskConfig, createFeedFactory(whiskConfig, instance), instance)
+    new MuxBalancer(whiskConfig, createFeedFactory(whiskConfig, instance), instance)
   }
 
   def requiredProperties =

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

@@ -598,12 +598,28 @@ case class ClusterConfig(useClusterBootstrap: Boolean)
  * @param timeoutFactor factor to influence the timeout period for forced active acks (time-limit.std * timeoutFactor + timeoutAddon)
  * @param timeoutAddon extra time to influence the timeout period for forced active acks (time-limit.std * timeoutFactor + timeoutAddon)
  */
-case class ShardingContainerPoolBalancerConfig(strategy: Map[String, String],
+case class ShardingContainerPoolBalancerConfig(strategy: ActivationStrategy,
                                                managedFraction: Double,
                                                blackboxFraction: Double,
                                                timeoutFactor: Int,
                                                timeoutAddon: FiniteDuration)
 
+/**
+  * Configuration for the annotation-based load balancer multiplexer
+  *
+  * @param default the default strategy to be used if nothing is configured for the given annotation
+  * @param custom the Map of the strategy name to strategy configuration
+  */
+case class ActivationStrategy(default: String,
+                              custom: Map[String, StrategyConfig])
+
+/**
+  * Configuration for the strategy
+  *
+  * @param className indicates the class which will handle this strategy name
+  */
+case class StrategyConfig(className: String)
+
 /**
  * State kept for each activation slot until completion.
  *

core/standalone/src/main/resources/standalone.conf

@@ -36,7 +36,7 @@ whisk {
   spi {
     ArtifactStoreProvider = "org.apache.openwhisk.core.database.memory.MemoryArtifactStoreProvider"
     MessagingProvider = "org.apache.openwhisk.connector.lean.LeanMessagingProvider"
-    LoadBalancerProvider = "org.apache.openwhisk.core.loadBalancer.KindBasedLoadBalancer"
+    LoadBalancerProvider = "org.apache.openwhisk.core.loadBalancer.MuxBalancer"
     # Use cli based log store for all setups as its more stable to use
     # and does not require root user access
     LogStoreProvider = "org.apache.openwhisk.core.containerpool.docker.DockerCliLogStoreProvider"