| meichstedt | |
| timcharper | |
| kensipe |
Use maxLaunchDelay to determine what delays to GC.
In addition to that, on conditions like Running, Created, existing delays are
advanced to make sure that delays are applied to failures, and time taken by
provisioning containers doesn't get subtracted from them.
In the future we might implement removal of rate-limiting delays when
a corresponding RunSpec becomes healthy.
This is a back-port from master branch.
sbt test
| No Linters Available |
| No Unit Test Coverage |
| Buildable 3412 | |
| Build 6521: Marathon (revised) | Jenkins |
| Build 6520: arc lint + arc unit |
I thought our process was to put back ports on GH PRs and not on phabricator... otherwise lgtm
| Commit | Tree | Parents | Author | Summary | Date |
|---|---|---|---|---|---|
| c31572492902 | 437118433e0a | 7b414d9046cb | Ivan Chernetsky | Do a better job at maintaining task failure rate limiting values per RunSpec (Show More…) | Aug 30 2017, 10:50 PM |
| Show First 20 Lines • Show All 84 Lines • ▼ Show 20 Line(s) | |||
| 85 | 85 | def asyncPurge(specId: PathId): Future[Done] | |
|---|---|---|---|
| 86 | 86 | ||
| 87 | 87 | /** Add delay to the given RunnableSpec because of a failed instance */ | |
| 88 | 88 | def addDelay(spec: RunSpec): Unit | |
| 89 | 89 | ||
| 90 | 90 | /** Reset the backoff delay for the given RunnableSpec. */ | |
| 91 | 91 | def resetDelay(spec: RunSpec): Unit | |
| 92 | 92 | ||
| 93 | /** Advance the reference time point of the delay for the given RunSpec */ | ||
| 94 | def advanceDelay(spec: RunSpec): Unit | ||
| 95 | |||
| 93 | 96 | /** Notify queue about InstanceUpdate */ | |
| 94 | 97 | def notifyOfInstanceUpdate(update: InstanceChange): Future[Done] | |
| 95 | 98 | } | |
| Show First 20 Lines • Show All 90 Lines • ▼ Show 20 Line(s) | |||
| 91 | 91 | case NonFatal(e) => throw new RuntimeException(s"in $method", e) | |
|---|---|---|---|
| 92 | 92 | } | |
| 93 | 93 | answerFuture.mapTo[R] | |
| 94 | 94 | } | |
| 95 | 95 | ||
| 96 | 96 | override def addDelay(spec: RunSpec): Unit = rateLimiterRef ! RateLimiterActor.AddDelay(spec) | |
| 97 | 97 | ||
| 98 | 98 | override def resetDelay(spec: RunSpec): Unit = rateLimiterRef ! RateLimiterActor.ResetDelay(spec) | |
| 99 | |||
| 100 | override def advanceDelay(spec: RunSpec): Unit = rateLimiterRef ! RateLimiterActor.AdvanceDelay(spec) | ||
| 99 | 101 | } | |
| 100 | 102 | ||
| 101 | 103 | private[impl] object LaunchQueueDelegate { | |
| 102 | 104 | sealed trait Request | |
| 103 | 105 | case object List extends Request | |
| 104 | 106 | case object ListWithStatistics extends Request | |
| 105 | 107 | case class Count(runSpecId: PathId) extends Request | |
| 106 | 108 | case class Purge(runSpecId: PathId) extends Request | |
| 107 | 109 | case object ConfirmPurge extends Request | |
| 108 | 110 | case class Add(spec: RunSpec, count: Int) extends Request | |
| 109 | 111 | } | |
| 1 | 1 | package mesosphere.marathon.core.launchqueue.impl | |
|---|---|---|---|
| 2 | 2 | ||
| 3 | 3 | import java.util.concurrent.TimeUnit | |
| 4 | 4 | ||
| 5 | 5 | import mesosphere.marathon.core.base.Clock | |
| 6 | 6 | import mesosphere.marathon.state.{ PathId, RunSpec, Timestamp } | |
| 7 | import mesosphere.util.DurationToHumanReadable | ||
| 8 | |||
| 7 | 9 | import org.slf4j.LoggerFactory | |
| 8 | 10 | ||
| 9 | 11 | import scala.concurrent.duration._ | |
| 10 | 12 | ||
| 11 | 13 | /** | |
| 12 | 14 | * Manages the task launch delays for every run spec and config version. | |
| 13 | 15 | * | |
| 14 | 16 | * We do not keep the delays for every version because that would include scaling changes or manual restarts. | |
| 15 | 17 | */ | |
| 16 | 18 | private[launchqueue] class RateLimiter(clock: Clock) { | |
| 17 | 19 | import RateLimiter._ | |
| 18 | 20 | ||
| 19 | 21 | /** The task launch delays per run spec and their last config change. */ | |
| 20 | 22 | private[this] var taskLaunchDelays = Map[(PathId, Timestamp), Delay]() | |
| 21 | 23 | ||
| 24 | /** Reset delays for tasks that have reached the maximum launch delay threshold. */ | ||
| 22 | 25 | def cleanUpOverdueDelays(): Unit = { | |
| 26 | val now = clock.now() | ||
| 23 | 27 | taskLaunchDelays = taskLaunchDelays.filter { | |
| 24 | case (_, delay) => delay.deadline > clock.now() | ||
| 28 | case (_, delay) => | ||
| 29 | now <= delay.referenceTimestamp + delay.maxLaunchDelay | ||
| 25 | 30 | } | |
| 26 | 31 | } | |
| 27 | 32 | ||
| 28 | def getDelay(spec: RunSpec): Timestamp = | ||
| 33 | def getDeadline(spec: RunSpec): Timestamp = | ||
| 29 | 34 | // TODO (pods): RunSpec has no versionInfo. Need this? | |
| 30 | 35 | taskLaunchDelays.get(spec.id -> spec.versionInfo.lastConfigChangeVersion).map(_.deadline) getOrElse clock.now() | |
| 31 | 36 | ||
| 32 | 37 | def addDelay(spec: RunSpec): Timestamp = { | |
| 33 | 38 | setNewDelay(spec, "Increasing delay") { | |
| 34 | case Some(delay) => Some(delay.increased(clock, spec)) | ||
| 35 | case None => Some(Delay(clock, spec)) | ||
| 39 | case Some(delay) => delay.increased(clock, spec) | ||
| 40 | case None => Delay(clock, spec) | ||
| 36 | 41 | } | |
| 37 | 42 | } | |
| 38 | 43 | ||
| 39 | private[this] def setNewDelay(spec: RunSpec, message: String)( | ||
| 40 | calcDelay: Option[Delay] => Option[Delay]): Timestamp = { | ||
| 44 | private[this] def setNewDelay(spec: RunSpec, message: String)(calcDelay: Option[Delay] => Delay): Timestamp = { | ||
| 41 | 45 | val maybeDelay: Option[Delay] = taskLaunchDelays.get(spec.id -> spec.versionInfo.lastConfigChangeVersion) | |
| 42 | calcDelay(maybeDelay) match { | ||
| 43 | case Some(newDelay) => | ||
| 44 | import mesosphere.util.DurationToHumanReadable | ||
| 46 | val newDelay = calcDelay(maybeDelay) | ||
| 45 | 47 | val now: Timestamp = clock.now() | |
| 46 | val priorTimeLeft = (now until maybeDelay.map(_.deadline).getOrElse(now)).toHumanReadable | ||
| 48 | |||
| 47 | 49 | val timeLeft = (now until newDelay.deadline).toHumanReadable | |
| 48 | 50 | ||
| 49 | if (newDelay.deadline <= now) { | ||
| 50 | resetDelay(spec) | ||
| 51 | } else { | ||
| 52 | log.info(s"$message. Task launch delay for [${spec.id}] changed from [$priorTimeLeft] to [$timeLeft].") | ||
| 53 | taskLaunchDelays += ((spec.id, spec.versionInfo.lastConfigChangeVersion) -> newDelay) | ||
| 54 | } | ||
| 51 | log.info( | ||
| 52 | s"$message. Task launch delay for [${spec.id} - ${spec.versionInfo.lastConfigChangeVersion}] is set to $timeLeft") | ||
| 53 | taskLaunchDelays += ((spec.id -> spec.versionInfo.lastConfigChangeVersion) -> newDelay) | ||
| 55 | 54 | newDelay.deadline | |
| 56 | |||
| 57 | case None => | ||
| 58 | resetDelay(spec) | ||
| 59 | clock.now() | ||
| 60 | 55 | } | |
| 61 | } | ||
| 62 | 56 | ||
| 63 | 57 | def resetDelay(runSpec: RunSpec): Unit = { | |
| 64 | if (taskLaunchDelays contains (runSpec.id -> runSpec.versionInfo.lastConfigChangeVersion)) { | ||
| 58 | val key = runSpec.id -> runSpec.versionInfo.lastConfigChangeVersion | ||
| 59 | taskLaunchDelays.get(key).foreach { _ => | ||
| 65 | 60 | log.info(s"Task launch delay for [${runSpec.id} - ${runSpec.versionInfo.lastConfigChangeVersion}}] reset to zero") | |
| 66 | taskLaunchDelays -= (runSpec.id -> runSpec.versionInfo.lastConfigChangeVersion) | ||
| 61 | taskLaunchDelays -= key | ||
| 67 | 62 | } | |
| 68 | 63 | } | |
| 64 | |||
| 65 | def advanceDelay(runSpec: RunSpec): Unit = { | ||
| 66 | val key = runSpec.id -> runSpec.versionInfo.lastConfigChangeVersion | ||
| 67 | taskLaunchDelays.get(key).foreach { delay => | ||
| 68 | log.info(s"Task launch delay for [${runSpec.id} - ${runSpec.versionInfo.lastConfigChangeVersion}}] got advanced") | ||
| 69 | taskLaunchDelays += key -> Delay(clock, delay.currentDelay, delay.maxLaunchDelay) | ||
| 69 | 70 | } | |
| 71 | } | ||
| 72 | } | ||
| 70 | 73 | ||
| 71 | 74 | private object RateLimiter { | |
| 72 | 75 | private val log = LoggerFactory.getLogger(getClass.getName) | |
| 73 | 76 | ||
| 74 | 77 | private object Delay { | |
| 75 | def apply(clock: Clock, runSpec: RunSpec): Delay = Delay(clock, runSpec.backoffStrategy.backoff) | ||
| 76 | def apply(clock: Clock, delay: FiniteDuration): Delay = Delay(clock.now() + delay, delay) | ||
| 78 | def apply(clock: Clock, runSpec: RunSpec): Delay = { | ||
| 79 | val delay = runSpec.backoffStrategy.backoff min runSpec.backoffStrategy.maxLaunchDelay | ||
| 80 | Delay(clock.now(), delay, runSpec.backoffStrategy.maxLaunchDelay) | ||
| 77 | 81 | } | |
| 82 | def apply(clock: Clock, currentDelay: FiniteDuration, maxLaunchDelay: FiniteDuration): Delay = | ||
| 83 | Delay(clock.now(), currentDelay, maxLaunchDelay) | ||
| 84 | } | ||
| 78 | 85 | ||
| 79 | 86 | private case class Delay( | |
| 80 | deadline: Timestamp, | ||
| 81 | delay: FiniteDuration) { | ||
| 87 | referenceTimestamp: Timestamp, | ||
| 88 | currentDelay: FiniteDuration, | ||
| 89 | maxLaunchDelay: FiniteDuration) { | ||
| 90 | def deadline: Timestamp = referenceTimestamp + currentDelay | ||
| 82 | 91 | ||
| 83 | 92 | def increased(clock: Clock, runSpec: RunSpec): Delay = { | |
| 84 | 93 | val newDelay: FiniteDuration = | |
| 85 | 94 | runSpec.backoffStrategy.maxLaunchDelay min FiniteDuration( | |
| 86 | (delay.toNanos * runSpec.backoffStrategy.factor).toLong, TimeUnit.NANOSECONDS) | ||
| 87 | Delay(clock, newDelay) | ||
| 95 | (currentDelay.toNanos * runSpec.backoffStrategy.factor).toLong, TimeUnit.NANOSECONDS) | ||
| 96 | Delay(clock, newDelay, runSpec.backoffStrategy.maxLaunchDelay) | ||
| 88 | 97 | } | |
| 89 | 98 | } | |
| 90 | 99 | } |
| 1 | 1 | package mesosphere.marathon.core.launchqueue.impl | |
|---|---|---|---|
| 2 | 2 | ||
| 3 | 3 | import akka.actor.{ Actor, ActorLogging, ActorRef, Cancellable, Props } | |
| 4 | 4 | import akka.event.LoggingReceive | |
| 5 | import mesosphere.marathon.core.launchqueue.impl.RateLimiterActor.{ | ||
| 6 | AddDelay, | ||
| 7 | CleanupOverdueDelays, | ||
| 8 | DecreaseDelay, | ||
| 9 | DelayUpdate, | ||
| 10 | GetDelay, | ||
| 11 | ResetDelay, | ||
| 12 | ResetDelayResponse | ||
| 13 | } | ||
| 5 | import mesosphere.marathon.core.launchqueue.impl.RateLimiterActor._ | ||
| 14 | 6 | import mesosphere.marathon.state.{ RunSpec, Timestamp } | |
| 15 | 7 | ||
| 16 | 8 | import scala.concurrent.duration._ | |
| 17 | 9 | ||
| 18 | 10 | private[launchqueue] object RateLimiterActor { | |
| 19 | 11 | def props( | |
| 20 | 12 | rateLimiter: RateLimiter, | |
| 21 | 13 | launchQueueRef: ActorRef): Props = | |
| 22 | 14 | Props(new RateLimiterActor( | |
| 23 | 15 | rateLimiter, launchQueueRef | |
| 24 | 16 | )) | |
| 25 | 17 | ||
| 26 | 18 | case class DelayUpdate(runSpec: RunSpec, delayUntil: Timestamp) | |
| 27 | 19 | ||
| 28 | 20 | case class ResetDelay(runSpec: RunSpec) | |
| 29 | case object ResetDelayResponse | ||
| 30 | |||
| 31 | 21 | case class GetDelay(runSpec: RunSpec) | |
| 32 | 22 | private[impl] case class AddDelay(runSpec: RunSpec) | |
| 33 | 23 | private[impl] case class DecreaseDelay(runSpec: RunSpec) | |
| 24 | private[impl] case class AdvanceDelay(runSpec: RunSpec) | ||
| 34 | 25 | ||
| 35 | 26 | private case object CleanupOverdueDelays | |
| 36 | 27 | } | |
| 37 | 28 | ||
| 38 | 29 | private class RateLimiterActor private ( | |
| 39 | 30 | rateLimiter: RateLimiter, | |
| 40 | 31 | launchQueueRef: ActorRef) extends Actor with ActorLogging { | |
| 41 | 32 | var cleanup: Cancellable = _ | |
| Show All 13 Lines | |||
| 55 | 46 | receiveCleanup, | |
| 56 | 47 | receiveDelayOps | |
| 57 | 48 | ).reduceLeft(_.orElse[Any, Unit](_)) | |
| 58 | 49 | } | |
| 59 | 50 | ||
| 60 | 51 | private[this] def receiveCleanup: Receive = { | |
| 61 | 52 | case CleanupOverdueDelays => | |
| 62 | 53 | // If a run spec gets removed or updated, the delay should be reset. | |
| 63 | // Still, we can remove overdue delays before that and also make leaks less likely | ||
| 54 | // In addition to that we remove overdue delays to ensure there are no leaks, | ||
| 64 | 55 | // by calling this periodically. | |
| 65 | 56 | rateLimiter.cleanUpOverdueDelays() | |
| 66 | 57 | } | |
| 67 | 58 | ||
| 68 | 59 | private[this] def receiveDelayOps: Receive = { | |
| 69 | 60 | case GetDelay(runSpec) => | |
| 70 | sender() ! DelayUpdate(runSpec, rateLimiter.getDelay(runSpec)) | ||
| 61 | sender() ! DelayUpdate(runSpec, rateLimiter.getDeadline(runSpec)) | ||
| 71 | 62 | ||
| 72 | 63 | case AddDelay(runSpec) => | |
| 73 | 64 | rateLimiter.addDelay(runSpec) | |
| 74 | launchQueueRef ! DelayUpdate(runSpec, rateLimiter.getDelay(runSpec)) | ||
| 65 | launchQueueRef ! DelayUpdate(runSpec, rateLimiter.getDeadline(runSpec)) | ||
| 75 | 66 | ||
| 76 | case DecreaseDelay(runSpec) => // ignore for now | ||
| 67 | case DecreaseDelay(_) => // ignore for now | ||
| 77 | 68 | ||
| 69 | case AdvanceDelay(runSpec) => | ||
| 70 | rateLimiter.advanceDelay(runSpec) | ||
| 71 | launchQueueRef ! DelayUpdate(runSpec, rateLimiter.getDeadline(runSpec)) | ||
| 72 | |||
| 78 | 73 | case ResetDelay(runSpec) => | |
| 79 | 74 | rateLimiter.resetDelay(runSpec) | |
| 80 | launchQueueRef ! DelayUpdate(runSpec, rateLimiter.getDelay(runSpec)) | ||
| 81 | sender() ! ResetDelayResponse | ||
| 75 | launchQueueRef ! DelayUpdate(runSpec, rateLimiter.getDeadline(runSpec)) | ||
| 82 | 76 | } | |
| 83 | 77 | } | |
| 1 | 1 | package mesosphere.marathon | |
|---|---|---|---|
| 2 | 2 | package core.task.update.impl.steps | |
| 3 | 3 | ||
| 4 | 4 | import java.time.OffsetDateTime | |
| 5 | 5 | ||
| 6 | 6 | import akka.Done | |
| 7 | 7 | import com.google.inject.{ Inject, Provider } | |
| 8 | 8 | import mesosphere.marathon.core.condition.Condition | |
| 9 | 9 | import mesosphere.marathon.core.group.GroupManager | |
| 10 | 10 | import mesosphere.marathon.core.instance.update.{ InstanceChange, InstanceChangeHandler } | |
| 11 | 11 | import mesosphere.marathon.core.launchqueue.LaunchQueue | |
| 12 | import mesosphere.marathon.state.PathId | ||
| 13 | import scala.async.Async._ | ||
| 12 | import mesosphere.marathon.state.{ PathId, RunSpec } | ||
| 14 | 13 | ||
| 14 | import scala.async.Async._ | ||
| 15 | 15 | import scala.concurrent.Future | |
| 16 | 16 | ||
| 17 | 17 | class NotifyRateLimiterStepImpl @Inject() ( | |
| 18 | 18 | launchQueueProvider: Provider[LaunchQueue], | |
| 19 | 19 | groupManagerProvider: Provider[GroupManager]) extends InstanceChangeHandler { | |
| 20 | 20 | ||
| 21 | 21 | import NotifyRateLimiterStep._ | |
| 22 | 22 | import scala.concurrent.ExecutionContext.Implicits.global | |
| 23 | 23 | ||
| 24 | 24 | private[this] lazy val launchQueue = launchQueueProvider.get() | |
| 25 | 25 | private[this] lazy val groupManager = groupManagerProvider.get() | |
| 26 | 26 | ||
| 27 | 27 | override def name: String = "notifyRateLimiter" | |
| 28 | 28 | ||
| 29 | 29 | override def process(update: InstanceChange): Future[Done] = { | |
| 30 | if (limitWorthy(update.condition)) { | ||
| 31 | notifyRateLimiter(update.runSpecId, update.instance.runSpecVersion.toOffsetDateTime) | ||
| 32 | } else { | ||
| 30 | update.condition match { | ||
| 31 | case condition if limitWorthy(condition) => | ||
| 32 | notifyRateLimiter(update.runSpecId, update.instance.runSpecVersion.toOffsetDateTime, launchQueue.addDelay) | ||
| 33 | case condition if advanceWorthy(condition) => | ||
| 34 | notifyRateLimiter(update.runSpecId, update.instance.runSpecVersion.toOffsetDateTime, launchQueue.advanceDelay) | ||
| 35 | case _ => | ||
| 33 | 36 | Future.successful(Done) | |
| 34 | 37 | } | |
| 35 | 38 | } | |
| 36 | 39 | ||
| 37 | 40 | @SuppressWarnings(Array("all")) // async/await | |
| 38 | private[this] def notifyRateLimiter(runSpecId: PathId, version: OffsetDateTime): Future[Done] = async { | ||
| 41 | private[this] def notifyRateLimiter(runSpecId: PathId, version: OffsetDateTime, fn: RunSpec => Unit): Future[Done] = | ||
| 42 | async { | ||
| 39 | 43 | val appFuture = groupManager.appVersion(runSpecId, version) | |
| 40 | 44 | val podFuture = groupManager.podVersion(runSpecId, version) | |
| 41 | 45 | val (app, pod) = (await(appFuture), await(podFuture)) | |
| 42 | app.foreach(launchQueue.addDelay) | ||
| 43 | pod.foreach(launchQueue.addDelay) | ||
| 46 | app.foreach(fn) | ||
| 47 | pod.foreach(fn) | ||
| 44 | 48 | Done | |
| 45 | 49 | } | |
| 46 | 50 | } | |
| 47 | 51 | ||
| 48 | 52 | private[steps] object NotifyRateLimiterStep { | |
| 49 | 53 | // A set of conditions that are worth rate limiting the associated runSpec | |
| 50 | 54 | val limitWorthy: Set[Condition] = Set( | |
| 51 | 55 | Condition.Dropped, Condition.Error, Condition.Failed, Condition.Gone, Condition.Finished | |
| 56 | ) | ||
| 57 | |||
| 58 | // A set of conditions that are worth advancing an existing delay of the corresponding runSpec | ||
| 59 | val advanceWorthy: Set[Condition] = Set( | ||
| 60 | Condition.Staging, Condition.Starting, Condition.Running, Condition.Created | ||
| 52 | 61 | ) | |
| 53 | 62 | } |
| Show All 11 Lines | |||
| 12 | 12 | val clock = ConstantClock(Timestamp.now()) | |
|---|---|---|---|
| 13 | 13 | ||
| 14 | 14 | test("addDelay") { | |
| 15 | 15 | val limiter = new RateLimiter(clock) | |
| 16 | 16 | val app = AppDefinition(id = "test".toPath, backoffStrategy = BackoffStrategy(backoff = 10.seconds)) | |
| 17 | 17 | ||
| 18 | 18 | limiter.addDelay(app) | |
| 19 | 19 | ||
| 20 | limiter.getDelay(app) should be(clock.now() + 10.seconds) | ||
| 20 | limiter.getDeadline(app) should be(clock.now() + 10.seconds) | ||
| 21 | 21 | } | |
| 22 | 22 | ||
| 23 | 23 | test("addDelay for existing delay") { | |
| 24 | 24 | val limiter = new RateLimiter(clock) | |
| 25 | 25 | val app = AppDefinition(id = "test".toPath, backoffStrategy = BackoffStrategy(backoff = 10.seconds, factor = 2.0)) | |
| 26 | 26 | ||
| 27 | 27 | limiter.addDelay(app) // linter:ignore:IdenticalStatements | |
| 28 | 28 | limiter.addDelay(app) | |
| 29 | 29 | ||
| 30 | limiter.getDelay(app) should be(clock.now() + 20.seconds) | ||
| 30 | limiter.getDeadline(app) should be(clock.now() + 20.seconds) | ||
| 31 | 31 | } | |
| 32 | 32 | ||
| 33 | 33 | test("cleanupOverdueDelays") { | |
| 34 | val time_origin = clock.now() | ||
| 34 | 35 | val limiter = new RateLimiter(clock) | |
| 35 | val overdue = AppDefinition(id = "overdue".toPath, backoffStrategy = BackoffStrategy(backoff = 10.seconds)) | ||
| 36 | limiter.addDelay(overdue) | ||
| 37 | val stillWaiting = AppDefinition(id = "test".toPath, backoffStrategy = BackoffStrategy(backoff = 20.seconds)) | ||
| 38 | limiter.addDelay(stillWaiting) | ||
| 36 | val threshold = 60.seconds | ||
| 39 | 37 | ||
| 40 | clock += 11.seconds | ||
| 38 | val appWithOverdueDelay = AppDefinition( | ||
| 39 | id = "overdue".toPath, | ||
| 40 | backoffStrategy = BackoffStrategy(backoff = 10.seconds, maxLaunchDelay = threshold)) | ||
| 41 | limiter.addDelay(appWithOverdueDelay) | ||
| 42 | val appWithValidDelay = AppDefinition( | ||
| 43 | id = "valid".toPath, | ||
| 44 | backoffStrategy = BackoffStrategy(backoff = 20.seconds, maxLaunchDelay = threshold + 10.seconds)) | ||
| 45 | limiter.addDelay(appWithValidDelay) | ||
| 41 | 46 | ||
| 47 | // after advancing the clock by (threshold + 1), the existing delays | ||
| 48 | // with maxLaunchDelay < (threshold + 1) should be gone | ||
| 49 | clock += threshold + 1.seconds | ||
| 42 | 50 | limiter.cleanUpOverdueDelays() | |
| 43 | |||
| 44 | limiter.getDelay(overdue) should be(clock.now()) | ||
| 45 | limiter.getDelay(stillWaiting) should be(clock.now() + 9.seconds) | ||
| 51 | limiter.getDeadline(appWithOverdueDelay) should be(clock.now()) | ||
| 52 | limiter.getDeadline(appWithValidDelay) should be(time_origin + 20.seconds) | ||
| 46 | 53 | } | |
| 47 | 54 | ||
| 48 | 55 | test("resetDelay") { | |
| 49 | 56 | val limiter = new RateLimiter(clock) | |
| 50 | 57 | val app = AppDefinition(id = "test".toPath, backoffStrategy = BackoffStrategy(backoff = 10.seconds)) | |
| 51 | 58 | ||
| 52 | 59 | limiter.addDelay(app) | |
| 53 | |||
| 54 | 60 | limiter.resetDelay(app) | |
| 55 | 61 | ||
| 56 | limiter.getDelay(app) should be(clock.now()) | ||
| 62 | limiter.getDeadline(app) should be(clock.now()) | ||
| 57 | 63 | } | |
| 58 | 64 | ||
| 59 | 65 | } | |