Introduce scheduling configurations. (apache#5232)

* Introduce scheduling configurations.

* Apply SchedulingConfig to MemoryQueue.

* Apply SchedulingConfig to SchedulingDecisionMaker.

* Apply ScalaFmt

* Remove unused import

* Change configs.

* Fix test cases.

* Apply scalaFmt

* Remove Java8-compat dependency.

Author: style95

ansible/group_vars/all

@@ -515,6 +515,10 @@ scheduler:
   inProgressJobRetention: "{{ scheduler_inProgressJobRetention | default('20 seconds') }}"
   managedFraction: "{{ scheduler_managed_fraction | default(1.0 - (scheduler_blackbox_fraction | default(__scheduler_blackbox_fraction))) }}"
   blackboxFraction: "{{ scheduler_blackbox_fraction | default(__scheduler_blackbox_fraction) }}"
+  scheduling:
+    staleThreshold: "{{ scheduler_scheduling_staleThreshold | default('100 milliseconds') }}"
+    checkInterval: "{{ scheduler_scheduling_checkInterval | default('100 milliseconds') }}"
+    dropInterval: "{{ scheduler_scheduling_dropInterval | default('10 seconds') }}"
   queueManager:
     maxSchedulingTime: "{{ scheduler_maxSchedulingTime | default('20 second') }}"
     maxRetriesToGetQueue: "{{ scheduler_maxRetriesToGetQueue | default(13) }}"

ansible/roles/schedulers/tasks/deploy.yml

@@ -113,6 +113,9 @@
       "CONFIG_whisk_scheduler_maxPeek": "{{ scheduler.maxPeek }}"
       "CONFIG_whisk_scheduler_dataManagementService_retryInterval": "{{ scheduler.dataManagementService.retryInterval }}"
       "CONFIG_whisk_scheduler_inProgressJobRetention": "{{ scheduler.inProgressJobRetention }}"
+      "CONFIG_whisk_scheduler_scheduling_staleThreshold": "{{ scheduler.scheduling.staleThreshold }}"
+      "CONFIG_whisk_scheduler_scheduling_checkInterval": "{{ scheduler.scheduling.checkInterval }}"
+      "CONFIG_whisk_scheduler_scheduling_dropInterval": "{{ scheduler.scheduling.dropInterval }}"
       "CONFIG_whisk_scheduler_queueManager_maxSchedulingTime": "{{ scheduler.queueManager.maxSchedulingTime }}"
       "CONFIG_whisk_scheduler_queueManager_maxRetriesToGetQueue": "{{ scheduler.queueManager.maxRetriesToGetQueue }}"
       "CONFIG_whisk_scheduler_queue_idleGrace": "{{ scheduler.queue.idleGrace }}"

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

@@ -301,9 +301,11 @@ object ConfigKeys {
 
   val schedulerGrpcService = "whisk.scheduler.grpc"
   val schedulerMaxPeek = "whisk.scheduler.max-peek"
+  val schedulerScheduling = "whisk.scheduler.scheduling"
   val schedulerQueue = "whisk.scheduler.queue"
   val schedulerQueueManager = "whisk.scheduler.queue-manager"
   val schedulerInProgressJobRetention = "whisk.scheduler.in-progress-job-retention"
+  val schedulerStaleThreshold = "whisk.scheduler.stale-threshold"
 
   val whiskClusterName = "whisk.cluster.name"
 

common/scala/src/main/scala/org/apache/openwhisk/core/database/memory/NoopActivationStore.scala

@@ -20,7 +20,12 @@ package org.apache.openwhisk.core.database.memory
 import java.time.Instant
 import akka.actor.ActorSystem
 import org.apache.openwhisk.common.{Logging, PrintStreamLogging, TransactionId, WhiskInstants}
-import org.apache.openwhisk.core.database.{ActivationStore, ActivationStoreProvider, CacheChangeNotification, UserContext}
+import org.apache.openwhisk.core.database.{
+  ActivationStore,
+  ActivationStoreProvider,
+  CacheChangeNotification,
+  UserContext
+}
 import org.apache.openwhisk.core.entity.{ActivationId, DocInfo, EntityName, EntityPath, Subject, WhiskActivation}
 import spray.json.{JsNumber, JsObject}
 

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

@@ -639,17 +639,16 @@ class FPCPoolBalancer(config: WhiskConfig,
       MetricEmitter.emitGaugeMetric(UNHEALTHY_INVOKERS, invokers.count(_.status == Unhealthy))
       MetricEmitter.emitGaugeMetric(OFFLINE_INVOKERS, invokers.count(_.status == Offline))
       // Add both user memory and busy memory because user memory represents free memory in this case
-      MetricEmitter.emitGaugeMetric(
-        INVOKER_TOTALMEM,
-        invokers.foldLeft(0L) { (total, curr) =>
-          if (curr.status.isUsable) {
-            curr.id.userMemory.toMB + curr.id.busyMemory.getOrElse(ByteSize(0, SizeUnits.BYTE)).toMB + total
-          } else {
-            total
-          }
-        })
+      MetricEmitter.emitGaugeMetric(INVOKER_TOTALMEM, invokers.foldLeft(0L) { (total, curr) =>
+        if (curr.status.isUsable) {
+          curr.id.userMemory.toMB + curr.id.busyMemory.getOrElse(ByteSize(0, SizeUnits.BYTE)).toMB + total
+        } else {
+          total
+        }
+      })
       MetricEmitter.emitGaugeMetric(LOADBALANCER_ACTIVATIONS_INFLIGHT(controllerInstance), totalActivations.longValue)
-      MetricEmitter.emitGaugeMetric(LOADBALANCER_MEMORY_INFLIGHT(controllerInstance, ""), totalActivationMemory.longValue)
+      MetricEmitter
+        .emitGaugeMetric(LOADBALANCER_MEMORY_INFLIGHT(controllerInstance, ""), totalActivationMemory.longValue)
     })
   }
 

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

@@ -64,6 +64,11 @@ whisk {
     grpc {
       tls = "false"
     }
+    scheduling {
+      stale-threshold = "100 milliseconds"
+      check-interval = "100 milliseconds"
+      drop-interval = "10 seconds"
+    }
     queue {
       idle-grace = "20 seconds"
       stop-grace = "20 seconds"

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/Scheduler.scala

@@ -70,6 +70,8 @@ class Scheduler(schedulerId: SchedulerInstanceId, schedulerEndpoints: SchedulerE
   val leaseService =
     actorSystem.actorOf(LeaseKeepAliveService.props(etcdClient, schedulerId, watcherService))
 
+  val schedulingConfig = loadConfigOrThrow[SchedulingConfig](ConfigKeys.schedulerQueue)
+
   implicit val entityStore = WhiskEntityStore.datastore()
   private val activationStore =
     SpiLoader.get[ActivationStoreProvider].instance(actorSystem, logging)
@@ -191,15 +193,15 @@ class Scheduler(schedulerId: SchedulerInstanceId, schedulerEndpoints: SchedulerE
     : (ActorRefFactory, String, FullyQualifiedEntityName, DocRevision, WhiskActionMetaData) => ActorRef =
     (factory, invocationNamespace, fqn, revision, actionMetaData) => {
       // Todo: Change this to SPI
-      val decisionMaker = factory.actorOf(SchedulingDecisionMaker.props(invocationNamespace, fqn))
+      val decisionMaker = factory.actorOf(SchedulingDecisionMaker.props(invocationNamespace, fqn, schedulingConfig))
 
       factory.actorOf(
         MemoryQueue.props(
           etcdClient,
           durationChecker,
           fqn,
           producer,
-          config,
+          schedulingConfig,
           invocationNamespace,
           revision,
           schedulerEndpoints,
@@ -395,3 +397,5 @@ object SchedulerStates extends DefaultJsonProtocol {
 
   def parse(states: String) = Try(serdes.read(states.parseJson))
 }
+
+case class SchedulingConfig(staleThreshold: FiniteDuration, checkInterval: FiniteDuration, dropInterval: FiniteDuration)

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/container/ContainerManager.scala

@@ -251,7 +251,8 @@ class ContainerManager(jobManagerFactory: ActorRefFactory => ActorRef,
   // Filter out messages which can use warmed container
   private def filterWarmedCreations(msgs: List[ContainerCreationMessage]) = {
     msgs.filter { msg =>
-      val warmedPrefix = containerPrefix(ContainerKeys.warmedPrefix, msg.invocationNamespace, msg.action, Some(msg.revision))
+      val warmedPrefix =
+        containerPrefix(ContainerKeys.warmedPrefix, msg.invocationNamespace, msg.action, Some(msg.revision))
       val chosenInvoker = warmedContainers
         .filter(!inProgressWarmedContainers.values.toSeq.contains(_))
         .find { container =>

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/MemoryQueue.scala

@@ -24,6 +24,7 @@ import akka.actor.Status.{Failure => FailureMessage}
 import akka.actor.{ActorRef, ActorSystem, Cancellable, FSM, Props, Stash}
 import akka.util.Timeout
 import org.apache.openwhisk.common._
+import org.apache.openwhisk.core.ConfigKeys
 import org.apache.openwhisk.core.ack.ActiveAck
 import org.apache.openwhisk.core.connector.ContainerCreationError.{TooManyConcurrentRequests, ZeroNamespaceLimit}
 import org.apache.openwhisk.core.connector._
@@ -33,22 +34,15 @@ import org.apache.openwhisk.core.entity.size._
 import org.apache.openwhisk.core.etcd.EtcdClient
 import org.apache.openwhisk.core.etcd.EtcdKV.ContainerKeys.containerPrefix
 import org.apache.openwhisk.core.etcd.EtcdKV.{ContainerKeys, QueueKeys, ThrottlingKeys}
-import org.apache.openwhisk.core.scheduler.SchedulerEndpoints
-import org.apache.openwhisk.core.scheduler.message.{
-  ContainerCreation,
-  ContainerDeletion,
-  FailedCreationJob,
-  SuccessfulCreationJob
-}
 import org.apache.openwhisk.core.scheduler.grpc.{GetActivation, ActivationResponse => GetActivationResponse}
 import org.apache.openwhisk.core.scheduler.message.{
   ContainerCreation,
   ContainerDeletion,
   FailedCreationJob,
   SuccessfulCreationJob
 }
+import org.apache.openwhisk.core.scheduler.{SchedulerEndpoints, SchedulingConfig}
 import org.apache.openwhisk.core.service._
-import org.apache.openwhisk.core.{ConfigKeys, WhiskConfig}
 import org.apache.openwhisk.http.Messages.{namespaceLimitUnderZero, tooManyConcurrentRequests}
 import pureconfig.generic.auto._
 import pureconfig.loadConfigOrThrow
@@ -59,7 +53,6 @@ import scala.collection.immutable.Queue
 import scala.collection.mutable
 import scala.concurrent.duration._
 import scala.concurrent.{duration, ExecutionContextExecutor, Future, Promise}
-import scala.language.postfixOps
 import scala.util.{Failure, Success}
 
 // States
@@ -116,7 +109,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
                   private val durationChecker: DurationChecker,
                   private val action: FullyQualifiedEntityName,
                   messagingProducer: MessageProducer,
-                  config: WhiskConfig,
+                  schedulingConfig: SchedulingConfig,
                   invocationNamespace: String,
                   revision: DocRevision,
                   endpoints: SchedulerEndpoints,
@@ -144,11 +137,8 @@ class MemoryQueue(private val etcdClient: EtcdClient,
   private implicit val timeout = Timeout(5.seconds)
   private implicit val order: Ordering[BufferedRequest] = Ordering.by(_.containerId)
 
+  private val StaleDuration = Duration.ofMillis(schedulingConfig.staleThreshold.toMillis)
   private val unversionedAction = action.copy(version = None)
-  private val checkInterval: FiniteDuration = 100 milliseconds
-  private val StaleThreshold: Double = 100.0
-  private val StaleDuration = Duration.ofMillis(StaleThreshold.toLong)
-  private val dropInterval: FiniteDuration = 10 seconds
   private val leaderKey = QueueKeys.queue(invocationNamespace, unversionedAction, leader = true)
   private val inProgressContainerPrefixKey =
     containerPrefix(ContainerKeys.inProgressPrefix, invocationNamespace, action, Some(revision))
@@ -834,7 +824,6 @@ class MemoryQueue(private val etcdClient: EtcdClient,
     }
   }
 
-
   private def handleStaleActivationsWhenActionUpdated(queueManager: ActorRef): Unit = {
     if (queue.size > 0) {
       // if doesn't exist old container to pull old memoryQueue's activation, send the old activations to queueManager
@@ -862,12 +851,12 @@ class MemoryQueue(private val etcdClient: EtcdClient,
   // since there is no initial delay, it will try to create a container at initialization time
   // these schedulers will run forever and stop when the memory queue stops
   private def startMonitoring(): (ActorRef, ActorRef) = {
-    val droppingScheduler = Scheduler.scheduleWaitAtLeast(dropInterval) { () =>
+    val droppingScheduler = Scheduler.scheduleWaitAtLeast(schedulingConfig.dropInterval) { () =>
       checkToDropStaleActivation(queue, queueConfig.maxRetentionMs, invocationNamespace, action, stateName, self)
       Future.successful(())
     }
 
-    val monitoringScheduler = Scheduler.scheduleWaitAtLeast(checkInterval) { () =>
+    val monitoringScheduler = Scheduler.scheduleWaitAtLeast(schedulingConfig.checkInterval) { () =>
       // the average duration is updated every checkInterval
       if (averageDurationBuffer.nonEmpty) {
         averageDuration = Some(averageDurationBuffer.average)
@@ -1048,7 +1037,7 @@ object MemoryQueue {
             durationChecker: DurationChecker,
             fqn: FullyQualifiedEntityName,
             messagingProducer: MessageProducer,
-            config: WhiskConfig,
+            schedulingConfig: SchedulingConfig,
             invocationNamespace: String,
             revision: DocRevision,
             endpoints: SchedulerEndpoints,
@@ -1067,7 +1056,7 @@ object MemoryQueue {
         durationChecker,
         fqn: FullyQualifiedEntityName,
         messagingProducer: MessageProducer,
-        config: WhiskConfig,
+        schedulingConfig: SchedulingConfig,
         invocationNamespace: String,
         revision,
         endpoints: SchedulerEndpoints,

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/SchedulingDecisionMaker.scala

@@ -20,16 +20,19 @@ package org.apache.openwhisk.core.scheduler.queue
 import akka.actor.{Actor, ActorSystem, Props}
 import org.apache.openwhisk.common.Logging
 import org.apache.openwhisk.core.entity.FullyQualifiedEntityName
+import org.apache.openwhisk.core.scheduler.SchedulingConfig
 
 import scala.concurrent.{ExecutionContext, Future}
 import scala.util.{Failure, Success}
 
 class SchedulingDecisionMaker(
   invocationNamespace: String,
   action: FullyQualifiedEntityName,
-  StaleThreshold: Double = 100.0)(implicit val actorSystem: ActorSystem, ec: ExecutionContext, logging: Logging)
+  schedulingConfig: SchedulingConfig)(implicit val actorSystem: ActorSystem, ec: ExecutionContext, logging: Logging)
     extends Actor {
 
+  private val staleThreshold: Double = schedulingConfig.staleThreshold.toMillis.toDouble
+
   override def receive: Receive = {
     case msg: QueueSnapshot =>
       decide(msg)
@@ -135,7 +138,7 @@ class SchedulingDecisionMaker(
 
           case (Running, Some(duration)) if staleActivationNum > 0 =>
             // we can safely get the value as we already checked the existence
-            val containerThroughput = StaleThreshold / duration
+            val containerThroughput = staleThreshold / duration
             val num = ceiling(availableMsg.toDouble / containerThroughput)
             // if it tries to create more containers than existing messages, we just create shortage
             val actualNum = (if (num > availableMsg) availableMsg else num) - inProgress
@@ -153,7 +156,7 @@ class SchedulingDecisionMaker(
           // need more containers and a message is already processed
           case (Running, Some(duration)) =>
             // we can safely get the value as we already checked the existence
-            val containerThroughput = StaleThreshold / duration
+            val containerThroughput = staleThreshold / duration
             val expectedTps = containerThroughput * (existing + inProgress)
 
             if (availableMsg >= expectedTps && existing + inProgress < availableMsg) {
@@ -180,7 +183,7 @@ class SchedulingDecisionMaker(
           // this case is for that as a last resort.
           case (Removing, Some(duration)) if staleActivationNum > 0 =>
             // we can safely get the value as we already checked the existence
-            val containerThroughput = StaleThreshold / duration
+            val containerThroughput = staleThreshold / duration
             val num = ceiling(availableMsg.toDouble / containerThroughput)
             // if it tries to create more containers than existing messages, we just create shortage
             val actualNum = (if (num > availableMsg) availableMsg else num) - inProgress
@@ -252,10 +255,10 @@ class SchedulingDecisionMaker(
 }
 
 object SchedulingDecisionMaker {
-  def props(invocationNamespace: String, action: FullyQualifiedEntityName, StaleThreshold: Double = 100.0)(
+  def props(invocationNamespace: String, action: FullyQualifiedEntityName, schedulingConfig: SchedulingConfig)(
     implicit actorSystem: ActorSystem,
     ec: ExecutionContext,
     logging: Logging): Props = {
-    Props(new SchedulingDecisionMaker(invocationNamespace, action, StaleThreshold))
+    Props(new SchedulingDecisionMaker(invocationNamespace, action, schedulingConfig))
   }
 }