HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D853

Added async methods to LaunchQueueDelegate interface and used them in deployments
ClosedPublic
Actions

Authored by zen-dog on Jun 14 2017, 5:33 PM.

Details

Reviewers
jeschkies
unterstein
timcharper
jdef
meichstedt
jenkins
Commits
rMARATHONae563c83d524: Added async methods to LaunchQueueDelegate interface and used them in…
rMARATHON541db8de8944: Added async methods to LaunchQueueDelegate interface and used them in…
rMARATHONa5460bbedf09: Added async methods to LaunchQueueDelegate interface and used them in…
rMARATHON8dddeedbf963: Added async methods to LaunchQueueDelegate interface and used them in…
rMARATHON97c751e68822: Added async methods to LaunchQueueDelegate interface and used them in…
rMARATHON23df6002da8b: Async LaunchQueue methods are now used in deployments
rMARATHON8e7e5309d219: Added async methods to LaunchQueueDelegate interface and used them in…
rMARATHON5bb5e39ca471: Added async methods to LaunchQueueDelegate interface and used them in…
rMARATHONbb618473da0d: Added async methods to LaunchQueueDelegate interface and used them in…
JIRA Issues
JIRA MARATHON-7358 Deployments performance degradation in 1.4
JIRA MARATHON-7472 Deployment never ends due to a failure in TaskStartActor initialisation
Summary
  • Every synchronous methods in the LaunchQueue has now it's async counterpart.
  • Replaced calling LaunchQueue sync methods with their respective async counterparts in deployment actors so they don't block any more.
  • Fixed relevant deployment tests
  • Added a supervision strategy to DeploymentActor that restart child actors even on ActorInitializationException
Test Plan

unit

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.
There are a very large number of changes, so older changes are hidden. Show Older Changes
zen-dog added inline comments.Jun 21 2017, 4:56 PM
src/main/scala/mesosphere/marathon/core/launchqueue/LaunchQueue.scala
51

Not that I know of. I could create one, but since this part (along with deployments) is probably part of Action/v3/what ever, this ticket would never be touched.

src/main/scala/mesosphere/marathon/core/task/tracker/impl/InstanceTrackerDelegate.scala
46

No, it conflicts with implicit ec: ExecutionContext used below.

src/test/scala/mesosphere/marathon/core/deployment/impl/TaskStartActorTest.scala
35

No, that's a different promise - DeploymentActor becomes this one and completes it when the deployment is done. It should be also Promise[Done] but that has nothing to do with current diff. I can make it a separate one.

zen-dog updated this revision to Diff 3589.Jun 21 2017, 4:56 PM
zen-dog marked an inline comment as done.
  • Third feedback round
jdef added inline comments.Jun 21 2017, 5:00 PM
src/main/scala/mesosphere/marathon/core/deployment/impl/DeploymentActor.scala
5

eh?

64

maybe check for NonFatal instead of catching everything here?

src/main/scala/mesosphere/marathon/core/task/tracker/impl/InstanceTrackerDelegate.scala
46

I thought we were supposed to standardize on the marathon core async global execution context in order for our MDC/actions context propagation to work properly. this looks like a red flag. what am I missing?

src/test/scala/mesosphere/marathon/core/deployment/impl/TaskStartActorTest.scala
255

NonFatal?

jenkins requested changes to this revision.Jun 21 2017, 5:15 PM
This revision now requires changes to proceed.Jun 21 2017, 5:15 PM
jenkins added a comment.Jun 21 2017, 5:15 PM
✗ Build of 3589 failed jenkins-public-marathon-phabricator-319.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

jenkins added a comment.Jun 21 2017, 5:29 PM
✗ Build of 3588 failed jenkins-public-marathon-phabricator-320.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

zen-dog updated this revision to Diff 3595.Jun 21 2017, 6:23 PM
zen-dog marked 2 inline comments as done.
  • Fourth round of feedback
src/main/scala/mesosphere/marathon/core/task/tracker/impl/InstanceTrackerDelegate.scala
46

Everbody is using our custom global execution context (linter rejects the scala's one) but if you import it here at the top of the file you'll get:

[error] /Users/adukhovniy/mesosphere/marathon/src/main/scala/mesosphere/marathon/core/task/tracker/impl/InstanceTrackerDelegate.scala:33: ambiguous implicit values:
[error]  both value ec of type scala.concurrent.ExecutionContext
[error]  and value global in object ExecutionContexts of type => scala.concurrent.ExecutionContext
...
jenkins requested changes to this revision.Jun 21 2017, 6:37 PM
This revision now requires changes to proceed.Jun 21 2017, 6:37 PM
jenkins added a comment.Jun 21 2017, 6:37 PM
✗ Build of 3595 failed jenkins-public-marathon-phabricator-326.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

jdef accepted this revision.Jun 21 2017, 7:00 PM
zen-dog updated this revision to Diff 3608.Jun 22 2017, 2:49 PM
zen-dog marked an inline comment as done.
  • Adding debug test logging to figure out AppDeployIntegrationTest failures and some minor improvements
zen-dog added inline comments.Jun 22 2017, 2:50 PM
src/main/scala/mesosphere/marathon/core/deployment/impl/TaskReplaceActor.scala
135–145

On a second thought - I realized yesterday why I didn't make the method completely async in the first place: method changes instancesStarted (actor's state) and doing it from an async block will create a race condition. I could make it Atomic of course but prefer not to. Reverted to first implementation.

jenkins accepted this revision.Jun 22 2017, 3:15 PM
This revision is now accepted and ready to land.Jun 22 2017, 3:15 PM
jenkins added a comment.Jun 22 2017, 3:15 PM
✔ Build of 3608 completed jenkins-public-marathon-phabricator-335.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-594-g197d5f5.tgz",
"sha1"" "48c0dfed54da6f430f7bf0b8603b79c171832f59"

＼\ ٩( ᐛ )و /／

jenkins added a comment.Jun 22 2017, 3:20 PM
✔ Build of 3608 completed jenkins-public-marathon-phabricator-336.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-594-g197d5f5.tgz",
"sha1"" "6ffd649de7558f44756cbd481c706c1dae1e8f73"

＼\ ٩( ᐛ )و /／

jenkins added a comment.Jun 22 2017, 3:36 PM
✔ Build of 3608 completed jenkins-public-marathon-phabricator-337.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-594-g197d5f5.tgz",
"sha1"" "9f71f447c17bc3ec8d30c632a41d4d8d24e40e23"

＼\ ٩( ᐛ )و /／

jdef added inline comments.Jun 22 2017, 3:43 PM
src/main/scala/mesosphere/marathon/core/deployment/impl/TaskReplaceActor.scala
133

thanks for this

zen-dog updated this revision to Diff 3610.Jun 22 2017, 3:57 PM
  • Setting the logging level back to INFO
zen-dog added a comment.Jun 22 2017, 3:58 PM

I don't know if that's the extra logging (on DEBUG level) but AppDeployIntegrationTest doesn't fail anymore :(

jeschkies added inline comments.Jun 22 2017, 4:02 PM
src/main/scala/mesosphere/marathon/core/deployment/impl/TaskReplaceActor.scala
135–145

Hm, Why is it return a Future[Done]? It will alway finished immediately without waiting for the addAsync. Is this what we want?

jenkins accepted this revision.Jun 22 2017, 4:14 PM
✔ Build of 3610 completed jenkins-public-marathon-phabricator-339.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-594-g197d5f5.tgz",
"sha1"" "5007fd2138789f182ef08ef100479a026db22ebe"

＼\ ٩( ᐛ )و /／

zen-dog updated this revision to Diff 3612.Jun 22 2017, 4:24 PM
  • A minor issue in TaskReplaceActor fixed
zen-dog added inline comments.Jun 22 2017, 4:26 PM
src/main/scala/mesosphere/marathon/core/deployment/impl/TaskReplaceActor.scala
135–145

You're right. Should be if else. Done

jenkins accepted this revision.Jun 22 2017, 4:40 PM
✔ Build of 3612 completed jenkins-public-marathon-phabricator-340.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-594-g197d5f5.tgz",
"sha1"" "8cc026e0597181c28596b4bad9eeac0a14a04c42"

＼\ ٩( ᐛ )و /／

jenkins requested changes to this revision.Jun 22 2017, 5:54 PM
This revision now requires changes to proceed.Jun 22 2017, 5:54 PM
jenkins added a comment.Jun 22 2017, 5:54 PM
✗ Build of 3612 failed jenkins-public-marathon-phabricator-341.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

zen-dog updated this revision to Diff 3614.Jun 22 2017, 7:06 PM
  • Trying to make tests great again
zen-dog added a comment.Jun 22 2017, 7:20 PM

It seems like at least a few tests AppDeployIntegrationTest are flaky. I'll keep iterating because I don't want to introduce new flaky tests. I'll set up a test loop here: Marathon/job/marathon-sandbox/job/marathon-loop-ad-async-launch-queue/ and will gather statistics tomorrow.

jenkins accepted this revision.Jun 22 2017, 7:31 PM
This revision is now accepted and ready to land.Jun 22 2017, 7:31 PM
jenkins added a comment.Jun 22 2017, 7:31 PM
✔ Build of 3614 completed jenkins-public-marathon-phabricator-342.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-594-g197d5f5.tgz",
"sha1"" "d318228b94e56e12b6d29295d5b0fe06ff2cf091"

＼\ ٩( ᐛ )و /／

jenkins added a comment.Jun 22 2017, 7:34 PM
✔ Build of 3612 completed jenkins-public-marathon-phabricator-343.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-594-g197d5f5.tgz",
"sha1"" "46123f8ea8cd73c0af885cd84d812e08e553fe89"

＼\ ٩( ᐛ )و /／

zen-dog updated this revision to Diff 3618.Jun 23 2017, 2:03 PM
  • Fixed a flaky TaskStartActorTest and added a few comments
jenkins requested changes to this revision.Jun 23 2017, 2:14 PM
This revision now requires changes to proceed.Jun 23 2017, 2:14 PM
jenkins added a comment.Jun 23 2017, 2:14 PM
✗ Build of 3618 failed jenkins-public-marathon-phabricator-347.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

jenkins accepted this revision.Jun 23 2017, 2:35 PM
This revision is now accepted and ready to land.Jun 23 2017, 2:35 PM
jenkins added a comment.Jun 23 2017, 2:35 PM
✔ Build of 3618 completed jenkins-public-marathon-phabricator-348.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-594-g197d5f5.tgz",
"sha1"" "7384a0a6f54a84f0aa17db10277f6ecdd3e5d8d0"

＼\ ٩( ᐛ )و /／

zen-dog updated this revision to Diff 3619.Jun 23 2017, 3:10 PM
  • ...and more veryfing in TaskStartActorTest
jenkins requested changes to this revision.Jun 23 2017, 3:24 PM
This revision now requires changes to proceed.Jun 23 2017, 3:24 PM
jenkins added a comment.Jun 23 2017, 3:24 PM
✗ Build of 3619 failed jenkins-public-marathon-phabricator-349.

Error message:

Stage Compile and Test failed.

(๑′°︿°๑)

jenkins accepted this revision.Jun 25 2017, 11:39 PM
This revision is now accepted and ready to land.Jun 25 2017, 11:39 PM
jenkins added a comment.Jun 25 2017, 11:39 PM
✔ Build of 3619 completed jenkins-public-marathon-phabricator-354.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-594-g197d5f5.tgz",
"sha1"" "ce64d62191fe40872e314c699b4333681ee29642"

＼\ ٩( ᐛ )و /／

zen-dog updated this revision to Diff 3634.Jun 27 2017, 12:01 PM

Rebased on master and moved IT stabilizing commits to a separate diff

jenkins accepted this revision.Jun 27 2017, 12:21 PM
✔ Build of 3634 completed jenkins-public-marathon-phabricator-361.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-618-g48bfffd.tgz",
"sha1": "ad5182db4e3ff2d1e3084fc9dde27e2eef9817c2"

＼\ ٩( ᐛ )و /／

jeschkies accepted this revision.Jun 27 2017, 12:26 PM

Nice. Let's go for it.

Closed by commit rMARATHON8e7e5309d219: Added async methods to LaunchQueueDelegate interface and used them in… (authored by zen-dog). · Explain WhyJun 27 2017, 12:37 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents

Diff 3635

View Options

project/BasicLintingPlugin.scala

Show First 20 Lines • Show All 43 Lines • ▼ Show 20 Line(s)
4444 |package ${pkg.replaceAll(s"package $doublePackageRoot.", "")}
4545 """.stripMargin)
4646 packageError = true
4747 }
4848 }
4949
5050 if (file.getName != "ExecutionContexts.scala" && (line.contains("ExecutionContext.global") || line.contains("ExecutionContext.Implicits"))) {
5151 logger.error(
52 s"$file:$lineNumber uses scala's global execution context. Please use mesosphere.marathon.async.ExecutionContexts.*"
52 s"$file:$lineNumber uses scala's global execution context. Please use mesosphere.marathon.core.async.ExecutionContexts.*"
5353 )
5454 otherLintError = true
5555 }
5656
5757 }
5858 }
5959 if (packageError) {
6060 sys.error("One or more files didn't use proper double package notation.")
6161 }
6262 if (otherLintError) {
6363 sys.error("Other lint errors found.")
6464 }
6565 }
6666
6767}
6868No newline at end of file
View Options

src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala

Show First 20 Lines • Show All 369 Lines • ▼ Show 20 Line(s)
370370 healthCheckManager: HealthCheckManager,
371371 instanceTracker: InstanceTracker,
372372 launchQueue: LaunchQueue,
373373 eventBus: EventStream,
374374 val killService: KillService)(implicit ec: ExecutionContext) extends StrictLogging {
375375
376376 // TODO move stuff below out of the scheduler
377377
378 def startRunSpec(runSpec: RunSpec): Unit = {
378 def startRunSpec(runSpec: RunSpec): Future[Done] = {
379379 logger.info(s"Starting runSpec ${runSpec.id}")
380380 scale(runSpec)
381381 }
382382
383383 @SuppressWarnings(Array("all")) // async/await
384384 def stopRunSpec(runSpec: RunSpec): Future[Done] = {
385385 healthCheckManager.removeAllFor(runSpec.id)
386386
▲ Show 20 Lines • Show All 170 Lines • Show Last 20 Lines
View Options

src/main/scala/mesosphere/marathon/core/deployment/impl/AppStartActor.scala

11package mesosphere.marathon
22package core.deployment.impl
33
44import akka.Done
55import akka.actor._
66import akka.event.EventStream
77import com.typesafe.scalalogging.StrictLogging
88import mesosphere.marathon.core.event.DeploymentStatus
99import mesosphere.marathon.core.instance.Instance
1010import mesosphere.marathon.core.launchqueue.LaunchQueue
1111import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1212import mesosphere.marathon.core.task.tracker.InstanceTracker
1313import mesosphere.marathon.state.RunSpec
1414
15import scala.concurrent.Promise
15import scala.concurrent.{ Future, Promise }
1616
1717class AppStartActor(
1818 val deploymentManager: ActorRef,
1919 val status: DeploymentStatus,
2020 val scheduler: SchedulerActions,
2121 val launchQueue: LaunchQueue,
2222 val instanceTracker: InstanceTracker,
2323 val eventBus: EventStream,
2424 val readinessCheckExecutor: ReadinessCheckExecutor,
2525 val runSpec: RunSpec,
2626 val scaleTo: Int,
2727 currentInstances: Seq[Instance],
2828 promise: Promise[Unit]) extends Actor with StartingBehavior with StrictLogging {
2929
30 override val nrToStart: Int = scaleTo
30 override val nrToStart = Future.successful(scaleTo)
3131
32 override def initializeStart(): Unit = {
32 override def initializeStart(): Future[Done] = {
3333 // In case we already have running instances (can happen on master abdication during deployment)
3434 // with the correct version those will not be killed.
3535 val runningInstances = currentInstances.count(_.isActive)
36 scheduler.startRunSpec(runSpec.withInstances(Math.max(runningInstances, nrToStart)))
37 Done
36 scheduler.startRunSpec(runSpec.withInstances(Math.max(runningInstances, scaleTo)))
3837 }
jeschkiesUnsubmitted
Done

Is async required? Does currentInstances.count take long?

zen-dogAuthorUnsubmitted
Not Done

currentInstances is a Seq[Instance]. the async part is actually not needed since scheduler.startRunSpec now returns Future[Done]

3938
4039 override def postStop(): Unit = {
4140 eventBus.unsubscribe(self)
4241 super.postStop()
4342 }
4443
4544 def success(): Unit = {
4645 logger.info(s"Successfully started $scaleTo instances of ${runSpec.id}")
47 promise.success(())
46 // Since a lot of StartingBehavior and this actor's code happens asynchronously now
47 // it can happen that this promise might succeed twice.
48 promise.trySuccess(())
4849 context.stop(self)
4950 }
5051}
5152
5253object AppStartActor {
5354 @SuppressWarnings(Array("MaxParameters"))
5455 def props(
5556 deploymentManager: ActorRef,
Show All 14 Lines
View Options

src/main/scala/mesosphere/marathon/core/deployment/impl/DeploymentActor.scala

11package mesosphere.marathon
22package core.deployment.impl
33
44import akka.Done
55import akka.actor._
jdefUnsubmitted
Done

eh?

66import akka.event.EventStream
77import com.typesafe.scalalogging.StrictLogging
88import mesosphere.marathon.core.deployment._
99import mesosphere.marathon.core.deployment.impl.DeploymentActor.{ Cancel, Fail, NextStep }
1010import mesosphere.marathon.core.deployment.impl.DeploymentManagerActor.DeploymentFinished
1111import mesosphere.marathon.core.event.{ DeploymentStatus, DeploymentStepFailure, DeploymentStepSuccess }
1212import mesosphere.marathon.core.health.HealthCheckManager
1313import mesosphere.marathon.core.instance.Instance
1414import mesosphere.marathon.core.launchqueue.LaunchQueue
1515import mesosphere.marathon.core.pod.PodDefinition
1616import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1717import mesosphere.marathon.core.task.termination.{ KillReason, KillService }
1818import mesosphere.marathon.core.task.tracker.InstanceTracker
1919import mesosphere.marathon.state.{ AppDefinition, RunSpec }
2020import mesosphere.mesos.Constraints
2121
2222import scala.async.Async._
2323import scala.concurrent.{ Future, Promise }
24import scala.util.control.NonFatal
2425import scala.util.{ Failure, Success }
2526
2627private class DeploymentActor(
2728 deploymentManager: ActorRef,
2829 promise: Promise[Done],
2930 killService: KillService,
3031 scheduler: SchedulerActions,
3132 plan: DeploymentPlan,
3233 instanceTracker: InstanceTracker,
3334 launchQueue: LaunchQueue,
3435 healthCheckManager: HealthCheckManager,
3536 eventBus: EventStream,
3637 readinessCheckExecutor: ReadinessCheckExecutor) extends Actor with StrictLogging {
3738
3839 import context.dispatcher
3940
4041 val steps = plan.steps.iterator
4142 var currentStepNr: Int = 0
4243
44 // Default supervision strategy is overridden here to restart deployment child actors (responsible for individual
45 // deployment steps e.g. AppStartActor, TaskStartActor etc.) even if any exception occurs (even during initialisation).
46 // This is due to the fact that child actors tend to gather information during preStart about the tasks that are
47 // already running from the TaskTracker and LaunchQueue and those calls can timeout. In general deployment child
48 // actors are built idempotent which should make restarting them possible.
49 // Additionally a BackOffSupervisor is used to make sure child actor failures are not overloading other parts of the system
50 // (like LaunchQueue and InstanceTracker) and are not filling the log with exceptions.
51 import scala.concurrent.duration._
jdefUnsubmitted
Done

is this too general? could we limit this to only things that are likely (such as timeouts)? concerned about infinite crashlooping in the event there's some unexpected (data?) problem that consistently crashes actor init.

even for timeouts, is it worth considering some kind of time-delay backoff strategy here? or would that be handled implicitly by the backoff* parameters of the app/pod? (I forget where that backoff is actually implemented)

zen-dogAuthorUnsubmitted
Not Done

I was thinking about this at first too, but marathon now doesn't have a concept of "failed deployment" - it will retry forever until someone cancels it. And if one of the DeploymentActor child actor stops, this is exactly what will happen.

zen-dogAuthorUnsubmitted
Not Done

I was testing this case locally (difficult to create meaningful unit/integration tests for that) and decided to use BackoffSupervisor to prevent log spamming and avoid additional pressure on other parts of the system e.g. LaunchQueue or InstanceTracker. Thx for mentioning it in your previous comment.

52 import akka.pattern.{ Backoff, BackoffSupervisor }
53
54 def childSupervisor(props: Props, name: String): Props = {
55 BackoffSupervisor.props(
56 Backoff.onFailure(
57 childProps = props,
58 childName = name,
59 minBackoff = 5.seconds,
60 maxBackoff = 1.minute,
61 randomFactor = 0.2 // adds 20% "noise" to vary the intervals slightly
jeschkiesUnsubmitted
Not Done

Can this be configured through the application.conf?

62 ).withSupervisorStrategy(
63 OneForOneStrategy() {
64 case NonFatal(_) => SupervisorStrategy.Restart
jdefUnsubmitted
Done

maybe check for NonFatal instead of catching everything here?

65 case _ => SupervisorStrategy.Escalate
66 }
67 ))
68 }
69
4370 override def preStart(): Unit = {
4471 self ! NextStep
4572 }
4673
4774 override def postStop(): Unit = {
4875 deploymentManager ! DeploymentFinished(plan)
4976 }
5077
▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Line(s)
109136 }
110137 }
111138
112139 // scalastyle:on
113140
114141 def startRunnable(runnableSpec: RunSpec, scaleTo: Int, status: DeploymentStatus): Future[Unit] = {
115142 val promise = Promise[Unit]()
116143 instanceTracker.specInstances(runnableSpec.id).map { instances =>
117 context.actorOf(AppStartActor.props(deploymentManager, status, scheduler, launchQueue, instanceTracker,
118 eventBus, readinessCheckExecutor, runnableSpec, scaleTo, instances, promise))
144 context.actorOf(childSupervisor(AppStartActor.props(deploymentManager, status, scheduler, launchQueue, instanceTracker,
145 eventBus, readinessCheckExecutor, runnableSpec, scaleTo, instances, promise), s"AppStart-${plan.id}"))
119146 }
120147 promise.future
121148 }
122149
123150 @SuppressWarnings(Array("all")) /* async/await */
124151 def scaleRunnable(runnableSpec: RunSpec, scaleTo: Int,
125152 toKill: Option[Seq[Instance]],
126153 status: DeploymentStatus): Future[Done] = {
Show All 17 Lines
144171 }
145172 }
146173 await(killTasksIfNeeded)
147174
148175 def startTasksIfNeeded: Future[Done] = {
149176 tasksToStart.fold(Future.successful(Done)) { tasksToStart =>
150177 logger.debug(s"Start next $tasksToStart tasks")
151178 val promise = Promise[Unit]()
152 context.actorOf(TaskStartActor.props(deploymentManager, status, scheduler, launchQueue, instanceTracker, eventBus,
153 readinessCheckExecutor, runnableSpec, scaleTo, promise))
179 context.actorOf(childSupervisor(TaskStartActor.props(deploymentManager, status, scheduler, launchQueue, instanceTracker, eventBus,
180 readinessCheckExecutor, runnableSpec, scaleTo, promise), s"TaskStart-${plan.id}"))
154181 promise.future.map(_ => Done)
155182 }
156183 }
157184 await(startTasksIfNeeded)
158185 }
159186 }
160187
161188 @SuppressWarnings(Array("all")) /* async/await */
Show All 10 Lines
172199 Done
173200 }
174201
175202 def restartRunnable(run: RunSpec, status: DeploymentStatus): Future[Done] = {
176203 if (run.instances == 0) {
177204 Future.successful(Done)
178205 } else {
179206 val promise = Promise[Unit]()
180 context.actorOf(TaskReplaceActor.props(deploymentManager, status, killService,
181 launchQueue, instanceTracker, eventBus, readinessCheckExecutor, run, promise))
207 context.actorOf(childSupervisor(TaskReplaceActor.props(deploymentManager, status, killService,
208 launchQueue, instanceTracker, eventBus, readinessCheckExecutor, run, promise), s"TaskReplace-${plan.id}"))
182209 promise.future.map(_ => Done)
183210 }
184211 }
185212}
186213
187214object DeploymentActor {
188215 case object NextStep
189216 case object Finished
Show All 31 Lines
View Options

src/main/scala/mesosphere/marathon/core/deployment/impl/StartingBehavior.scala

11package mesosphere.marathon
22package core.deployment.impl
33
4import akka.actor.Actor
4import akka.Done
5import akka.pattern._
6import akka.actor.{ Actor, Status }
57import akka.event.EventStream
8import com.typesafe.scalalogging.StrictLogging
69import mesosphere.marathon.core.condition.Condition.Terminal
710import mesosphere.marathon.core.deployment.impl.StartingBehavior.Sync
811import mesosphere.marathon.core.event.{ InstanceChanged, InstanceHealthChanged }
912import mesosphere.marathon.core.instance.Instance
1013import mesosphere.marathon.core.launchqueue.LaunchQueue
1114import mesosphere.marathon.core.task.tracker.InstanceTracker
12import org.slf4j.LoggerFactory
1315
16import scala.async.Async.{ async, await }
17import scala.concurrent.Future
1418import scala.concurrent.duration._
1519
16trait StartingBehavior extends ReadinessBehavior { this: Actor =>
20trait StartingBehavior extends ReadinessBehavior with StrictLogging { this: Actor =>
1721 import context.dispatcher
1822
1923 def eventBus: EventStream
2024 def scaleTo: Int
21 def nrToStart: Int
25 def nrToStart: Future[Int]
2226 def launchQueue: LaunchQueue
2327 def scheduler: SchedulerActions
2428 def instanceTracker: InstanceTracker
2529
26 def initializeStart(): Unit
30 def initializeStart(): Future[Done]
2731
28 private[this] val log = LoggerFactory.getLogger(getClass)
29
32 @SuppressWarnings(Array("all")) // async/await
3033 final override def preStart(): Unit = {
3134 if (hasHealthChecks) eventBus.subscribe(self, classOf[InstanceHealthChanged])
3235 eventBus.subscribe(self, classOf[InstanceChanged])
3336
34 initializeStart()
37 async {
38 await(initializeStart())
3539 checkFinished()
3640 context.system.scheduler.scheduleOnce(1.seconds, self, Sync)
41 }.pipeTo(self)
3742 }
3843
3944 final override def receive: Receive = readinessBehavior orElse commonBehavior
4045
46 @SuppressWarnings(Array("all")) // async/await
4147 def commonBehavior: Receive = {
4248 case InstanceChanged(id, `version`, `pathId`, _: Terminal, _) =>
43 log.warn(s"New instance [$id] failed during app ${runSpec.id.toString} scaling, queueing another instance")
49 logger.warn(s"New instance [$id] failed during app ${runSpec.id.toString} scaling, queueing another instance")
4450 instanceTerminated(id)
45 launchQueue.add(runSpec)
51 launchQueue.addAsync(runSpec, 1).pipeTo(self)
4652
47 case Sync =>
48 val actualSize = launchQueue.get(runSpec.id)
49 .fold(instanceTracker.countLaunchedSpecInstancesSync(runSpec.id))(_.finalInstanceCount)
53 case Sync => async {
54 val launchedInstances = await(instanceTracker.countLaunchedSpecInstances(runSpec.id))
55 val actualSize = await(launchQueue.getAsync(runSpec.id)).fold(launchedInstances)(_.finalInstanceCount)
5056 val instancesToStartNow = Math.max(scaleTo - actualSize, 0)
51 log.debug(s"Sync start instancesToStartNow=$instancesToStartNow appId=${runSpec.id}")
57 logger.debug(s"Sync start instancesToStartNow=$instancesToStartNow appId=${runSpec.id}")
5258 if (instancesToStartNow > 0) {
53 log.info(s"Reconciling app ${runSpec.id} scaling: queuing $instancesToStartNow new instances")
54 launchQueue.add(runSpec, instancesToStartNow)
59 logger.info(s"Reconciling app ${runSpec.id} scaling: queuing $instancesToStartNow new instances")
60 await(launchQueue.addAsync(runSpec, instancesToStartNow))
5561 }
5662 context.system.scheduler.scheduleOnce(5.seconds, self, Sync)
63 }.pipeTo(self)
64
65 case Status.Failure(e) =>
66 // This is the result of failed initializeStart(...) call. Log the message and
67 // restart this actor. Next reincarnation should try to start from the beginning.
68 logger.warn(s"Failure in the ${getClass.getSimpleName} deployment actor: ", e)
jdefUnsubmitted
Done

do we have a policy about the log level that should be used when logging an exception that's rethrown, unchanged? for example, in this case I'd imagine we'll see multiple copies of the same stack trace -- maybe not very useful

jeschkiesUnsubmitted
Not Done

That is a good point. I used warn but I don't think we have a convention.

zen-dogAuthorUnsubmitted
Not Done

Sounds reasonable.

69 throw e
70
71 case Done => // This is the result of successful initializeStart(...) call. Nothing to do here
5772 }
5873
5974 override def instanceConditionChanged(instanceId: Instance.Id): Unit = {
60 log.debug(s"New instance $instanceId changed during app ${runSpec.id} scaling, " +
61 s"${readyInstances.size} ready ${healthyInstances.size} healthy need $nrToStart")
75 logger.debug(s"New instance $instanceId changed during app ${runSpec.id} scaling, " +
76 s"${readyInstances.size} ready ${healthyInstances.size} healthy need ${nrToStart.value}")
6277 checkFinished()
6378 }
6479
6580 def checkFinished(): Unit = {
66 if (targetCountReached(nrToStart)) success()
81 nrToStart.foreach{ n =>
82 if (targetCountReached(n)) success()
83 }
6784 }
6885
6986 def success(): Unit
7087}
7188
7289object StartingBehavior {
7390 case object Sync
7491}
7592
View Options

src/main/scala/mesosphere/marathon/core/deployment/impl/TaskReplaceActor.scala

11package mesosphere.marathon
22package core.deployment.impl
33
4import akka.Done
45import akka.actor._
56import akka.event.EventStream
7import akka.pattern._
68import com.typesafe.scalalogging.StrictLogging
9import mesosphere.marathon.core.async.ExecutionContexts.global
710import mesosphere.marathon.core.event._
811import mesosphere.marathon.core.instance.Instance
912import mesosphere.marathon.core.instance.Instance.Id
1013import mesosphere.marathon.core.launchqueue.LaunchQueue
1114import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1215import mesosphere.marathon.core.task.termination.InstanceChangedPredicates.considerTerminal
1316import mesosphere.marathon.core.task.termination.{ KillReason, KillService }
1417import mesosphere.marathon.core.task.tracker.InstanceTracker
1518import mesosphere.marathon.state.RunSpec
1619
1720import scala.collection.{ SortedSet, mutable }
18import scala.concurrent.Promise
21import scala.concurrent.{ Future, Promise }
1922
2023class TaskReplaceActor(
2124 val deploymentManager: ActorRef,
2225 val status: DeploymentStatus,
2326 val killService: KillService,
2427 val launchQueue: LaunchQueue,
2528 val instanceTracker: InstanceTracker,
2629 val eventBus: EventStream,
▲ Show 20 Lines • Show All 68 Lines • ▼ Show 20 Line(s)
9598 logger.error(s"New instance $id failed on agent ${instance.agentInfo.agentId} during app $pathId restart")
9699 instanceTerminated(id)
97100 instancesStarted -= 1
98101 launchInstances()
99102
100103 // Old instance successfully killed
101104 case InstanceChanged(id, _, `pathId`, condition, _) if oldInstanceIds(id) && considerTerminal(condition) =>
102105 oldInstanceIds -= id
103 launchInstances()
106 launchInstances().foreach(_ => checkFinished())
jeschkiesUnsubmitted
Done

Hm, I think this should be a onComplete or foreach since checkFinished() has side effects. Also, the new future is not used.

104 checkFinished()
105107
106108 // Ignore change events, that are not handled in parent receives
107109 case _: InstanceChanged =>
108110
111 case Status.Failure(e) =>
112 // This is the result of failed launchQueue.addAsync(...) call. Log the message and
113 // restart this actor. Next reincarnation should try to start from the beginning.
114 logger.warn("Failed to launch instances: ", e)
115 throw e
116
117 case Done => // This is the result of successful launchQueue.addAsync(...) call. Nothing to do here
118
109119 }
110120
111121 override def instanceConditionChanged(instanceId: Instance.Id): Unit = {
112122 if (healthyInstances(instanceId) && readyInstances(instanceId)) killNextOldInstance(Some(instanceId))
113123 checkFinished()
114124 }
115125
116126 def reconcileAlreadyStartedInstances(): Unit = {
117127 logger.info(s"reconcile: found ${instancesAlreadyStarted.size} already started instances " +
118128 s"and ${oldInstanceIds.size} old instances")
119129 instancesAlreadyStarted.foreach(reconcileHealthAndReadinessCheck)
120130 }
121131
122 def launchInstances(): Unit = {
132 // Careful not to make this method completely asynchronous - it changes local actor's state `instancesStarted`.
133 // Only launching new instances needs to be asynchronous.
jdefUnsubmitted
Not Done

thanks for this

134 def launchInstances(): Future[Done] = {
123135 val leftCapacity = math.max(0, ignitionStrategy.maxCapacity - oldInstanceIds.size - instancesStarted)
124136 val instancesNotStartedYet = math.max(0, runSpec.instances - instancesStarted)
125137 val instancesToStartNow = math.min(instancesNotStartedYet, leftCapacity)
126138 if (instancesToStartNow > 0) {
127139 logger.info(s"Reconciling instances during app $pathId restart: queuing $instancesToStartNow new instances")
128 launchQueue.add(runSpec, instancesToStartNow)
129140 instancesStarted += instancesToStartNow
141 launchQueue.addAsync(runSpec, instancesToStartNow).pipeTo(self)
142 } else {
143 Future.successful(Done)
130144 }
131145 }
zen-dogAuthorUnsubmitted
Not Done

On a second thought - I realized yesterday why I didn't make the method completely async in the first place: method changes instancesStarted (actor's state) and doing it from an async block will create a race condition. I could make it Atomic of course but prefer not to. Reverted to first implementation.

jeschkiesUnsubmitted
Not Done

Hm, Why is it return a Future[Done]? It will alway finished immediately without waiting for the addAsync. Is this what we want?

zen-dogAuthorUnsubmitted
Not Done

You're right. Should be if else. Done

132146
133147 def killNextOldInstance(maybeNewInstanceId: Option[Instance.Id] = None): Unit = {
134148 if (toKill.nonEmpty) {
135149 val nextOldInstance = toKill.dequeue()
136150
137151 maybeNewInstanceId match {
138152 case Some(newInstanceId: Instance.Id) =>
139153 logger.info(s"Killing old ${nextOldInstance.instanceId} because $newInstanceId became reachable")
▲ Show 20 Lines • Show All 78 Lines • Show Last 20 Lines
View Options

src/main/scala/mesosphere/marathon/core/deployment/impl/TaskStartActor.scala

11package mesosphere.marathon
22package core.deployment.impl
33
4import akka.Done
5import akka.pattern._
46import akka.actor.{ Actor, ActorRef, Props }
57import akka.event.EventStream
68import com.typesafe.scalalogging.StrictLogging
79import mesosphere.marathon.core.event.DeploymentStatus
810import mesosphere.marathon.core.launchqueue.LaunchQueue
911import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1012import mesosphere.marathon.core.task.tracker.InstanceTracker
1113import mesosphere.marathon.state.RunSpec
1214
13import scala.concurrent.Promise
15import scala.async.Async.{ async, await }
16import scala.concurrent.{ Future, Promise }
17import mesosphere.marathon.core.async.ExecutionContexts.global
1418
19@SuppressWarnings(Array("all")) // async/await
1520class TaskStartActor(
1621 val deploymentManager: ActorRef,
1722 val status: DeploymentStatus,
1823 val scheduler: SchedulerActions,
1924 val launchQueue: LaunchQueue,
2025 val instanceTracker: InstanceTracker,
2126 val eventBus: EventStream,
2227 val readinessCheckExecutor: ReadinessCheckExecutor,
2328 val runSpec: RunSpec,
2429 val scaleTo: Int,
2530 promise: Promise[Unit]) extends Actor with StrictLogging with StartingBehavior {
2631
27 val nrToStart: Int = Math.max(
28 0,
29 scaleTo - launchQueue.get(runSpec.id).map(_.finalInstanceCount)
30 .getOrElse(instanceTracker.countLaunchedSpecInstancesSync(runSpec.id)))
32 override val nrToStart: Future[Int] = async {
33 val alreadyLaunched = await(launchQueue.getAsync(runSpec.id)) match {
34 case Some(info) => info.finalInstanceCount
35 case None => await(instanceTracker.countLaunchedSpecInstances(runSpec.id))
36 }
37 Math.max(0, scaleTo - alreadyLaunched)
38 }.pipeTo(self)
3139
32 def initializeStart(): Unit = if (nrToStart > 0) launchQueue.add(runSpec, nrToStart)
40 @SuppressWarnings(Array("all")) // async/await
41 override def initializeStart(): Future[Done] = async {
42 val toStart = await(nrToStart)
43 if (toStart > 0) await(launchQueue.addAsync(runSpec, toStart))
44 else Done
45 }.pipeTo(self)
3346
3447 override def postStop(): Unit = {
3548 eventBus.unsubscribe(self)
3649 super.postStop()
3750 }
3851
3952 override def success(): Unit = {
4053 logger.info(s"Successfully started $nrToStart instances of ${runSpec.id}")
41 promise.success(())
54 // Since a lot of StartingBehavior and this actor's code happens asynchronously now
55 // it can happen that this promise might succeed twice.
56 promise.trySuccess(())
4257 context.stop(self)
4358 }
4459}
4560
4661object TaskStartActor {
4762 @SuppressWarnings(Array("MaxParameters"))
4863 def props(
4964 deploymentManager: ActorRef,
Show All 14 Lines
View Options

src/main/scala/mesosphere/marathon/core/launchqueue/LaunchQueue.scala

Show All 40 Lines
4141 unusedOffersCount: Int,
4242 lastMatch: Option[OfferMatchResult.Match],
4343 lastNoMatch: Option[OfferMatchResult.NoMatch],
4444 lastNoMatches: Seq[OfferMatchResult.NoMatch]
4545 )
4646}
4747
4848/**
49 * The LaunchQueue contains requests to launch new instances for a run spec.
49 * The LaunchQueue contains requests to launch new instances for a run spec. For every method returning T
50 * there is a corresponding async method which returns a Future[T]. Async methods should be preferred
51 * where synchronous methods will be deprecated and gradually removed.
jdefUnsubmitted
Not Done

are there any that we can deprecate now? if so, let's do it

zen-dogAuthorUnsubmitted
Not Done

They are used all over the place so it'll generate lot of warnings. I'd rather not do it now.

jdefUnsubmitted
Not Done

OK, can we file a JIRA to track this tech debt (and link to it here)?

zen-dogAuthorUnsubmitted
Not Done

Not that I know of. I could create one, but since this part (along with deployments) is probably part of Action/v3/what ever, this ticket would never be touched.

5052 */
5153trait LaunchQueue {
5254
5355 /** Returns all entries of the queue. */
5456 def list: Seq[QueuedInstanceInfo]
5557
58 def listAsync: Future[Seq[QueuedInstanceInfo]]
59
5660 /** Returns all entries of the queue with embedded statistics */
5761 def listWithStatistics: Seq[QueuedInstanceInfoWithStatistics]
5862
63 def listWithStatisticsAsync: Future[Seq[QueuedInstanceInfoWithStatistics]]
64
5965 /** Returns all runnable specs for which queue entries exist. */
6066 def listRunSpecs: Seq[RunSpec]
6167
68 def listRunSpecsAsync: Future[Seq[RunSpec]]
69
6270 /** Request to launch `count` additional instances conforming to the given run spec. */
63 def add(spec: RunSpec, count: Int = 1): Unit
71 def add(spec: RunSpec, count: Int = 1): Done
6472
73 def addAsync(spec: RunSpec, count: Int = 1): Future[Done]
74
6575 /** Get information for the given run spec id. */
6676 def get(specId: PathId): Option[QueuedInstanceInfo]
6777
78 def getAsync(specId: PathId): Future[Option[QueuedInstanceInfo]]
79
6880 /** Return how many instances are still to be launched for this PathId. */
6981 def count(specId: PathId): Int
82
83 def countAsync(specId: PathId): Future[Int]
7084
7185 /** Remove all instance launch requests for the given PathId from this queue. */
7286 def asyncPurge(specId: PathId): Future[Done]
7387
7488 /** Add delay to the given RunnableSpec because of a failed instance */
7589 def addDelay(spec: RunSpec): Unit
7690
7791 /** Reset the backoff delay for the given RunnableSpec. */
7892 def resetDelay(spec: RunSpec): Unit
7993
8094 /** Notify queue about InstanceUpdate */
8195 def notifyOfInstanceUpdate(update: InstanceChange): Future[Done]
8296}
View Options

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

Show First 20 Lines • Show All 182 Lines • ▼ Show 20 Line(s)
183183
184184 case Add(app, count) =>
185185 launchers.get(app.id) match {
186186 case None =>
187187 import context.dispatcher
188188 val actorRef = createAppTaskLauncher(app, count)
189189 val eventualCount: Future[QueuedInstanceInfo] =
190190 (actorRef ? TaskLauncherActor.GetCount).mapTo[QueuedInstanceInfo]
191 eventualCount.map(_ => ()).pipeTo(sender())
191 eventualCount.map(_ => Done).pipeTo(sender())
192192
193193 case Some(actorRef) =>
194194 import context.dispatcher
195195 val eventualCount: Future[QueuedInstanceInfo] =
196196 (actorRef ? TaskLauncherActor.AddInstances(app, count)).mapTo[QueuedInstanceInfo]
197 eventualCount.map(_ => ()).pipeTo(sender())
197 eventualCount.map(_ => Done).pipeTo(sender())
198198 }
199199
200200 case msg @ RateLimiterActor.DelayUpdate(app, _) =>
201201 launchers.get(app.id).foreach(_.forward(msg))
202202 }
203203
204204 private[this] def createAppTaskLauncher(app: RunSpec, initialCount: Int): ActorRef = {
205205 val actorRef = context.actorOf(runSpecActorProps(app, initialCount), s"$childSerial-${app.id.safePath}")
Show All 16 Lines
View Options

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

11package mesosphere.marathon
22package core.launchqueue.impl
33
44import akka.Done
55import akka.actor.ActorRef
66import akka.pattern.ask
77import akka.util.Timeout
88import com.typesafe.scalalogging.StrictLogging
9import mesosphere.marathon.core.async.ExecutionContexts
910import mesosphere.marathon.core.instance.update.InstanceChange
1011import mesosphere.marathon.core.launchqueue.LaunchQueue.{ QueuedInstanceInfo, QueuedInstanceInfoWithStatistics }
1112import mesosphere.marathon.core.launchqueue.{ LaunchQueue, LaunchQueueConfig }
1213import mesosphere.marathon.state.{ PathId, RunSpec }
1314
1415import scala.collection.immutable.Seq
1516import scala.concurrent.duration._
1617import scala.concurrent.{ Await, Future }
Show All 10 Lines
2728 val purgeTimeout: Timeout = config.launchQueueRequestTimeout().milliseconds + config.taskOpNotificationTimeout().millisecond
2829
2930 val launchQueueRequestTimeout: Timeout = config.launchQueueRequestTimeout().milliseconds
3031
3132 override def list: Seq[QueuedInstanceInfo] = {
3233 askQueueActor[LaunchQueueDelegate.Request, Seq[QueuedInstanceInfo]]("list")(LaunchQueueDelegate.List)
3334 }
3435
36 override def listAsync: Future[Seq[QueuedInstanceInfo]] =
37 askQueueActorFuture[LaunchQueueDelegate.Request, Seq[QueuedInstanceInfo]]("list")(LaunchQueueDelegate.List)
38
3539 override def listWithStatistics: Seq[QueuedInstanceInfoWithStatistics] = {
3640 askQueueActor[LaunchQueueDelegate.Request, Seq[QueuedInstanceInfoWithStatistics]]("listWithStatistics")(LaunchQueueDelegate.ListWithStatistics)
3741 }
3842
43 override def listWithStatisticsAsync: Future[Seq[QueuedInstanceInfoWithStatistics]] =
44 askQueueActorFuture[LaunchQueueDelegate.Request, Seq[QueuedInstanceInfoWithStatistics]]("listWithStatistics")(LaunchQueueDelegate.ListWithStatistics)
45
3946 override def get(runSpecId: PathId): Option[QueuedInstanceInfo] =
4047 askQueueActor[LaunchQueueDelegate.Request, Option[QueuedInstanceInfo]]("get")(LaunchQueueDelegate.Count(runSpecId))
4148
49 override def getAsync(runSpecId: PathId): Future[Option[QueuedInstanceInfo]] =
50 askQueueActorFuture[LaunchQueueDelegate.Request, Option[QueuedInstanceInfo]]("get")(LaunchQueueDelegate.Count(runSpecId))
51
4252 override def notifyOfInstanceUpdate(update: InstanceChange): Future[Done] =
4353 askQueueActorFuture[InstanceChange, Done]("notifyOfInstanceUpdate")(update)
4454
4555 override def count(runSpecId: PathId): Int = get(runSpecId).map(_.instancesLeftToLaunch).getOrElse(0)
4656
57 override def countAsync(runSpecId: PathId): Future[Int] =
58 getAsync(runSpecId).map {
59 case Some(i) => i.instancesLeftToLaunch
60 case None => 0
61 }(ExecutionContexts.global)
62
4763 override def listRunSpecs: Seq[RunSpec] = list.map(_.runSpec)
4864
49 override def asyncPurge(runSpecId: PathId): Future[Done] = {
65 override def listRunSpecsAsync: Future[Seq[RunSpec]] =
66 listAsync.map(_.map(_.runSpec))(ExecutionContexts.global)
67
68 override def asyncPurge(runSpecId: PathId): Future[Done] =
5069 askQueueActorFuture[LaunchQueueDelegate.Request, Done]("asyncPurge", timeout = purgeTimeout)(LaunchQueueDelegate.Purge(runSpecId))
51 }
5270
53 override def add(runSpec: RunSpec, count: Int): Unit = askQueueActor[LaunchQueueDelegate.Request, Unit]("add")(LaunchQueueDelegate.Add(runSpec, count))
71 override def add(runSpec: RunSpec, count: Int): Done =
72 askQueueActor[LaunchQueueDelegate.Request, Done]("add")(LaunchQueueDelegate.Add(runSpec, count))
73
74 override def addAsync(runSpec: RunSpec, count: Int): Future[Done] =
75 askQueueActorFuture[LaunchQueueDelegate.Request, Done]("add")(LaunchQueueDelegate.Add(runSpec, count))
5476
5577 private[this] def askQueueActor[T, R: ClassTag](
5678 method: String,
5779 timeout: Timeout = launchQueueRequestTimeout)(message: T): R = {
5880
5981 val answerFuture = askQueueActorFuture[T, R](method, timeout)(message)
6082 Await.result(answerFuture, timeout.duration)
6183 }
Show All 28 Lines
View Options

src/main/scala/mesosphere/marathon/core/task/tracker/InstanceTracker.scala

Show All 23 Lines
2424 def specInstances(pathId: PathId)(implicit ec: ExecutionContext): Future[Seq[Instance]]
2525
2626 def instance(instanceId: Instance.Id): Future[Option[Instance]]
2727
2828 def instancesBySpecSync: InstanceTracker.InstancesBySpec
2929 def instancesBySpec()(implicit ec: ExecutionContext): Future[InstanceTracker.InstancesBySpec]
3030
3131 def countLaunchedSpecInstancesSync(appId: PathId): Int
32 def countLaunchedSpecInstances(appId: PathId): Future[Int]
3233
3334 def hasSpecInstancesSync(appId: PathId): Boolean
3435 def hasSpecInstances(appId: PathId)(implicit ec: ExecutionContext): Future[Boolean]
3536}
3637
3738object InstanceTracker {
3839 /**
3940 * Contains all tasks grouped by app ID.
▲ Show 20 Lines • Show All 77 Lines • Show Last 20 Lines
View Options

src/main/scala/mesosphere/marathon/core/task/tracker/impl/InstanceTrackerDelegate.scala

Show All 20 Lines
2121 * is used internally in this package to communicate with the TaskTracker.
2222 */
2323private[tracker] class InstanceTrackerDelegate(
2424 config: InstanceTrackerConfig,
2525 taskTrackerRef: ActorRef) extends InstanceTracker {
2626
2727 override def instancesBySpecSync: InstanceTracker.InstancesBySpec = {
2828 import mesosphere.marathon.core.async.ExecutionContexts.global
2929 Await.result(instancesBySpec(), taskTrackerQueryTimeout.duration)
jdefUnsubmitted
Done

why this change? it's not consistent...

3030 }
3131
3232 override def instancesBySpec()(implicit ec: ExecutionContext): Future[InstanceTracker.InstancesBySpec] = tasksByAppTimer {
3333 (taskTrackerRef ? InstanceTrackerActor.List).mapTo[InstanceTracker.InstancesBySpec].recover {
3434 case e: AskTimeoutException =>
3535 throw new TimeoutException(
3636 "timeout while calling list. If you know what you are doing, you can adjust the timeout " +
3737 s"with --${config.internalTaskTrackerRequestTimeout.name}."
3838 )
3939 }
4040 }
4141
4242 // TODO(jdef) support pods when counting launched instances
4343 override def countLaunchedSpecInstancesSync(appId: PathId): Int =
4444 instancesBySpecSync.specInstances(appId).count(_.isLaunched)
45 override def countLaunchedSpecInstances(appId: PathId): Future[Int] = {
46 import mesosphere.marathon.core.async.ExecutionContexts.global
jeschkiesUnsubmitted
Done

Should this be imported at the top? It's used at several places.

zen-dogAuthorUnsubmitted
Not Done

No, it conflicts with implicit ec: ExecutionContext used below.

jdefUnsubmitted
Not Done

I thought we were supposed to standardize on the marathon core async global execution context in order for our MDC/actions context propagation to work properly. this looks like a red flag. what am I missing?

zen-dogAuthorUnsubmitted
Not Done

Everbody is using our custom global execution context (linter rejects the scala's one) but if you import it here at the top of the file you'll get:

[error] /Users/adukhovniy/mesosphere/marathon/src/main/scala/mesosphere/marathon/core/task/tracker/impl/InstanceTrackerDelegate.scala:33: ambiguous implicit values:
[error]  both value ec of type scala.concurrent.ExecutionContext
[error]  and value global in object ExecutionContexts of type => scala.concurrent.ExecutionContext
...
47 instancesBySpec().map(_.specInstances(appId).count(_.isLaunched))
48 }
4549
4650 override def hasSpecInstancesSync(appId: PathId): Boolean = instancesBySpecSync.hasSpecInstances(appId)
4751 override def hasSpecInstances(appId: PathId)(implicit ec: ExecutionContext): Future[Boolean] =
4852 instancesBySpec().map(_.hasSpecInstances(appId))
4953
5054 override def specInstancesSync(appId: PathId): Seq[Instance] =
5155 instancesBySpecSync.specInstances(appId)
5256 override def specInstances(appId: PathId)(implicit ec: ExecutionContext): Future[Seq[Instance]] =
Show All 9 Lines
View Options

src/test/scala/mesosphere/marathon/core/deployment/impl/AppStartActorTest.scala

11package mesosphere.marathon
22package core.deployment.impl
33
4import akka.Done
45import akka.testkit.{ TestActorRef, TestProbe }
56import mesosphere.AkkaUnitTest
67import mesosphere.marathon.core.condition.Condition
78import mesosphere.marathon.core.event.{ DeploymentStatus, InstanceChanged, InstanceHealthChanged }
89import mesosphere.marathon.core.health.{ MarathonHttpHealthCheck, PortReference }
910import mesosphere.marathon.core.instance.Instance
1011import mesosphere.marathon.core.launchqueue.LaunchQueue
1112import mesosphere.marathon.core.leadership.AlwaysElectedLeadershipModule
1213import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1314import mesosphere.marathon.core.task.tracker.InstanceTracker
1415import mesosphere.marathon.state.{ AppDefinition, PathId }
1516import mesosphere.marathon.test.MarathonTestHelper
1617import org.scalatest.concurrent.PatienceConfiguration.Timeout
1718
1819import scala.concurrent.duration._
19import scala.concurrent.{ Promise }
20import scala.concurrent.{ Future, Promise }
2021
2122class AppStartActorTest extends AkkaUnitTest {
2223 "AppStartActor" should {
2324 "Without Health Checks" in {
2425 val f = new Fixture
2526 val app = AppDefinition(id = f.appId, instances = 10)
2627 val promise = Promise[Unit]()
2728 val ref = f.startActor(app, scaleTo = 2, promise)
▲ Show 20 Lines • Show All 63 Lines • ▼ Show 20 Line(s)
9192 val instanceTracker: InstanceTracker = MarathonTestHelper.createTaskTracker(
9293 AlwaysElectedLeadershipModule.forRefFactory(system))
9394 val deploymentManager: TestProbe = TestProbe()
9495 val deploymentStatus: DeploymentStatus = mock[DeploymentStatus]
9596 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
9697 val appId = PathId("/app")
9798
9899 launchQueue.get(appId) returns None
100 scheduler.startRunSpec(any) returns Future.successful(Done)
99101
100102 def instanceChanged(app: AppDefinition, condition: Condition): InstanceChanged = {
101103 val instanceId = Instance.Id.forRunSpec(app.id)
102104 val instance: Instance = mock[Instance]
103105 instance.instanceId returns instanceId
104106 InstanceChanged(instanceId, app.version, app.id, condition, instance)
105107 }
106108
Show All 11 Lines
View Options

src/test/scala/mesosphere/marathon/core/deployment/impl/DeploymentActorTest.scala

Show First 20 Lines • Show All 113 Lines • ▼ Show 20 Line(s)
114114 val instance4_1 = {
115115 val instance = TestInstanceBuilder.newBuilder(app4.id, version = app4.version).addTaskRunning().getInstance()
116116 val state = instance.state.copy(condition = Condition.Running)
117117 instance.copy(state = state)
118118 }
119119
120120 val plan = DeploymentPlan(origGroup, targetGroup)
121121
122 scheduler.startRunSpec(any) returns Future.successful(Done)
122123 tracker.specInstances(Matchers.eq(app1.id))(any[ExecutionContext]) returns Future.successful(Seq(instance1_1, instance1_2))
123 when(tracker.specInstancesSync(app2.id)).thenReturn(Seq(instance2_1))
124 tracker.specInstancesSync(app2.id) returns Seq(instance2_1)
124125 tracker.specInstances(Matchers.eq(app2.id))(any[ExecutionContext]) returns Future.successful(Seq(instance2_1))
125126 tracker.specInstances(Matchers.eq(app3.id))(any[ExecutionContext]) returns Future.successful(Seq(instance3_1))
126127 tracker.specInstances(Matchers.eq(app4.id))(any[ExecutionContext]) returns Future.successful(Seq(instance4_1))
127128
128 when(queue.add(same(app2New), any[Int])).thenAnswer(new Answer[Boolean] {
129 def answer(invocation: InvocationOnMock): Boolean = {
129 when(queue.addAsync(same(app2New), any[Int])).thenAnswer(new Answer[Future[Done]] {
130 def answer(invocation: InvocationOnMock): Future[Done] = {
130131 for (i <- 0 until invocation.getArguments()(1).asInstanceOf[Int])
131132 system.eventStream.publish(instanceChanged(app2New, Condition.Running))
132 true
133 Future.successful(Done)
133134 }
134135 })
135136
136137 deploymentActor(managerProbe.ref, Promise[Done](), plan)
137138 plan.steps.zipWithIndex.foreach {
138139 case (step, num) => managerProbe.expectMsg(7.seconds, DeploymentStepInfo(plan, step, num + 1))
139140 }
140141
Show All 17 Lines
158159 val version2 = VersionInfo.forNewConfig(Timestamp(1000))
159160 val appNew = app.copy(cmd = Some("cmd new"), versionInfo = version2)
160161
161162 val targetGroup = createRootGroup(groups = Set(createGroup(PathId("/foo/bar"), Map(appNew.id -> appNew))))
162163
163164 val instance1_1 = TestInstanceBuilder.newBuilder(app.id, version = app.version).addTaskRunning(startedAt = Timestamp.zero).getInstance()
164165 val instance1_2 = TestInstanceBuilder.newBuilder(app.id, version = app.version).addTaskRunning(startedAt = Timestamp(1000)).getInstance()
165166
166 when(tracker.specInstancesSync(app.id)).thenReturn(Seq(instance1_1, instance1_2))
167 tracker.specInstancesSync(app.id) returns Seq(instance1_1, instance1_2)
168 tracker.specInstances(app.id) returns Future.successful(Seq(instance1_1, instance1_2))
167169
168170 val plan = DeploymentPlan("foo", origGroup, targetGroup, List(DeploymentStep(List(RestartApplication(appNew)))), Timestamp.now())
169171
170 when(queue.count(appNew.id)).thenAnswer(new Answer[Int] {
171 override def answer(p1: InvocationOnMock): Int = appNew.instances
172 })
172 queue.countAsync(appNew.id) returns Future.successful(appNew.instances)
173173
174 when(queue.add(same(appNew), any[Int])).thenAnswer(new Answer[Boolean] {
175 def answer(invocation: InvocationOnMock): Boolean = {
174 when(queue.addAsync(same(appNew), any[Int])).thenAnswer(new Answer[Future[Done]] {
175 def answer(invocation: InvocationOnMock): Future[Done] = {
176176 for (i <- 0 until invocation.getArguments()(1).asInstanceOf[Int])
177177 system.eventStream.publish(instanceChanged(appNew, Condition.Running))
178 true
178 Future.successful(Done)
179179 }
180180 })
181181
182182 deploymentActor(managerProbe.ref, promise, plan)
183183 promise.future.futureValue should be (Done)
184184
185185 killService.killed should contain(instance1_1.instanceId)
186186 killService.killed should contain(instance1_2.instanceId)
187 verify(queue).add(appNew, 2)
187 verify(queue).addAsync(appNew, 2)
188188 }
189189
190190 "Restart suspended app" in new Fixture {
191191 val managerProbe = TestProbe()
192192 val promise = Promise[Done]()
193193
194194 val app = AppDefinition(id = PathId("/app1"), cmd = Some("cmd"), instances = 0)
195195 val origGroup = createRootGroup(groups = Set(createGroup(PathId("/foo/bar"), Map(app.id -> app))))
196196
197197 val version2 = VersionInfo.forNewConfig(Timestamp(1000))
198198 val appNew = app.copy(cmd = Some("cmd new"), versionInfo = version2)
199199 val targetGroup = createRootGroup(groups = Set(createGroup(PathId("/foo/bar"), Map(appNew.id -> appNew))))
200200
201201 val plan = DeploymentPlan("foo", origGroup, targetGroup, List(DeploymentStep(List(RestartApplication(appNew)))), Timestamp.now())
202202
203 when(tracker.specInstancesSync(app.id)).thenReturn(Seq.empty[Instance])
203 tracker.specInstancesSync(app.id) returns Seq.empty[Instance]
204 queue.addAsync(app, 2) returns Future.successful(Done)
204205
205206 deploymentActor(managerProbe.ref, promise, plan)
206207 promise.future.futureValue should be (Done)
207208 }
208209
209210 "Scale with tasksToKill" in new Fixture {
210211 val managerProbe = TestProbe()
211212 val app1 = AppDefinition(id = PathId("/app1"), cmd = Some("cmd"), instances = 3)
Show All 28 Lines
View Options

src/test/scala/mesosphere/marathon/core/deployment/impl/DeploymentManagerActorTest.scala

Show First 20 Lines • Show All 217 Lines • ▼ Show 20 Line(s)
218218 eventBus,
219219 readinessCheckExecutor,
220220 deploymentRepo,
221221 deploymentActorProps)
222222 )
223223 deploymentRepo.store(any[DeploymentPlan]) returns Future.successful(Done)
224224 deploymentRepo.delete(any[String]) returns Future.successful(Done)
225225 deploymentRepo.all() returns Source.empty
226 launchQueue.getAsync(any) returns Future.successful(None)
227 launchQueue.addAsync(any, any) returns Future.successful(Done)
226228 }
227229}
View Options

src/test/scala/mesosphere/marathon/core/deployment/impl/TaskReplaceActorTest.scala

11package mesosphere.marathon
22package core.deployment.impl
33
4import akka.Done
45import akka.actor.{ Actor, Props }
56import akka.testkit.TestActorRef
67import mesosphere.AkkaUnitTest
78import mesosphere.marathon.core.condition.Condition
89import mesosphere.marathon.core.condition.Condition.Running
910import mesosphere.marathon.core.deployment.{ DeploymentPlan, DeploymentStep }
1011import mesosphere.marathon.core.event._
1112import mesosphere.marathon.core.health.{ MarathonHttpHealthCheck, PortReference }
1213import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
1314import mesosphere.marathon.core.launchqueue.LaunchQueue
1415import mesosphere.marathon.core.readiness.{ ReadinessCheck, ReadinessCheckExecutor, ReadinessCheckResult }
1516import mesosphere.marathon.core.task.tracker.InstanceTracker
1617import mesosphere.marathon.core.task.{ KillServiceMock, Task }
1718import mesosphere.marathon.state.PathId._
1819import mesosphere.marathon.state._
19import org.mockito.Mockito
20import org.mockito.{ Mockito }
2021import org.mockito.Mockito._
2122import org.scalatest.concurrent.Eventually
2223import org.scalatest.concurrent.PatienceConfiguration.Timeout
2324import rx.lang.scala.Observable
2425
25import scala.concurrent.Promise
26import scala.concurrent.{ Future, Promise }
2627import scala.concurrent.duration._
2728
2829class TaskReplaceActorTest extends AkkaUnitTest with Eventually {
2930 "TaskReplaceActor" should {
3031 "Replace without health checks" in {
3132 val f = new Fixture
3233 val app = AppDefinition(
3334 id = "/myApp".toPath,
3435 instances = 5,
3536 versionInfo = VersionInfo.forNewConfig(Timestamp(0)),
3637 upgradeStrategy = UpgradeStrategy(0.0))
3738 val instanceA = f.runningInstance(app)
3839 val instanceB = f.runningInstance(app)
3940
40 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceB))
41 f.tracker.specInstancesSync(app.id) returns Seq(instanceA, instanceB)
4142
4243 val promise = Promise[Unit]()
4344 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
45 f.queue.addAsync(newApp, 5) returns Future.successful(Done)
46
4447 val ref = f.replaceActor(newApp, promise)
4548 watch(ref)
4649
4750 for (_ <- 0 until newApp.instances)
4851 ref ! f.instanceChanged(newApp, Running)
4952
5053 promise.future.futureValue
5154 verify(f.queue).resetDelay(newApp)
Show All 10 Lines
6265 instances = 5,
6366 versionInfo = VersionInfo.forNewConfig(Timestamp(0)),
6467 upgradeStrategy = UpgradeStrategy(0.0))
6568
6669 val instanceA = f.runningInstance(app)
6770
6871 val promise = Promise[Unit]()
6972 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
73 f.queue.addAsync(eq(newApp), any) returns Future.successful(Done)
7074
7175 val instanceC = f.runningInstance(newApp)
7276
73 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceC))
77 f.tracker.specInstancesSync(app.id) returns Seq(instanceA, instanceC)
7478
7579 val ref = f.replaceActor(newApp, promise)
7680 watch(ref)
7781
7882 for (_ <- 0 until newApp.instances)
7983 ref ! f.instanceChanged(newApp, Running)
8084
8185 promise.future.futureValue
Show All 15 Lines
97101
98102 val instanceA = f.runningInstance(app)
99103 val instanceB = f.runningInstance(app)
100104
101105 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceB))
102106
103107 val promise = Promise[Unit]()
104108 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
109 f.queue.addAsync(newApp, 5) returns Future.successful(Done)
110
105111 val ref = f.replaceActor(newApp, promise)
106112 watch(ref)
107113
108114 for (_ <- 0 until newApp.instances)
109115 ref ! f.healthChanged(newApp, healthy = true)
110116
111117 promise.future.futureValue
112118 verify(f.queue).resetDelay(newApp)
Show All 13 Lines
126132 val instanceA = f.runningInstance(app)
127133 val instanceB = f.runningInstance(app)
128134 val instanceC = f.runningInstance(app)
129135
130136 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceB, instanceC))
131137
132138 val promise = Promise[Unit]()
133139 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
140 f.queue.addAsync(newApp, 1) returns Future.successful(Done)
141
134142 val ref = f.replaceActor(newApp, promise)
135143 watch(ref)
136144
137145 eventually {
138146 f.killService.numKilled should be(1)
139147 }
140148
141149 ref ! f.instanceChanged(newApp, Running)
142150 eventually {
143151 f.killService.numKilled should be(2)
144152 }
145153
146154 ref ! f.instanceChanged(newApp, Running)
147 eventually { app: AppDefinition => verify(f.queue, times(2)).add(app) }
155 eventually { app: AppDefinition => verify(f.queue, times(2)).addAsync(app) }
jeschkiesUnsubmitted
Not Done

With D868 this can be verify(f.queue, withinTimeout().times(2)).addAsync(app). I wonder where eventually is getting the app parameter.

148156
149157 promise.future.futureValue
150158
151159 eventually {
152160 f.killService.numKilled should be(3)
153161 }
154162 verify(f.queue).resetDelay(newApp)
155163
Show All 9 Lines
165173 healthChecks = Set(MarathonHttpHealthCheck(portIndex = Some(PortReference(0)))),
166174 upgradeStrategy = UpgradeStrategy(0.5)
167175 )
168176
169177 val instanceA = f.runningInstance(app)
170178 val instanceB = f.runningInstance(app)
171179 val instanceC = f.runningInstance(app)
172180
173 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceB, instanceC))
181 f.tracker.specInstancesSync(app.id) returns Seq(instanceA, instanceB, instanceC)
174182
175183 val promise = Promise[Unit]()
176184 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
185 f.queue.addAsync(newApp, 3) returns Future.successful(Done)
177186 val ref = f.replaceActor(newApp, promise)
178187 watch(ref)
179188
180189 // all new tasks are queued directly
181 eventually { app: AppDefinition => verify(f.queue, times(3)).add(app) }
190 eventually { app: AppDefinition => verify(f.queue, times(3)).addAsync(app) }
182191
183192 // ceiling(minimumHealthCapacity * 3) = 2 are left running
184193 assert(f.killService.numKilled == 1)
185194
186195 // first new task becomes healthy and another old task is killed
187196 ref ! f.healthChanged(newApp, healthy = true)
188197 eventually {
189198 f.killService.numKilled should be(2)
Show All 28 Lines
218227 healthChecks = Set(MarathonHttpHealthCheck(portIndex = Some(PortReference(0)))),
219228 upgradeStrategy = UpgradeStrategy(0.5, 0.0)
220229 )
221230
222231 val instanceA = f.runningInstance(app)
223232 val instanceB = f.runningInstance(app)
224233 val instanceC = f.runningInstance(app)
225234
226 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceB, instanceC))
235 f.tracker.specInstancesSync(app.id) returns Seq(instanceA, instanceB, instanceC)
227236
228237 val promise = Promise[Unit]()
229238 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
239 f.queue.addAsync(newApp, 1) returns Future.successful(Done)
230240 val ref = f.replaceActor(newApp, promise)
231241 watch(ref)
232242
233243 // only one task is queued directly
234244 val queueOrder = org.mockito.Mockito.inOrder(f.queue)
235245 eventually {
236 queueOrder.verify(f.queue).add(_: AppDefinition, 1)
246 queueOrder.verify(f.queue).addAsync(_: AppDefinition, 1)
237247 }
238248
239249 // ceiling(minimumHealthCapacity * 3) = 2 are left running
240250 assert(f.killService.numKilled == 1)
241251
242252 // first new task becomes healthy and another old task is killed
243253 ref ! f.healthChanged(newApp, healthy = true)
244254 eventually {
245255 f.killService.numKilled should be(2)
246256 }
247257 eventually {
248 queueOrder.verify(f.queue).add(_: AppDefinition, 1)
258 queueOrder.verify(f.queue).addAsync(_: AppDefinition, 1)
249259 }
250260
251261 // second new task becomes healthy and the last old task is killed
252262 ref ! f.healthChanged(newApp, healthy = true)
253263 eventually {
254264 f.killService.numKilled should be(3)
255265 }
256266 eventually {
257 queueOrder.verify(f.queue).add(_: AppDefinition, 1)
267 queueOrder.verify(f.queue).addAsync(_: AppDefinition, 1)
258268 }
259269
260270 // third new task becomes healthy
261271 ref ! f.healthChanged(newApp, healthy = true)
262272 f.killService.numKilled should be(3)
263273
264274 promise.future.futureValue
265275
Show All 15 Lines
281291 healthChecks = Set(MarathonHttpHealthCheck(portIndex = Some(PortReference(0)))),
282292 upgradeStrategy = UpgradeStrategy(1.0, 0.0) // 1 task over-capacity is ok
283293 )
284294
285295 val instanceA = f.runningInstance(app)
286296 val instanceB = f.runningInstance(app)
287297 val instanceC = f.runningInstance(app)
288298
289 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceB, instanceC))
299 f.tracker.specInstancesSync(app.id) returns Seq(instanceA, instanceB, instanceC)
290300
291301 val promise = Promise[Unit]()
292302 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
303 f.queue.addAsync(newApp, 1) returns Future.successful(Done)
293304 val ref = f.replaceActor(newApp, promise)
294305 watch(ref)
295306
296307 // only one task is queued directly, all old still running
297308 val queueOrder = org.mockito.Mockito.inOrder(f.queue)
298309 eventually {
299310 queueOrder.verify(f.queue).add(_: AppDefinition, 1)
300311 }
▲ Show 20 Lines • Show All 43 Lines • ▼ Show 20 Line(s)
344355 healthChecks = Set(MarathonHttpHealthCheck(portIndex = Some(PortReference(0)))),
345356 upgradeStrategy = UpgradeStrategy(1.0, 0.7)
346357 )
347358
348359 val instanceA = f.runningInstance(app)
349360 val instanceB = f.runningInstance(app)
350361 val instanceC = f.runningInstance(app)
351362
352 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceB, instanceC))
363 f.tracker.specInstancesSync(app.id) returns Seq(instanceA, instanceB, instanceC)
353364
354365 val promise = Promise[Unit]()
355366 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
367 f.queue.addAsync(eq(newApp), any) returns Future.successful(Done)
356368 val ref = f.replaceActor(newApp, promise)
357369 watch(ref)
358370
359371 // two tasks are queued directly, all old still running
360372 val queueOrder = org.mockito.Mockito.inOrder(f.queue)
361373 eventually {
362374 queueOrder.verify(f.queue).add(_: AppDefinition, 2)
363375 }
▲ Show 20 Lines • Show All 42 Lines • ▼ Show 20 Line(s)
406418 upgradeStrategy = UpgradeStrategy(minimumHealthCapacity = 1.0, maximumOverCapacity = 0.3)
407419 )
408420
409421 val instanceA = f.runningInstance(app)
410422 val instanceB = f.runningInstance(app)
411423 val instanceC = f.runningInstance(app)
412424 val instanceD = f.runningInstance(app)
413425
414 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceB, instanceC, instanceD))
415
416426 val promise = Promise[Unit]()
417427 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
428
429 f.tracker.specInstancesSync(app.id) returns Seq(instanceA, instanceB, instanceC, instanceD)
430 f.queue.addAsync(newApp, 1) returns Future.successful(Done)
431
418432 val ref = f.replaceActor(newApp, promise)
419433 watch(ref)
420434
421435 // one task is killed directly because we are over capacity
422 val order = org.mockito.Mockito.inOrder(f.queue)
436 eventually {
423437 f.killService.killed should contain(instanceA.instanceId)
438 }
424439
425440 // the kill is confirmed (see answer above) and the first new task is queued
426441 eventually {
427 order.verify(f.queue).add(newApp, 1)
442 verify(f.queue, times(1)).addAsync(newApp, 1)
428443 }
429444 assert(f.killService.numKilled == 1)
430445
431446 // first new task becomes healthy and another old task is killed
432447 ref ! f.healthChanged(newApp, healthy = true)
433448 eventually {
434449 f.killService.numKilled should be(2)
435450 }
436451 eventually {
437 order.verify(f.queue).add(newApp, 1)
452 verify(f.queue, times(2)).addAsync(newApp, 1)
438453 }
439454
440455 // second new task becomes healthy and another old task is killed
441456 ref ! f.healthChanged(newApp, healthy = true)
442457 eventually {
443458 f.killService.numKilled should be(3)
444459 }
445460 eventually {
446 order.verify(f.queue).add(newApp, 1)
461 verify(f.queue, times(3)).addAsync(newApp, 1)
447462 }
448463
449464 // third new task becomes healthy and last old task is killed
450465 ref ! f.healthChanged(newApp, healthy = true)
451466 eventually {
452467 f.killService.numKilled should be(4)
453468 }
454 eventually {
455 order.verify(f.queue, never).add(newApp, 1)
456 }
457469
458470 promise.future.futureValue
459471
460472 // all remaining old tasks are killed
461473 f.killService.killed should contain(instanceB.instanceId)
462474 f.killService.killed should contain(instanceC.instanceId)
463475 f.killService.killed should contain(instanceD.instanceId)
464476
477 verify(f.queue, times(3)).addAsync(newApp, 1)
478
465479 expectTerminated(ref)
466480 }
467481
468482 "If all tasks are replaced already, the actor stops immediately" in {
469483 Given("An app without health checks and readiness checks, as well as 2 tasks of this version")
470484 val f = new Fixture
471485 val app = AppDefinition(id = "/myApp".toPath, instances = 2)
472486 val instanceA = f.runningInstance(app)
473487 val instanceB = f.runningInstance(app)
474 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceB))
488 f.tracker.specInstancesSync(app.id) returns Seq(instanceA, instanceB)
475489 val promise = Promise[Unit]()
476490
477491 When("The replace actor is started")
478492 val ref = f.replaceActor(app, promise)
479493 watch(ref)
480494
481495 Then("The replace actor finishes immediately")
482496 expectTerminated(ref)
483497 promise.isCompleted should be(true)
484498 }
485499
486500 "If all tasks are replaced already, we will wait for the readiness checks" in {
487501 Given("An app without health checks but readiness checks, as well as 1 task of this version")
488502 val f = new Fixture
489503 val check = ReadinessCheck()
490504 val port = PortDefinition(0, name = Some(check.portName))
491505 val app = AppDefinition(id = "/myApp".toPath, instances = 1, portDefinitions = Seq(port), readinessChecks = Seq(check))
492506 val instance = f.runningInstance(app)
493 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instance))
507 f.tracker.specInstancesSync(app.id) returns Seq(instance)
494508 val readyCheck = Observable.from(instance.tasksMap.values.map(task => ReadinessCheckResult(check.name, task.taskId, ready = true, None)))
495 when(f.readinessCheckExecutor.execute(any[ReadinessCheckExecutor.ReadinessCheckSpec])).thenReturn(readyCheck)
509 f.readinessCheckExecutor.execute(any[ReadinessCheckExecutor.ReadinessCheckSpec]) returns readyCheck
496510 val promise = Promise[Unit]()
497511
498512 When("The replace actor is started")
499513 val ref = f.replaceActor(app, promise)
500514 watch(ref)
501515
502516 Then("It needs to wait for the readiness checks to pass")
503517 expectTerminated(ref)
Show All 9 Lines
513527 val app = AppDefinition(
514528 id = "/myApp".toPath,
515529 instances = 1,
516530 portDefinitions = Seq(port),
517531 readinessChecks = Seq(ready),
518532 healthChecks = Set(MarathonHttpHealthCheck())
519533 )
520534 val instance = f.runningInstance(app)
521 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instance))
535 f.tracker.specInstancesSync(app.id) returns Seq(instance)
522536 val readyCheck = Observable.from(instance.tasksMap.values.map(task => ReadinessCheckResult(ready.name, task.taskId, ready = true, None)))
523 when(f.readinessCheckExecutor.execute(any[ReadinessCheckExecutor.ReadinessCheckSpec])).thenReturn(readyCheck)
537 f.readinessCheckExecutor.execute(any[ReadinessCheckExecutor.ReadinessCheckSpec]) returns readyCheck
524538 val promise = Promise[Unit]()
525539
526540 When("The replace actor is started")
527541 val ref = f.replaceActor(app, promise)
528542 watch(ref)
529543 ref ! InstanceHealthChanged(instance.instanceId, app.version, app.id, healthy = Some(true))
530544
531545 Then("It needs to wait for the readiness checks to pass")
532546 expectTerminated(ref)
533547 promise.isCompleted should be(true)
534548 }
535549
536550 "Wait until the tasks are killed" in {
537551 val f = new Fixture
538552 val app = AppDefinition(
539553 id = "/myApp".toPath,
540554 instances = 5,
541555 versionInfo = VersionInfo.forNewConfig(Timestamp(0)),
542556 upgradeStrategy = UpgradeStrategy(0.0))
543557 val instanceA = f.runningInstance(app)
544558 val instanceB = f.runningInstance(app)
545559
546 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instanceA, instanceB))
560 f.tracker.specInstancesSync(app.id) returns Seq(instanceA, instanceB)
547561
548562 val promise = Promise[Unit]()
549563 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
564 f.queue.addAsync(newApp, 5) returns Future.successful(Done)
565
550566 val ref = f.replaceActor(newApp, promise)
551567 watch(ref)
552568
553569 for (_ <- 0 until newApp.instances)
554570 ref.receive(f.instanceChanged(newApp, Running))
555571
556572 verify(f.queue, Mockito.timeout(1000)).resetDelay(newApp)
557573 f.killService.killed should contain(instanceA.instanceId)
Show All 11 Lines
569585 versionInfo = VersionInfo.forNewConfig(Timestamp(0)),
570586 healthChecks = Set(MarathonHttpHealthCheck()),
571587 readinessChecks = Seq(ReadinessCheck()),
572588 upgradeStrategy = UpgradeStrategy(1.0, 1.0)
573589 )
574590
575591 val instance = f.runningInstance(app)
576592
577 when(f.tracker.specInstancesSync(app.id)).thenReturn(Seq(instance))
593 f.tracker.specInstancesSync(app.id) returns Seq(instance)
578594
579595 val promise = Promise[Unit]()
580596 val newApp = app.copy(versionInfo = VersionInfo.forNewConfig(Timestamp(1)))
597 f.queue.addAsync(newApp, 1) returns Future.successful(Done)
581598 val ref = f.replaceActor(newApp, promise)
582599 watch(ref)
583600
584601 // only one task is queued directly
585602 val queueOrder = org.mockito.Mockito.inOrder(f.queue)
586603 eventually {
587604 queueOrder.verify(f.queue).add(_: AppDefinition, 1)
588605 }
▲ Show 20 Lines • Show All 64 Lines • Show Last 20 Lines
View Options

src/test/scala/mesosphere/marathon/core/deployment/impl/TaskStartActorTest.scala

11package mesosphere.marathon
22package core.deployment.impl
33
4import akka.Done
5import akka.actor.{ OneForOneStrategy, Props, SupervisorStrategy }
6import akka.pattern.{ Backoff, BackoffSupervisor }
47import akka.testkit.{ TestActorRef, TestProbe }
58import mesosphere.AkkaUnitTest
69import mesosphere.marathon.core.condition.Condition
710import mesosphere.marathon.core.condition.Condition.{ Failed, Running }
811import mesosphere.marathon.core.event.{ DeploymentStatus, _ }
912import mesosphere.marathon.core.health.MesosCommandHealthCheck
10import mesosphere.marathon.core.instance.update.InstanceUpdateOperation
11import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
13import mesosphere.marathon.core.instance.Instance
1214import mesosphere.marathon.core.launcher.impl.LaunchQueueTestHelper
1315import mesosphere.marathon.core.launchqueue.LaunchQueue
14import mesosphere.marathon.core.leadership.AlwaysElectedLeadershipModule
1516import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
16import mesosphere.marathon.core.task.tracker.{ InstanceCreationHandler, InstanceTracker }
17import mesosphere.marathon.core.task.tracker.InstanceTracker
1718import mesosphere.marathon.state.PathId._
18import mesosphere.marathon.state.{ AppDefinition, Command, Timestamp }
19import mesosphere.marathon.test.MarathonTestHelper
20import org.mockito.Mockito.{ spy, when }
19import mesosphere.marathon.state.{ AppDefinition, Command }
20import org.scalatest.concurrent.Eventually
2121
22import scala.concurrent.Promise
22import scala.concurrent.{ Future, Promise }
23import scala.util.control.NonFatal
2324
24class TaskStartActorTest extends AkkaUnitTest {
25class TaskStartActorTest extends AkkaUnitTest with Eventually {
2526 "TaskStartActor" should {
2627 for (
2728 (counts, description) <- Seq(
2829 None -> "with no item in queue",
2930 Some(LaunchQueueTestHelper.zeroCounts) -> "with zero count queue item"
3031 )
3132 ) {
3233 s"Start success $description" in {
3334 val f = new Fixture
3435 val promise = Promise[Unit]()
jeschkiesUnsubmitted
Not Done

Should this be Promise[Done]?

zen-dogAuthorUnsubmitted
Not Done

No, that's a different promise - DeploymentActor becomes this one and completes it when the deployment is done. It should be also Promise[Done] but that has nothing to do with current diff. I can make it a separate one.

3536 val app = AppDefinition("/myApp".toPath, instances = 5)
3637
37 when(f.launchQueue.get(app.id)).thenReturn(counts)
38 f.launchQueue.getAsync(app.id) returns Future.successful(counts)
39 f.taskTracker.countLaunchedSpecInstances(app.id) returns Future.successful(0)
3840 val ref = f.startActor(app, app.instances, promise)
3941 watch(ref)
4042
41 verify(f.launchQueue, timeout(3000)).add(app, app.instances)
43 eventually { verify(f.launchQueue, atLeastOnce).addAsync(app, app.instances) }
4244
4345 for (i <- 0 until app.instances)
4446 system.eventStream.publish(f.instanceChange(app, Instance.Id.forRunSpec(app.id), Running))
4547
4648 promise.future.futureValue should be(())
4749
4850 expectTerminated(ref)
4951 }
5052 }
5153
5254 "Start success with one task left to launch" in {
5355 val f = new Fixture
5456 val counts = Some(LaunchQueueTestHelper.zeroCounts.copy(instancesLeftToLaunch = 1, finalInstanceCount = 1))
5557 val promise = Promise[Unit]()
5658 val app = AppDefinition("/myApp".toPath, instances = 5)
5759
58 when(f.launchQueue.get(app.id)).thenReturn(counts)
60 f.launchQueue.getAsync(app.id) returns Future.successful(counts)
5961
6062 val ref = f.startActor(app, app.instances, promise)
6163 watch(ref)
6264
63 verify(f.launchQueue, timeout(3000)).add(app, app.instances - 1)
65 eventually { verify(f.launchQueue, atLeastOnce).addAsync(app, app.instances - 1) }
6466
6567 for (i <- 0 until (app.instances - 1))
6668 system.eventStream.publish(f.instanceChange(app, Instance.Id(s"task-$i"), Running))
6769
6870 promise.future.futureValue should be(())
6971
7072 expectTerminated(ref)
7173 }
7274
7375 "Start success with existing task in launch queue" in {
7476 val f = new Fixture
7577 val promise = Promise[Unit]()
7678 val app = AppDefinition("/myApp".toPath, instances = 5)
7779
78 when(f.launchQueue.get(app.id)).thenReturn(None)
79 val instance = TestInstanceBuilder.newBuilder(app.id, version = Timestamp(1024)).addTaskStarting().getInstance()
80 f.taskCreationHandler.created(InstanceUpdateOperation.LaunchEphemeral(instance)).futureValue
80 f.launchQueue.getAsync(app.id) returns Future.successful(None)
81 f.taskTracker.countLaunchedSpecInstances(app.id) returns Future.successful(1)
8182
8283 val ref = f.startActor(app, app.instances, promise)
8384 watch(ref)
8485
85 verify(f.launchQueue, timeout(3000)).add(app, app.instances - 1)
86 eventually { verify(f.launchQueue, atLeastOnce).addAsync(app, app.instances - 1) }
8687
8788 for (i <- 0 until (app.instances - 1))
8889 system.eventStream.publish(f.instanceChange(app, Instance.Id(s"task-$i"), Running))
8990
9091 promise.future.futureValue should be(())
9192
9293 expectTerminated(ref)
9394 }
9495
9596 "Start success with no instances to start" in {
9697 val f = new Fixture
9798 val promise = Promise[Unit]()
9899 val app = AppDefinition("/myApp".toPath, instances = 0)
99 when(f.launchQueue.get(app.id)).thenReturn(None)
100 f.launchQueue.getAsync(app.id) returns Future.successful(None)
101 f.taskTracker.countLaunchedSpecInstances(app.id) returns Future.successful(0)
100102
101103 val ref = f.startActor(app, app.instances, promise)
102104 watch(ref)
103105
104106 promise.future.futureValue should be(())
105107
106108 expectTerminated(ref)
107109 }
108110
109111 "Start with health checks" in {
110112 val f = new Fixture
111113 val promise = Promise[Unit]()
112114 val app = AppDefinition(
113115 "/myApp".toPath,
114116 instances = 5,
115117 healthChecks = Set(MesosCommandHealthCheck(command = Command("true")))
116118 )
117 when(f.launchQueue.get(app.id)).thenReturn(None)
119 f.launchQueue.getAsync(app.id) returns Future.successful(None)
120 f.taskTracker.countLaunchedSpecInstances(app.id) returns Future.successful(0)
118121
119122 val ref = f.startActor(app, app.instances, promise)
120123 watch(ref)
121124
122 verify(f.launchQueue, timeout(3000)).add(app, app.instances)
125 eventually { verify(f.launchQueue, atLeastOnce).addAsync(app, app.instances) }
123126
124127 for (i <- 0 until app.instances)
125128 system.eventStream.publish(f.healthChange(app, Instance.Id(s"task_$i"), healthy = true))
126129
127130 promise.future.futureValue should be(())
128131
129132 expectTerminated(ref)
130133 }
131134
132135 "Start with health checks with no instances to start" in {
133136 val f = new Fixture
134137 val promise = Promise[Unit]()
135138 val app = AppDefinition(
136139 "/myApp".toPath,
137140 instances = 0,
138141 healthChecks = Set(MesosCommandHealthCheck(command = Command("true")))
139142 )
140 when(f.launchQueue.get(app.id)).thenReturn(None)
143 f.launchQueue.getAsync(app.id) returns Future.successful(None)
144 f.taskTracker.countLaunchedSpecInstances(app.id) returns Future.successful(0)
141145
142146 val ref = f.startActor(app, app.instances, promise)
143147 watch(ref)
144148
145149 promise.future.futureValue should be(())
146150
147151 expectTerminated(ref)
148152 }
149153
150154 "Task fails to start" in {
151155 val f = new Fixture
152156 val promise = Promise[Unit]()
153157 val app = AppDefinition("/myApp".toPath, instances = 1)
154158
155 when(f.launchQueue.get(app.id)).thenReturn(None)
159 f.launchQueue.getAsync(app.id) returns Future.successful(None)
160 f.taskTracker.countLaunchedSpecInstances(app.id) returns Future.successful(0)
161
156162 val ref = f.startActor(app, app.instances, promise)
157163 watch(ref)
158164
159 verify(f.launchQueue, timeout(3000)).add(app, app.instances)
165 eventually { verify(f.launchQueue, atLeastOnce).addAsync(app, app.instances) }
160166
161167 system.eventStream.publish(f.instanceChange(app, Instance.Id.forRunSpec(app.id), Failed))
162168
163 verify(f.launchQueue, timeout(3000)).add(app, 1)
169 eventually { verify(f.launchQueue, atLeastOnce).addAsync(app, 1) }
164170
165171 for (i <- 0 until app.instances)
166172 system.eventStream.publish(f.instanceChange(app, Instance.Id.forRunSpec(app.id), Running))
167173
168174 promise.future.futureValue should be(())
169175
170176 expectTerminated(ref)
171177 }
172178
173 "Start success with dying existing task, reschedules, but finishes early" in {
179 // This is a dumb test - we're verifying that the actor called methods we programmed it to call.
180 // However since we plan to replace Deployments with Actions anyway, I'll not going to start a rewrite.
181 "Start success with dying existing task, reschedules and finishes" in {
174182 val f = new Fixture
175183 val promise = Promise[Unit]()
176184 val app = AppDefinition("/myApp".toPath, instances = 5)
177 when(f.launchQueue.get(app.id)).thenReturn(None)
178185
179 val outdatedInstance = TestInstanceBuilder.newBuilder(app.id, version = Timestamp(1024)).addTaskStaged().getInstance()
180 val instanceId = outdatedInstance.instanceId
181 f.taskCreationHandler.created(InstanceUpdateOperation.LaunchEphemeral(outdatedInstance)).futureValue
186 f.launchQueue.getAsync(app.id) returns Future.successful(None)
187 f.taskTracker.countLaunchedSpecInstances(app.id) returns Future.successful(1)
182188
183189 val ref = f.startActor(app, app.instances, promise)
184190 watch(ref)
185191
186 // wait for initial sync
187 verify(f.launchQueue, timeout(3000)).get(app.id)
188 verify(f.launchQueue, timeout(3000)).add(app, app.instances - 1)
192 // 4 initial instances should be added to the launch queue
193 eventually { verify(f.launchQueue, atLeastOnce).addAsync(eq(app), any) }
189194
190 noMoreInteractions(f.launchQueue)
191 reset(f.launchQueue)
192
193195 // let existing task die
194 when(f.taskTracker.countLaunchedSpecInstancesSync(app.id)).thenReturn(0)
195 when(f.launchQueue.get(app.id)).thenReturn(Some(LaunchQueueTestHelper.zeroCounts.copy(instancesLeftToLaunch = 4, finalInstanceCount = 4)))
196 // The version does not match the app.version so that it is filtered in StartingBehavior.
197 // does that make sense?
198 val (_, outdatedTask) = outdatedInstance.tasksMap.head
199 system.eventStream.publish(f.instanceChange(app, instanceId, Condition.Error).copy(runSpecVersion = outdatedTask.runSpecVersion))
196 f.taskTracker.countLaunchedSpecInstances(app.id) returns Future.successful(0)
197 f.launchQueue.getAsync(app.id) returns Future.successful(Some(LaunchQueueTestHelper.zeroCounts.copy(instancesLeftToLaunch = 4, finalInstanceCount = 4)))
198 system.eventStream.publish(f.instanceChange(app, Instance.Id("task-4"), Condition.Error))
200199
201 // sync will reschedule task
200 // trigger a Sync and wait for another task to be added to the launch queue
202201 ref ! StartingBehavior.Sync
203 verify(f.launchQueue, timeout(3000)).get(app.id)
204 verify(f.launchQueue, timeout(3000)).add(app, 1)
202 eventually { verify(f.launchQueue, times(4)).addAsync(eq(app), any) }
205203
206 noMoreInteractions(f.launchQueue)
207 reset(f.launchQueue)
208
209 // launch 4 of the tasks
210 when(f.launchQueue.get(app.id)).thenReturn(Some(LaunchQueueTestHelper.zeroCounts.copy(instancesLeftToLaunch = app.instances, finalInstanceCount = 4)))
211 when(f.taskTracker.countLaunchedSpecInstancesSync(app.id)).thenReturn(4)
204 // let 4 other tasks start successfully
212205 List(0, 1, 2, 3) foreach { i =>
213206 system.eventStream.publish(f.instanceChange(app, Instance.Id(s"task-$i"), Running))
214207 }
215208
216 // it finished early
209 // and make sure that the actor should finishes
217210 promise.future.futureValue should be(())
218211
219 noMoreInteractions(f.launchQueue)
220
221212 expectTerminated(ref)
222213 }
223214 }
224215 class Fixture {
225216
226217 val scheduler: SchedulerActions = mock[SchedulerActions]
227218 val launchQueue: LaunchQueue = mock[LaunchQueue]
228 val leadershipModule = AlwaysElectedLeadershipModule.forRefFactory(system)
229 val taskTrackerModule = MarathonTestHelper.createTaskTrackerModule(leadershipModule)
230 val taskTracker: InstanceTracker = spy(taskTrackerModule.instanceTracker)
231 val taskCreationHandler: InstanceCreationHandler = taskTrackerModule.instanceCreationHandler
219 val taskTracker: InstanceTracker = mock[InstanceTracker]
232220 val deploymentManager = TestProbe()
233221 val status: DeploymentStatus = mock[DeploymentStatus]
234222 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
235223
224 launchQueue.addAsync(any, any) returns Future.successful(Done)
225
236226 def instanceChange(app: AppDefinition, id: Instance.Id, condition: Condition): InstanceChanged = {
237227 val instance: Instance = mock[Instance]
238228 instance.instanceId returns id
239229 InstanceChanged(id, app.version, app.id, condition, instance)
240230 }
241231
242232 def healthChange(app: AppDefinition, id: Instance.Id, healthy: Boolean): InstanceHealthChanged = {
243233 InstanceHealthChanged(id, app.version, app.id, Some(healthy))
244234 }
245235
246236 def startActor(app: AppDefinition, scaleTo: Int, promise: Promise[Unit]): TestActorRef[TaskStartActor] =
247 TestActorRef(TaskStartActor.props(
237 TestActorRef(childSupervisor(TaskStartActor.props(
248238 deploymentManager.ref, status, scheduler, launchQueue, taskTracker, system.eventStream, readinessCheckExecutor,
249 app, scaleTo, promise))
239 app, scaleTo, promise), "Test-TaskStartActor"))
240
241 // Prevents the TaskActor from restarting too many times (filling the log with exceptions) similar to how it's
242 // parent actor (DeploymentActor) does it.
243 def childSupervisor(props: Props, name: String): Props = {
244 import scala.concurrent.duration._
245
246 BackoffSupervisor.props(
247 Backoff.onFailure(
248 childProps = props,
249 childName = name,
250 minBackoff = 5.seconds,
251 maxBackoff = 30.seconds,
252 randomFactor = 0.2 // adds 20% "noise" to vary the intervals slightly
253 ).withSupervisorStrategy(
254 OneForOneStrategy() {
255 case NonFatal(_) => SupervisorStrategy.Restart
jdefUnsubmitted
Done

NonFatal?

256 case _ => SupervisorStrategy.Escalate
257 }
258 ))
259 }
250260 }
251261}