add config to fail async scheduler throttles as whisk errors (apache#5305)

* add config to fail async scheduler throttles as whisk errors

* fix tests

Co-authored-by: Brendan Doyle <[email protected]>

Author: 2 people

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

@@ -79,6 +79,7 @@ whisk {
       max-blackbox-retention-ms = "300000"
       throttling-fraction = "0.9"
       duration-buffer-size = "10"
+      fail-throttle-as-whisk-error = "false"
     }
     queue-manager {
       max-scheduling-time = "20 seconds"

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

@@ -227,7 +227,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
       enableNamespaceThrottling()
 
       if (dropMsg)
-        completeAllActivations(tooManyConcurrentRequests, isWhiskError = false)
+        completeAllActivations(tooManyConcurrentRequests, isWhiskError = queueConfig.failThrottleAsWhiskError)
       goto(NamespaceThrottled) using ThrottledData(data.schedulerActor, data.droppingActor)
 
     case Event(StateTimeout, data: RunningData) =>
@@ -269,7 +269,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
   when(NamespaceThrottled) {
     case Event(msg: ActivationMessage, _: ThrottledData) =>
       if (containers.size + creationIds.size == 0) {
-        completeErrorActivation(msg, tooManyConcurrentRequests, isWhiskError = false)
+        completeErrorActivation(msg, tooManyConcurrentRequests, isWhiskError = queueConfig.failThrottleAsWhiskError)
       } else {
         handleActivationMessage(msg)
       }
@@ -285,7 +285,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
   when(ActionThrottled) {
     // since there are already too many activation messages, it drops the new messages
     case Event(msg: ActivationMessage, ThrottledData(_, _)) =>
-      completeErrorActivation(msg, tooManyConcurrentRequests, isWhiskError = false)
+      completeErrorActivation(msg, tooManyConcurrentRequests, isWhiskError = queueConfig.failThrottleAsWhiskError)
       stay
   }
 
@@ -823,7 +823,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
       CompletionMessage(activation.transid, activationResponse, instance)
     }
 
-    if (!isWhiskError && message == tooManyConcurrentRequests) {
+    if (message == tooManyConcurrentRequests) {
       val metric = Metric("ConcurrentRateLimit", 1)
       UserEvents.send(
         messagingProducer,
@@ -1208,7 +1208,8 @@ case class QueueConfig(idleGrace: FiniteDuration,
                        maxRetentionMs: Long,
                        maxBlackboxRetentionMs: Long,
                        throttlingFraction: Double,
-                       durationBufferSize: Int)
+                       durationBufferSize: Int,
+                       failThrottleAsWhiskError: Boolean)
 
 case class BufferedRequest(containerId: String, promise: Promise[Either[MemoryQueueError, ActivationMessage]])
 case object DropOld

tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/MemoryQueueFlowTests.scala

@@ -517,7 +517,7 @@ class MemoryQueueFlowTests
 
     // max retention size is 10 and throttling fraction is 0.8
     // queue will be action throttled at 10 messages and disabled action throttling at 8 messages
-    val queueConfig = QueueConfig(5 seconds, 10 seconds, 10 seconds, 5 seconds, 10, 5000, 10000, 0.8, 10)
+    val queueConfig = QueueConfig(5 seconds, 10 seconds, 10 seconds, 5 seconds, 10, 5000, 10000, 0.8, 10, false)
 
     // limit is 1
     val getUserLimit = (_: String) => Future.successful(1)

tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/MemoryQueueTests.scala

@@ -1116,7 +1116,7 @@ class MemoryQueueTests
 
     expectDurationChecking(mockEsClient, testInvocationNamespace)
 
-    val queueConfig = QueueConfig(5 seconds, 10 seconds, 10 seconds, 5 seconds, 10, 10000, 20000, 0.9, 10)
+    val queueConfig = QueueConfig(5 seconds, 10 seconds, 10 seconds, 5 seconds, 10, 10000, 20000, 0.9, 10, false)
 
     val fsm =
       TestFSMRef(
@@ -1342,7 +1342,7 @@ class MemoryQueueTests
 
     expectDurationChecking(mockEsClient, testInvocationNamespace)
 
-    val queueConfig = QueueConfig(5 seconds, 10 seconds, 10 seconds, 5 seconds, 10, 10000, 20000, 0.9, 10)
+    val queueConfig = QueueConfig(5 seconds, 10 seconds, 10 seconds, 5 seconds, 10, 10000, 20000, 0.9, 10, false)
 
     val fsm =
       TestFSMRef(
@@ -1585,7 +1585,7 @@ class MemoryQueueTests
     // it always induces the throttling
     val getZeroLimit = (_: String) => { Future.successful(2) }
 
-    val queueConfig = QueueConfig(5 seconds, 10 seconds, 10 seconds, 5 seconds, 1, 5000, 10000, 0.9, 10)
+    val queueConfig = QueueConfig(5 seconds, 10 seconds, 10 seconds, 5 seconds, 1, 5000, 10000, 0.9, 10, false)
 
     expectDurationChecking(mockEsClient, testInvocationNamespace)
 
@@ -1632,7 +1632,7 @@ class MemoryQueueTests
     val probe = TestProbe()
     val parent = TestProbe()
 
-    val queueConfig = QueueConfig(5 seconds, 10 seconds, 10 seconds, 5 seconds, 10, 5000, 10000, 0.9, 10)
+    val queueConfig = QueueConfig(5 seconds, 10 seconds, 10 seconds, 5 seconds, 10, 5000, 10000, 0.9, 10, false)
     val msgRetentionSize = queueConfig.maxRetentionSize
 
     val tid = TransactionId(TransactionId.generateTid())

tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/MemoryQueueTestsFixture.scala

@@ -154,7 +154,7 @@ class MemoryQueueTestsFixture
   val actionThrottlingKey = ThrottlingKeys.action(testInvocationNamespace, fqn.copy(version = None))
 
   // queue variables
-  val queueConfig = QueueConfig(5 seconds, 10 seconds, 5 seconds, 5 seconds, 10, 10000, 20000, 0.9, 10)
+  val queueConfig = QueueConfig(5 seconds, 10 seconds, 5 seconds, 5 seconds, 10, 10000, 20000, 0.9, 10, false)
   val idleGrace = queueConfig.idleGrace
   val stopGrace = queueConfig.stopGrace
   val flushGrace = queueConfig.flushGrace