HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D323

The LaunchQueueActor now forwards InstanceUpdate Events and sends Done in case of a non existing actor.
ClosedAll Users
Actions

Authored by aquamatthias on Dec 13 2016, 9:22 AM.

Details

Reviewers
unterstein
meichstedt
Commits
rMARATHON11d6261ec186: The LaunchQueueActor now forwards InstanceUpdate Events and sends Done in case…
rMARATHONd7bb4edd4641: The LaunchQueueActor now forwards InstanceUpdate Events and sends Done in case…
Summary

The LaunchQueueActor still has expected an QueuedInstanceInfo when talking to the TaskLauncherActor for InstanceUpdate.

Test Plan

sbt integration:test

Diff Detail

Repository
rMARATHON marathon
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.
aquamatthias retitled this revision from to The LaunchQueueActor now forwards InstanceUpdate Events and sends Done in case of a non existing actor..Dec 13 2016, 9:22 AM
aquamatthias updated this object.
aquamatthias edited the test plan for this revision. (Show Details)
aquamatthias added reviewers: meichstedt, unterstein.
meichstedt added a comment.Dec 13 2016, 11:02 AM

LGTM. A test would be awesome but doesn't block landing this for RC3 imo.

src/main/scala/mesosphere/marathon/core/launchqueue/impl/LaunchQueueActor.scala
148–149

How about a test that verifies that the TaskLauncherActor will answer with Done?

src/main/scala/mesosphere/marathon/core/launchqueue/impl/LaunchQueueDelegate.scala
26

thanks for extracting this

meichstedt accepted this revision.Dec 13 2016, 11:03 AM
This revision is now accepted and ready to land.Dec 13 2016, 11:03 AM
unterstein accepted this revision.Dec 13 2016, 11:31 AM
Closed by commit rMARATHONd7bb4edd4641: The LaunchQueueActor now forwards InstanceUpdate Events and sends Done in case… (authored by aquamatthias, committed by unterstein). · Explain WhyDec 13 2016, 12:03 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents

Diff 1250

View Options

src/main/scala/mesosphere/marathon/core/launchqueue/impl/LaunchQueueActor.scala

11package mesosphere.marathon
22package core.launchqueue.impl
33
44import akka.actor.SupervisorStrategy.Stop
55import akka.actor.{ Actor, ActorLogging, ActorRef, OneForOneStrategy, Props, SupervisorStrategy, Terminated }
66import akka.event.LoggingReceive
77import akka.pattern.{ ask, pipe }
88import akka.util.Timeout
99import mesosphere.marathon.core.launchqueue.{ LaunchQueue, LaunchQueueConfig }
1010import mesosphere.marathon.state.{ PathId, RunSpec }
1111import LaunchQueue.QueuedInstanceInfo
12import akka.Done
1213import mesosphere.marathon.core.instance.update.InstanceChange
1314
1415import scala.concurrent.Future
1516import scala.concurrent.duration._
1617import scala.util.control.NonFatal
1718
1819private[launchqueue] object LaunchQueueActor {
1920 def props(
▲ Show 20 Lines • Show All 118 Lines • ▼ Show 20 Line(s)
138139 val actorRef = launchers(appId)
139140 val deferredMessages: Vector[DeferredMessage] =
140141 suspendedLaunchersMessages(actorRef) :+ DeferredMessage(sender(), msg)
141142 suspendedLaunchersMessages += actorRef -> deferredMessages
142143 }
143144
144145 private[this] def receiveInstanceUpdate: Receive = {
145146 case update: InstanceChange =>
146 import context.dispatcher
147147 launchers.get(update.runSpecId) match {
148 case Some(actorRef) =>
149 val eventualCount: Future[QueuedInstanceInfo] =
148 case Some(actorRef) => actorRef.forward(update)
149 case None => sender() ! Done
meichstedtUnsubmitted
Not Done

How about a test that verifies that the TaskLauncherActor will answer with Done?

150 (actorRef ? update).mapTo[QueuedInstanceInfo]
151 eventualCount.map(Some(_)).pipeTo(sender())
152 case None => sender() ! None
153150 }
154151 }
155152
156153 private[this] def list(): Future[Seq[QueuedInstanceInfo]] = {
157154 import context.dispatcher
158155 val scatter = launchers
159156 .keys
160157 .map(appId => (self ? Count(appId)).mapTo[Option[QueuedInstanceInfo]])
▲ Show 20 Lines • Show All 63 Lines • Show Last 20 Lines
View Options

src/main/scala/mesosphere/marathon/core/launchqueue/impl/LaunchQueueDelegate.scala

11package mesosphere.marathon.core.launchqueue.impl
22
33import akka.Done
44import akka.actor.ActorRef
55import akka.pattern.ask
66import akka.util.Timeout
7import com.typesafe.scalalogging.StrictLogging
78import mesosphere.marathon.core.instance.update.InstanceChange
89import mesosphere.marathon.core.launchqueue.LaunchQueue.{ QueuedInstanceInfo, QueuedInstanceInfoWithStatistics }
910import mesosphere.marathon.core.launchqueue.{ LaunchQueue, LaunchQueueConfig }
1011import mesosphere.marathon.state.{ PathId, RunSpec }
1112
1213import scala.collection.immutable.Seq
1314import scala.concurrent.duration._
1415import scala.concurrent.{ Await, Future }
1516import scala.reflect.ClassTag
1617import scala.util.control.NonFatal
1718
1819private[launchqueue] class LaunchQueueDelegate(
1920 config: LaunchQueueConfig,
2021 actorRef: ActorRef,
21 rateLimiterRef: ActorRef) extends LaunchQueue {
22 rateLimiterRef: ActorRef) extends LaunchQueue with StrictLogging {
2223
24 // When purging, we wait for the TaskLauncherActor to shut down. This actor will wait for
25 // in-flight task op notifications before complying, therefore we need to adjust the timeout accordingly.
26 val purgeTimeout: Timeout = config.launchQueueRequestTimeout().milliseconds + config.taskOpNotificationTimeout().millisecond
meichstedtUnsubmitted
Not Done

thanks for extracting this

27
28 val launchQueueRequestTimeout: Timeout = config.launchQueueRequestTimeout().milliseconds
29
2330 override def list: Seq[QueuedInstanceInfo] = {
2431 askQueueActor[LaunchQueueDelegate.Request, Seq[QueuedInstanceInfo]]("list")(LaunchQueueDelegate.List)
2532 }
2633
2734 override def listWithStatistics: Seq[QueuedInstanceInfoWithStatistics] = {
2835 askQueueActor[LaunchQueueDelegate.Request, Seq[QueuedInstanceInfoWithStatistics]]("listWithStatistics")(LaunchQueueDelegate.ListWithStatistics)
2936 }
3037
3138 override def get(runSpecId: PathId): Option[QueuedInstanceInfo] =
3239 askQueueActor[LaunchQueueDelegate.Request, Option[QueuedInstanceInfo]]("get")(LaunchQueueDelegate.Count(runSpecId))
3340
3441 override def notifyOfInstanceUpdate(update: InstanceChange): Future[Done] =
3542 askQueueActorFuture[InstanceChange, Done]("notifyOfInstanceUpdate")(update)
3643
3744 override def count(runSpecId: PathId): Int = get(runSpecId).map(_.instancesLeftToLaunch).getOrElse(0)
3845
3946 override def listRunSpecs: Seq[RunSpec] = list.map(_.runSpec)
4047
4148 override def purge(runSpecId: PathId): Unit = {
42 // When purging, we wait for the TaskLauncherActor to shut down. This actor will wait for
43 // in-flight task op notifications before complying, therefore we need to adjust the timeout accordingly.
44 val purgeTimeout = config.launchQueueRequestTimeout().milliseconds + config.taskOpNotificationTimeout().millisecond
4549 askQueueActor[LaunchQueueDelegate.Request, Unit]("purge", timeout = purgeTimeout)(LaunchQueueDelegate.Purge(runSpecId))
4650 }
4751
4852 override def add(runSpec: RunSpec, count: Int): Unit = askQueueActor[LaunchQueueDelegate.Request, Unit]("add")(LaunchQueueDelegate.Add(runSpec, count))
4953
5054 private[this] def askQueueActor[T, R: ClassTag](
5155 method: String,
52 timeout: FiniteDuration = config.launchQueueRequestTimeout().milliseconds)(message: T): R = {
56 timeout: Timeout = launchQueueRequestTimeout)(message: T): R = {
5357
5458 val answerFuture = askQueueActorFuture[T, R](method, timeout)(message)
55 Await.result(answerFuture, timeout)
59 Await.result(answerFuture, timeout.duration)
5660 }
5761
5862 private[this] def askQueueActorFuture[T, R: ClassTag](
5963 method: String,
60 timeout: FiniteDuration = config.launchQueueRequestTimeout().milliseconds)(message: T): Future[R] = {
64 timeout: Timeout = launchQueueRequestTimeout)(message: T): Future[R] = {
6165
6266 implicit val timeoutImplicit: Timeout = timeout
6367 val answerFuture = actorRef ? message
6468 import scala.concurrent.ExecutionContext.Implicits.global
6569 answerFuture.recover {
6670 case NonFatal(e) => throw new RuntimeException(s"in $method", e)
6771 }
6872 answerFuture.mapTo[R]
Show All 16 Lines