HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D403

Fixes #4921 by giving current running instances to the AppStartActor, instead of requesting this list inside the actor.
ClosedAll Users
Actions

Authored by aquamatthias on Jan 6 2017, 3:47 PM.

Details

Reviewers
timcharper
janisz
jeschkies
Commits
rMARATHONf43e91e5c888: Fixes #4921 by giving current running instances to the AppStartActor, instead…
rMARATHON5f027d54d2f5: Fixes #4921 by giving current running instances to the AppStartActor, instead…
rMARATHONe5e3acc7773b: Fixes #4921 by giving current running instances to the AppStartActor… (#5323)
rMARATHONaffe91aa4eb9: Fixes #4921 by giving current running instances to the AppStartActor, instead…
JIRA Issues
JIRA MARATHON-1853 AppStartActorTest is unhealthy
Summary

This list has to be known before the actor actually starts.

Test Plan

sbt 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.
Changes from before your most recent comment are hidden. Show Older Changes
jeschkies accepted this revision.Jan 6 2017, 5:37 PM

Nice! Thanks.

This revision is now accepted and ready to land.Jan 6 2017, 5:37 PM
janisz requested changes to this revision.Jan 6 2017, 8:24 PM
janisz added a reviewer: janisz.
janisz added a subscriber: janisz.
janisz added inline comments.
src/main/scala/mesosphere/marathon/upgrade/StartingBehavior.scala
29–30

No longer needed

This revision now requires changes to proceed.Jan 6 2017, 8:24 PM
aquamatthias updated this revision to Diff 1523.Jan 9 2017, 9:16 AM
aquamatthias marked an inline comment as done.
  • Remove not needed SuppressWarnings.
timcharper accepted this revision.Jan 9 2017, 10:59 AM
janisz accepted this revision.Jan 9 2017, 11:23 AM
This revision is now accepted and ready to land.Jan 9 2017, 11:23 AM
Closed by commit rMARATHONaffe91aa4eb9: Fixes #4921 by giving current running instances to the AppStartActor, instead… (authored by aquamatthias, committed by jeschkies). · Explain WhyJan 9 2017, 11:55 AM
This revision was automatically updated to reflect the committed changes.
timcharper added 1 JIRA issue(s): MARATHON-1853.Mar 22 2017, 1:19 AM

Revision Contents

Diff 1526

View Options

src/main/scala/mesosphere/marathon/upgrade/AppStartActor.scala

11package mesosphere.marathon.upgrade
22
33import akka.Done
44import akka.actor._
55import akka.event.EventStream
66import com.typesafe.scalalogging.StrictLogging
77import mesosphere.marathon.core.event.DeploymentStatus
8import mesosphere.marathon.core.instance.Instance
89import mesosphere.marathon.core.launchqueue.LaunchQueue
910import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1011import mesosphere.marathon.core.task.tracker.InstanceTracker
1112import mesosphere.marathon.state.RunSpec
1213import mesosphere.marathon.{ AppStartCanceledException, SchedulerActions }
1314
14import scala.async.Async.{ async, await }
15import scala.concurrent.ExecutionContext.Implicits.global
16import scala.concurrent.{ Future, Promise }
15import scala.concurrent.Promise
1716import scala.util.control.NonFatal
1817
1918class AppStartActor(
2019 val deploymentManager: ActorRef,
2120 val status: DeploymentStatus,
2221 val scheduler: SchedulerActions,
2322 val launchQueue: LaunchQueue,
2423 val instanceTracker: InstanceTracker,
2524 val eventBus: EventStream,
2625 val readinessCheckExecutor: ReadinessCheckExecutor,
2726 val runSpec: RunSpec,
2827 val scaleTo: Int,
28 currentInstances: Seq[Instance],
2929 promise: Promise[Unit]) extends Actor with StartingBehavior with StrictLogging {
3030
3131 override val nrToStart: Int = scaleTo
3232
33 @SuppressWarnings(Array("all")) // async/await
34 override def initializeStart(): Future[Done] = async {
33 override def initializeStart(): Unit = {
3534 // In case we already have running instances (can happen on master abdication during deployment)
3635 // with the correct version those will not be killed.
37 val runningInstances = await(instanceTracker.specInstances(runSpec.id)).count(_.isActive)
36 val runningInstances = currentInstances.count(_.isActive)
3837 scheduler.startRunSpec(runSpec.withInstances(Math.max(runningInstances, nrToStart)))
3938 Done
4039 }
4140
4241 override def postStop(): Unit = {
4342 eventBus.unsubscribe(self)
4443 super.postStop()
4544 }
Show All 21 Lines
6766 status: DeploymentStatus,
6867 scheduler: SchedulerActions,
6968 launchQueue: LaunchQueue,
7069 taskTracker: InstanceTracker,
7170 eventBus: EventStream,
7271 readinessCheckExecutor: ReadinessCheckExecutor,
7372 runSpec: RunSpec,
7473 scaleTo: Int,
74 currentInstances: Seq[Instance],
7575 promise: Promise[Unit]): Props = {
7676 Props(new AppStartActor(deploymentManager, status, scheduler, launchQueue, taskTracker, eventBus,
77 readinessCheckExecutor, runSpec, scaleTo, promise))
77 readinessCheckExecutor, runSpec, scaleTo, currentInstances, promise))
7878 }
7979}
View Options

src/main/scala/mesosphere/marathon/upgrade/DeploymentActor.scala

Show First 20 Lines • Show All 126 Lines • ▼ Show 20 Line(s)
127127 }
128128 }
129129 }
130130
131131 // scalastyle:on
132132
133133 def startRunnable(runnableSpec: RunSpec, scaleTo: Int, status: DeploymentStatus): Future[Unit] = {
134134 val promise = Promise[Unit]()
135 instanceTracker.specInstances(runnableSpec.id).map { instances =>
135136 context.actorOf(AppStartActor.props(deploymentManager, status, scheduler, launchQueue, instanceTracker,
136 eventBus, readinessCheckExecutor, runnableSpec, scaleTo, promise))
137 eventBus, readinessCheckExecutor, runnableSpec, scaleTo, instances, promise))
138 }
137139 promise.future
138140 }
139141
140142 @SuppressWarnings(Array("all")) /* async/await */
141143 def scaleRunnable(runnableSpec: RunSpec, scaleTo: Int,
142144 toKill: Option[Seq[Instance]],
143145 status: DeploymentStatus): Future[Unit] = {
144146 logger.debug("Scale runnable {}", runnableSpec)
▲ Show 20 Lines • Show All 105 Lines • Show Last 20 Lines
View Options

src/main/scala/mesosphere/marathon/upgrade/StartingBehavior.scala

11package mesosphere.marathon.upgrade
22
3import akka.Done
43import akka.actor.Actor
54import akka.event.EventStream
65import mesosphere.marathon.SchedulerActions
76import mesosphere.marathon.core.condition.Condition.Terminal
87import mesosphere.marathon.core.event.{ InstanceChanged, InstanceHealthChanged }
98import mesosphere.marathon.core.instance.Instance
109import mesosphere.marathon.core.launchqueue.LaunchQueue
1110import mesosphere.marathon.core.task.tracker.InstanceTracker
1211import org.slf4j.LoggerFactory
1312
14import scala.async.Async.{ async, await }
15import scala.concurrent.Future
1613import scala.concurrent.duration._
1714
1815trait StartingBehavior extends ReadinessBehavior { this: Actor =>
1916 import context.dispatcher
2017 import mesosphere.marathon.upgrade.StartingBehavior._
2118
2219 def eventBus: EventStream
2320 def scaleTo: Int
2421 def nrToStart: Int
2522 def launchQueue: LaunchQueue
2623 def scheduler: SchedulerActions
2724 def instanceTracker: InstanceTracker
2825
29 def initializeStart(): Future[Done]
26 def initializeStart(): Unit
3027
3128 private[this] val log = LoggerFactory.getLogger(getClass)
3229
33 @SuppressWarnings(Array("all")) // async/await
3430 final override def preStart(): Unit = {
janiszUnsubmitted
Done

No longer needed

3531 if (hasHealthChecks) eventBus.subscribe(self, classOf[InstanceHealthChanged])
3632 eventBus.subscribe(self, classOf[InstanceChanged])
3733
38 async {
39 await(initializeStart())
34 initializeStart()
4035 checkFinished()
41 //TODO: make this value configurable
4236 context.system.scheduler.scheduleOnce(1.seconds, self, Sync)
43 }
4437 }
4538
4639 final override def receive: Receive = readinessBehavior orElse commonBehavior
4740
4841 def commonBehavior: Receive = {
4942 case InstanceChanged(id, `version`, `pathId`, _: Terminal, _) =>
5043 log.warn(s"New instance [$id] failed during app ${runSpec.id.toString} scaling, queueing another instance")
5144 instanceTerminated(id)
Show All 36 Lines
View Options

src/main/scala/mesosphere/marathon/upgrade/TaskStartActor.scala

11
22package mesosphere.marathon.upgrade
33
4import akka.Done
54import akka.actor.{ Actor, ActorLogging, ActorRef, Props }
65import akka.event.EventStream
76import mesosphere.marathon.core.event.DeploymentStatus
87import mesosphere.marathon.core.launchqueue.LaunchQueue
98import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
109import mesosphere.marathon.core.task.tracker.InstanceTracker
1110import mesosphere.marathon.state.RunSpec
1211import mesosphere.marathon.{ SchedulerActions, TaskUpgradeCanceledException }
1312
14import scala.concurrent.{ Future, Promise }
13import scala.concurrent.Promise
1514
1615class TaskStartActor(
1716 val deploymentManager: ActorRef,
1817 val status: DeploymentStatus,
1918 val scheduler: SchedulerActions,
2019 val launchQueue: LaunchQueue,
2120 val instanceTracker: InstanceTracker,
2221 val eventBus: EventStream,
2322 val readinessCheckExecutor: ReadinessCheckExecutor,
2423 val runSpec: RunSpec,
2524 val scaleTo: Int,
2625 promise: Promise[Unit]) extends Actor with ActorLogging with StartingBehavior {
2726
2827 val nrToStart: Int = Math.max(
2928 0,
3029 scaleTo - launchQueue.get(runSpec.id).map(_.finalInstanceCount)
3130 .getOrElse(instanceTracker.countLaunchedSpecInstancesSync(runSpec.id)))
3231
33 def initializeStart(): Future[Done] = Future.successful{
34 if (nrToStart > 0)
35 launchQueue.add(runSpec, nrToStart)
36 Done
37 }
32 def initializeStart(): Unit = if (nrToStart > 0) launchQueue.add(runSpec, nrToStart)
3833
3934 override def postStop(): Unit = {
4035 eventBus.unsubscribe(self)
4136 super.postStop()
4237 }
4338
4439 override def success(): Unit = {
4540 log.info(s"Successfully started $nrToStart instances of ${runSpec.id}")
Show All 31 Lines
View Options

src/test/scala/mesosphere/marathon/upgrade/AppStartActorTest.scala

1package mesosphere.marathon.upgrade
1package mesosphere.marathon
2package upgrade
23
34import akka.testkit.{ TestActorRef, TestProbe }
4import mesosphere.Unstable
55import mesosphere.marathon.core.condition.Condition
66import mesosphere.marathon.core.event.{ DeploymentStatus, InstanceChanged, InstanceHealthChanged }
77import mesosphere.marathon.core.health.{ MarathonHttpHealthCheck, PortReference }
88import mesosphere.marathon.core.instance.Instance
99import mesosphere.marathon.core.launchqueue.LaunchQueue
1010import mesosphere.marathon.core.leadership.AlwaysElectedLeadershipModule
1111import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1212import mesosphere.marathon.core.task.tracker.InstanceTracker
1313import mesosphere.marathon.state.{ AppDefinition, PathId }
1414import mesosphere.marathon.test.{ MarathonActorSupport, MarathonSpec, MarathonTestHelper, Mockito }
1515import mesosphere.marathon.{ AppStartCanceledException, SchedulerActions }
1616import org.scalatest.{ BeforeAndAfter, Matchers }
1717
1818import scala.concurrent.duration._
1919import scala.concurrent.{ Await, Future, Promise }
2020
2121class AppStartActorTest
2222 extends MarathonActorSupport
2323 with MarathonSpec
2424 with Matchers
2525 with BeforeAndAfter
2626 with Mockito {
2727
28 test("Without Health Checks", Unstable) {
28 test("Without Health Checks") {
2929 val f = new Fixture
30 val app = AppDefinition(id = PathId("/app"), instances = 10)
30 val app = AppDefinition(id = f.appId, instances = 10)
3131 val promise = Promise[Unit]()
3232 val ref = f.startActor(app, scaleTo = 2, promise)
3333 watch(ref)
3434
3535 system.eventStream.publish(f.instanceChanged(app, Condition.Running)) // linter:ignore:IdenticalStatements
3636 system.eventStream.publish(f.instanceChanged(app, Condition.Running))
3737
3838 Await.result(promise.future, 5.seconds)
3939
4040 verify(f.scheduler).startRunSpec(app.copy(instances = 2))
4141 expectTerminated(ref)
4242 }
4343
44 test("With Health Checks", Unstable) {
44 test("With Health Checks") {
4545 val f = new Fixture
4646 val app = AppDefinition(
47 id = PathId("/app"),
47 id = f.appId,
4848 instances = 10,
4949 healthChecks = Set(MarathonHttpHealthCheck(portIndex = Some(PortReference(0)))))
5050 val promise = Promise[Unit]()
5151 val ref = f.startActor(app, scaleTo = 2, promise)
5252 watch(ref)
5353
5454 system.eventStream.publish(f.healthChanged(app, healthy = true)) // linter:ignore:IdenticalStatements
5555 system.eventStream.publish(f.healthChanged(app, healthy = true))
5656
5757 Await.result(promise.future, 5.seconds)
5858
5959 verify(f.scheduler).startRunSpec(app.copy(instances = 2))
6060 expectTerminated(ref)
6161 }
6262
63 test("Failed", Unstable) {
63 test("Failed") {
6464 val f = new Fixture
6565 f.scheduler.stopRunSpec(any).asInstanceOf[Future[Unit]] returns Future.successful(())
6666
67 val app = AppDefinition(id = PathId("/app"), instances = 10)
67 val app = AppDefinition(id = f.appId, instances = 10)
6868 val promise = Promise[Unit]()
6969 val ref = f.startActor(app, scaleTo = 2, promise)
7070 watch(ref)
7171
7272 ref ! DeploymentActor.Shutdown
7373
7474 intercept[AppStartCanceledException] {
7575 Await.result(promise.future, 5.seconds)
7676 }
7777
7878 verify(f.scheduler).startRunSpec(app.copy(instances = 2))
7979 verify(f.scheduler).stopRunSpec(app)
8080 expectTerminated(ref)
8181 }
8282
83 test("No tasks to start without health checks", Unstable) {
83 test("No tasks to start without health checks") {
8484 val f = new Fixture
85 val app = AppDefinition(id = PathId("/app"), instances = 10)
85 val app = AppDefinition(id = f.appId, instances = 10)
8686 val promise = Promise[Unit]()
8787 val ref = f.startActor(app, scaleTo = 0, promise)
8888 watch(ref)
8989
9090 Await.result(promise.future, 5.seconds)
9191
9292 verify(f.scheduler).startRunSpec(app.copy(instances = 0))
9393 expectTerminated(ref)
9494 }
9595
96 test("No tasks to start with health checks", Unstable) {
96 test("No tasks to start with health checks") {
9797 val f = new Fixture
9898 val app = AppDefinition(
99 id = PathId("/app"),
99 id = f.appId,
100100 instances = 10,
101101 healthChecks = Set(MarathonHttpHealthCheck(portIndex = Some(PortReference(0)))))
102102 val promise = Promise[Unit]()
103103 val ref = f.startActor(app, scaleTo = 0, promise)
104104 watch(ref)
105105
106106 Await.result(promise.future, 5.seconds)
107107
108108 verify(f.scheduler).startRunSpec(app.copy(instances = 0))
109109 expectTerminated(ref)
110110 }
111111
112112 class Fixture {
113113
114114 val scheduler: SchedulerActions = mock[SchedulerActions]
115115 val launchQueue: LaunchQueue = mock[LaunchQueue]
116 val taskTracker: InstanceTracker = MarathonTestHelper.createTaskTracker(AlwaysElectedLeadershipModule.forActorSystem(system))
116 val instanceTracker: InstanceTracker = MarathonTestHelper.createTaskTracker(AlwaysElectedLeadershipModule.forActorSystem(system))
117117 val deploymentManager: TestProbe = TestProbe()
118118 val deploymentStatus: DeploymentStatus = mock[DeploymentStatus]
119119 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
120 val appId = PathId("/app")
120121
122 launchQueue.get(appId) returns None
123
121124 def instanceChanged(app: AppDefinition, condition: Condition): InstanceChanged = {
122125 val instanceId = Instance.Id.forRunSpec(app.id)
123126 val instance: Instance = mock[Instance]
124127 instance.instanceId returns instanceId
125128 InstanceChanged(instanceId, app.version, app.id, condition, instance)
126129 }
127130
128131 def healthChanged(app: AppDefinition, healthy: Boolean): InstanceHealthChanged = {
129132 InstanceHealthChanged(Instance.Id.forRunSpec(app.id), app.version, app.id, healthy = Some(healthy))
130133 }
131134
132135 def startActor(app: AppDefinition, scaleTo: Int, promise: Promise[Unit]): TestActorRef[AppStartActor] =
133136 TestActorRef(AppStartActor.props(deploymentManager.ref, deploymentStatus, scheduler,
134 launchQueue, taskTracker, system.eventStream, readinessCheckExecutor, app, scaleTo, promise)
137 launchQueue, instanceTracker, system.eventStream, readinessCheckExecutor, app, scaleTo, Seq.empty, promise)
135138 )
136139 }
137140}