HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D195

Rework DeploymentManager logic
ClosedAll Users
Actions

Authored by zen-dog on Nov 18 2016, 5:53 PM.

Details

Reviewers
aquamatthias
meichstedt
Commits
rMARATHON95969d4ec96d: DeploymentManagers handles leader abdication correctly
rMARATHONa8bac97b8bf0: Rework DeploymentManager logic
rMARATHON3edb7b76bb6c: Remove SchedulerDriver from the PerformDeployment message
rMARATHON1a4d91cddd63: Rework DeploymentManager logic
rMARATHON07a96bff38a4: Fixed small typo in the DeploymentManagerTest
rMARATHON77480be2f189: Implemented part of the review change requests
rMARATHON0757f9ae7cfc: Rework DeploymentManager logic
rMARATHON6ac141df7dff: Remove SchedulerDriver from the PerformDeployment message
rMARATHONc571775db076: Fix MarathonSchedulerActor "Cancellation timeout" test
rMARATHON1b3b8ee4dd79: Fix MarathonSchedulerActor "Cancellation timeout" test
rMARATHONfcf377b9f45f: Rework DeploymentManager logic
rMARATHON0512601d1f99: DeploymentManagers handles leader abdication correctly
rMARATHON25c3b470fae0: Fix MarathonSchedulerActor "Cancellation timeout" test
rMARATHON4ff8a81fb223: Rework DeploymentManager logic
rMARATHON902525004ad1: Fix MarathonSchedulerActor "Cancellation timeout" test
rMARATHONe21dfeab7024: Rebased on master and implemented review change requests
rMARATHON11769f986b47: Fixed small typo in the DeploymentManagerTest
rMARATHON5ef326a717de: Rebased on master and implemented review change requests
rMARATHON29a981004e3c: Rework DeploymentManager logic
rMARATHONe0aa5d87671c: Fix MarathonSchedulerActor "Cancellation timeout" test
rMARATHONe7b3f8652e08: Fix MarathonSchedulerActor "Cancellation timeout" test
rMARATHON50c5b624725c: Implemented part of the review change requests
rMARATHONba74e901da0b: Rework DeploymentManager logic
JIRA Issues
JIRA MARATHON_EE-1042 Move deployment and logic out of MarathonSchedulerActor -> DeploymentManager
Summary

Rework DeploymentManager logic

  • Deployments are started with the new StartDeployment message
  • Removed MarathonSchedulerActor "awaitCancelation" logic - forced deployments are now completely handled by the DeploymentManager
  • DeploymentManager has it's own deployment repository reference and makes all the repository operations
  • DeploymentManager also handles a failed deployment when it's locked (force = false) through StartDeployment message
  • Fixed unit tests to reflect the changes and added new tests
  • Correct handling of leader abdication/reelection. DeploymentManager now goes into suspended receive mode where it ignores all the messages except for "RecoverDeployments" which signals leader election.
  • Added documentation on deployment manager message flow
  • Removed deployment repository reference from MarathonSchedulerActor
Test Plan

unit-test, integration tests

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.Nov 28 2016, 6:18 PM
src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala
264–265

I left it there because this way the even publishing happens strictly after the lock is removed. Otherwise there is no guarantee for that.

zen-dog updated this revision to Diff 815.Nov 28 2016, 8:44 PM

Implemented review change requests

  • Minor improvement in the MarathonSchedulerActor logic
zen-dog marked 2 inline comments as done.Nov 28 2016, 8:46 PM
zen-dog added inline comments.
src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala
246–247

It is.

263–264

I simplified the calling logic and added a comment to it.

aquamatthias requested changes to this revision.Nov 29 2016, 2:24 PM
aquamatthias added inline comments.
src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala
263–264

Oversimplified. Now this is possible:

  • Deployment A starts --> lock is acquired by A
  • Deployment B starts with force=true --> lock is acquired by B
  • Deployment A fails --> this will remove the lock acquired by B Deployment B now is in progress and no lock for all apps of A and B is available.
This revision now requires changes to proceed.Nov 29 2016, 2:24 PM
zen-dog updated this revision to Diff 883.Nov 30 2016, 12:09 PM
zen-dog marked 2 inline comments as done.

Fixed incorrect lock handling in the MarathonSchedulerActor

zen-dog marked 2 inline comments as done.Nov 30 2016, 12:15 PM
zen-dog added inline comments.
src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala
263–264

Right. Thanks for the catch. I've made lockedRunSpecs a Map[PathId, Int] to save the number of locks being held. This makes it better but the solution is not very robust. Since DeploymentManager has all the information about running deployments I would suggest removing the locks from MarathonSchedulerActor completely and ask DeploymentManager instead. This should be fairly simple to implement but I would wait with it until our ITs are green again.

aquamatthias requested changes to this revision.Dec 1 2016, 11:23 AM

Thanks. Mostly minor comments.

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

According to the code, this comment is misleading:

  • a lock is acquired if deployment is started
  • a lock is acquired if a kill operation is executed
  • a lock is acquired if a scale operation is executed

This basically means:

  • a kill/scale operation should not be performed, while a deployment is in progress
  • a deployment should not be started, if a scale/kill operation is in progress

Asking the deployment manager would not be sufficient.

254

Please add a sentence to force:
If a deployment is forced

  • the new deployment will be started
  • the old deployment will be cancelled and release all claimed locks
  • only in this case, one RunSpec can have 2 locks!
src/main/scala/mesosphere/marathon/upgrade/DeploymentManager.scala
150

The DeploymentManager is started with Marathon even if we are not elected as leader.
That is the reason why this is an Option.
You can not call get safely here.

193–198

Just for clarity: please add: if !isScheduledDeployment(plan.id)

323

You not only mark this deployment, but also start a deployment actor.
Suggestion: startDeployment??

src/test/scala/mesosphere/marathon/MarathonSchedulerActorTest.scala
509–510

yes.

This revision now requires changes to proceed.Dec 1 2016, 11:23 AM
zen-dog updated this revision to Diff 1037.EditedDec 4 2016, 8:13 PM
zen-dog marked 5 inline comments as done.
  • Rebased on master and implemented review change requests
  • Handled one case where DeploymentManager should send MarathonSchedulerActor a notification about canceled deployment
  • Removed SchedulerDriver from the DeploymentManager since it wasn't really needed anywhere
  • Removed "Cancellation timeout" MarathonSchedulerActor test since it wasn't needed anymore
  • Fixed some comments
zen-dog updated this revision to Diff 1048.Dec 5 2016, 12:03 PM

Retest this please

zen-dog updated this revision to Diff 1058.Dec 5 2016, 2:56 PM

Yet another rebase

aquamatthias accepted this revision.Dec 5 2016, 3:32 PM

@zen-dog Thanks. A step in the right direction. This definitively needs further cleanup.

This revision is now accepted and ready to land.Dec 5 2016, 3:32 PM
zen-dog updated this revision to Diff 1062.Dec 5 2016, 3:42 PM

Fixed scapegoat warnings

zen-dog updated this revision to Diff 1074.Dec 5 2016, 7:49 PM
  • Fixed small typo in the DeploymentManagerTest
  • Implemented part of the review change requests
  • Rebased on master and implemented review change requests
meichstedt requested changes to this revision.Dec 6 2016, 2:11 PM
meichstedt added inline comments.
src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala
86

The name is not clear imo. What the message does is basically a GET, although I understand that it should only be used when trying to continue deployments that have been suspended because of a loss of leadership, which should be reflected in the name of this message. What about RetrieveDeployments or RetrieveSuspendedDeployments?

The associated response should be named accordingly, DeploymentsRecovered is not good name either (imo).

88

See above: not a good name. DeploymentsRecovered implies that deployments have been recovered which is not the case here. Rather DeploymentsToRecover or SuspendedDeployments or something like that.

91

Not introduced with this diff, but why do we send from deadLetters?

223

not high prio for this PR, but couldn't we extract all that lock stuff into something that only handles those, thus moving all of these helpers out of this actor?

src/main/scala/mesosphere/marathon/upgrade/DeploymentManager.scala
213

If storing the plan fails, it is still marked as scheduled. It should only be marked as scheduled if storing succeeded, otherwise the actor's state and the repo are out of sync.

I also wonder why you wait for the plan to be stored and then send a message to self. Wouldn't it be clearer to

  1. store new plan and map into that future with self ! LaunchDeploymentActor(plan, recipient)
  2. Do the internal state mutation on receiving that message, after the plan has been persisted:
case LaunchDeploymentActor(plan, recipient) => {
  markScheduled(plan)
  origSender ! DeploymentStarted(plan)
  startDeployment(plan, origSender)

?

233

It's easier to understand that way now, but it's easy to mess up with permutations in the future. The current state is exhaustive wrt StartDeployment. If one of these cases is slightly changed, there are chances the the PF doesn't catch certain messages anymore. How about catching the message in 1 case and calling distinct methods from there? Note all 3 cases will compute conflicting deployments to check whether the case matches, and in 2 of them the conflicting deployments will be computed again in the function body. When matching once you can store the result of this computation.

Also, case matches with more than 2 lines prevent readability imo.

253

Same as above: the internal state is changed before persisting the change. please persist and then change state.

This revision now requires changes to proceed.Dec 6 2016, 2:11 PM
zen-dog added inline comments.Dec 6 2016, 5:04 PM
src/main/scala/mesosphere/marathon/upgrade/DeploymentManager.scala
213

Consider following scenario with two deployments (same pathId) A and B where B is forced and ZK storing is slow:

  • StartDeployment A is coming in
  • A is stored (wait for store to succeed)
  • StartDeployment B is coming in and is forced
  • B is stored too since it's not in the internal state yet and therefore there are no conflicts
  • LaunchDeploymentActor A comes in and proceeds
  • LaunchDeploymentActor B comes in and... proceeds too!

The whole point of saving deployments in the internal state first was to avoid such conflicts.
The case where storing fails is not handled (and wasn't handled in the previous version) since then master fails and the new master picks it up from there with a new state. This feels like a longer discussion so maybe we should sit down together and talk about it.

meichstedt added inline comments.Dec 7 2016, 1:27 PM
src/main/scala/mesosphere/marathon/upgrade/DeploymentManager.scala
213

I don't see why starting DeploymentActor B couldn't mean stopping DeploymentActor A when DeploymentPlan B overrules A. But I see that this probably complicates things on another level, I'm open for discussion.

It might be that failing futures re persisting the plans wasn't handled before, but that's no excuse :P
And no – Marathon does not fail over just because of a failed write. It fails over when the zk connection is lost, not when a read or write operation fails. This should definitely be handled here, no matter in which order the steps are performed.

219

could we stop doing that then? if there is one place in the code where we send with deadletters as sender, we basically need to check for noSender everywhere, as it's really hard to track originalSender's throughout the codebase.

zen-dog updated this revision to Diff 1146.Dec 8 2016, 2:29 AM
  • Implemented basic recovery from failing repository operations
zen-dog updated this revision to Diff 1147.Dec 8 2016, 11:01 AM

Rebased and removed an unused import

zen-dog updated this revision to Diff 1149.Dec 8 2016, 11:34 AM
  • Fixed scapegoat async/await warnings
zen-dog updated this revision to Diff 1150.Dec 8 2016, 12:25 PM

Fixed scaladoc error

meichstedt requested changes to this revision.Dec 8 2016, 4:06 PM

Thanks, just 3 super mini comments :)

src/main/scala/mesosphere/marathon/upgrade/DeploymentManager.scala
249

NonFatal(e)

266

NonFatal(e)

306

NonFatal(e)

This revision now requires changes to proceed.Dec 8 2016, 4:06 PM
zen-dog updated this revision to Diff 1158.Dec 8 2016, 4:38 PM
zen-dog marked 3 inline comments as done.

Implemented feedback: catching NonFatal instead of Throwable

zen-dog added a comment.Dec 8 2016, 4:38 PM
This comment was removed by zen-dog.
meichstedt accepted this revision.Dec 8 2016, 4:42 PM

Thanks!

This revision is now accepted and ready to land.Dec 8 2016, 4:42 PM
Closed by commit rMARATHON1a4d91cddd63: Rework DeploymentManager logic (authored by zen-dog). · Explain WhyDec 8 2016, 4:58 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents

Diff 1160

View Options

src/main/scala/mesosphere/marathon/MarathonModule.scala

Show First 20 Lines • Show All 140 Lines • ▼ Show 20 Line(s)
141141 new DeploymentManager(
142142 instanceTracker,
143143 killService,
144144 launchQueue,
145145 schedulerActions,
146146 storage,
147147 healthCheckManager,
148148 eventBus,
149 readinessCheckExecutor
149 readinessCheckExecutor,
150 deploymentRepository
150151 )
151152 )
152153 }
153154
154155 system.actorOf(
155156 MarathonSchedulerActor.props(
156157 createSchedulerActions,
157158 deploymentManagerProps,
158159 historyActorProps,
159 deploymentRepository,
160160 healthCheckManager,
161161 killService,
162162 launchQueue,
163163 driverHolder,
164164 electionService,
165165 eventBus
166166 )(mat).withRouter(RoundRobinPool(nrOfInstances = 1, supervisorStrategy = supervision)),
167167 "MarathonScheduler")
Show All 28 Lines
View Options

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

11package mesosphere.marathon
22
3import java.util.concurrent.TimeoutException
4
53import akka.actor._
64import akka.event.{ EventStream, LoggingReceive }
7import akka.pattern.ask
85import akka.stream.Materializer
96import mesosphere.marathon.MarathonSchedulerActor.ScaleRunSpec
107import mesosphere.marathon.core.election.{ ElectionService, LocalLeadershipEvent }
118import mesosphere.marathon.core.event.{ AppTerminatedEvent, DeploymentFailed, DeploymentSuccess }
129import mesosphere.marathon.core.health.HealthCheckManager
1310import mesosphere.marathon.core.instance.Instance
1411import mesosphere.marathon.core.instance.Instance.AgentInfo
1512import mesosphere.marathon.core.launchqueue.LaunchQueue
1613import mesosphere.marathon.core.task.Task
1714import mesosphere.marathon.core.task.termination.{ KillReason, KillService }
1815import mesosphere.marathon.core.task.tracker.InstanceTracker
1916import mesosphere.marathon.state.{ PathId, RunSpec }
20import mesosphere.marathon.storage.repository.{ DeploymentRepository, GroupRepository, ReadOnlyAppRepository, ReadOnlyPodRepository }
17import mesosphere.marathon.storage.repository.{ GroupRepository, ReadOnlyAppRepository, ReadOnlyPodRepository }
2118import mesosphere.marathon.stream._
2219import mesosphere.marathon.upgrade.DeploymentManager._
2320import mesosphere.marathon.upgrade.{ DeploymentManager, DeploymentPlan, ScalingProposition }
2421import mesosphere.mesos.Constraints
2522import org.apache.mesos
2623import org.apache.mesos.Protos.{ Status, TaskState }
2724import org.apache.mesos.SchedulerDriver
2825import org.slf4j.LoggerFactory
2926
3027import scala.async.Async.{ async, await }
31import scala.concurrent.duration._
3228import scala.concurrent.{ ExecutionContext, Future }
3329import scala.util.control.NonFatal
3430import scala.util.{ Failure, Success, Try }
3531
3632class LockingFailedException(msg: String) extends Exception(msg)
3733
3834class MarathonSchedulerActor private (
3935 createSchedulerActions: ActorRef => SchedulerActions,
4036 deploymentManagerProps: SchedulerActions => Props,
4137 historyActorProps: Props,
42 deploymentRepository: DeploymentRepository,
4338 healthCheckManager: HealthCheckManager,
4439 killService: KillService,
4540 launchQueue: LaunchQueue,
4641 marathonSchedulerDriverHolder: MarathonSchedulerDriverHolder,
4742 electionService: ElectionService,
48 eventBus: EventStream,
49 cancellationTimeout: FiniteDuration = 1.minute)(implicit val mat: Materializer) extends Actor
43 eventBus: EventStream)(implicit val mat: Materializer) extends Actor
5044 with ActorLogging with Stash {
5145 import context.dispatcher
5246 import mesosphere.marathon.MarathonSchedulerActor._
5347
54 var lockedRunSpecs = Set.empty[PathId]
48 /**
49 * About locks:
50 * - a lock is acquired if deployment is started
51 * - a lock is acquired if a kill operation is executed
52 * - a lock is acquired if a scale operation is executed
aquamatthiasUnsubmitted
Done

According to the code, this comment is misleading:

  • a lock is acquired if deployment is started
  • a lock is acquired if a kill operation is executed
  • a lock is acquired if a scale operation is executed

This basically means:

  • a kill/scale operation should not be performed, while a deployment is in progress
  • a deployment should not be started, if a scale/kill operation is in progress

Asking the deployment manager would not be sufficient.

53 *
54 * This basically means:
55 * - a kill/scale operation should not be performed, while a deployment is in progress
56 * - a deployment should not be started, if a scale/kill operation is in progress
57 * Since multiple conflicting deployment can be handled at the same time lockedRunSpecs saves
58 * the lock count for each affected PathId. Lock is removed if lock count == 0.
59 */
60 // TODO (AD): DeploymentManager has already all the information about running deployments.
61 // MarathonSchedulerActor should only save the locks resulting from scale and kill operations,
62 // asking DeploymentManager for deployment locks.
63 val lockedRunSpecs = collection.mutable.Map[PathId, Int]().withDefaultValue(0)
5564 var schedulerActions: SchedulerActions = _
5665 var deploymentManager: ActorRef = _
5766 var historyActor: ActorRef = _
5867 var activeReconciliation: Option[Future[Status]] = None
5968
6069 override def preStart(): Unit = {
6170 schedulerActions = createSchedulerActions(self)
6271 deploymentManager = context.actorOf(deploymentManagerProps(schedulerActions), "DeploymentManager")
6372 historyActor = context.actorOf(historyActorProps, "HistoryActor")
6473
6574 electionService.subscribe(self)
6675 }
6776
6877 override def postStop(): Unit = {
6978 electionService.unsubscribe(self)
7079 }
7180
7281 def receive: Receive = suspended
7382
7483 def suspended: Receive = LoggingReceive.withLabel("suspended"){
7584 case LocalLeadershipEvent.ElectedAsLeader =>
7685 log.info("Starting scheduler actor")
77 deploymentRepository.all().runWith(Sink.seq).onComplete {
86 deploymentManager ! LoadDeploymentsOnLeaderElection
meichstedtUnsubmitted
Not Done

The name is not clear imo. What the message does is basically a GET, although I understand that it should only be used when trying to continue deployments that have been suspended because of a loss of leadership, which should be reflected in the name of this message. What about RetrieveDeployments or RetrieveSuspendedDeployments?

The associated response should be named accordingly, DeploymentsRecovered is not good name either (imo).

78 case Success(deployments) => self ! RecoverDeployments(deployments)
79 case Failure(_) => self ! RecoverDeployments(Nil)
80 }
8187
82 case RecoverDeployments(deployments) =>
88 case LoadedDeploymentsOnLeaderElection(deployments) =>
meichstedtUnsubmitted
Not Done

See above: not a good name. DeploymentsRecovered implies that deployments have been recovered which is not the case here. Rather DeploymentsToRecover or SuspendedDeployments or something like that.

8389 deployments.foreach { plan =>
8490 log.info(s"Recovering deployment:\n$plan")
8591 deploy(context.system.deadLetters, Deploy(plan, force = false))
meichstedtUnsubmitted
Not Done

Not introduced with this diff, but why do we send from deadLetters?

8692 }
8793
8894 log.info("Scheduler actor ready")
8995 unstashAll()
9096 context.become(started)
9197 self ! ReconcileHealthChecks
9298
9399 case LocalLeadershipEvent.Standby =>
94100 // ignored
95101 // FIXME: When we get this while recovering deployments, we become active anyway
96102 // and drop this message.
97103
98104 case _ => stash()
99105 }
100106
101 def started: Receive = LoggingReceive.withLabel("started")(sharedHandlers orElse {
107 def started: Receive = LoggingReceive.withLabel("started") {
102108 case LocalLeadershipEvent.Standby =>
103109 log.info("Suspending scheduler actor")
104110 healthCheckManager.removeAll()
105 deploymentManager ! StopAllDeployments
106 lockedRunSpecs = Set.empty
111 deploymentManager ! ShutdownDeployments
112 lockedRunSpecs.clear()
107113 context.become(suspended)
108114
109115 case LocalLeadershipEvent.ElectedAsLeader => // ignore
110116
111117 case ReconcileTasks =>
112118 import akka.pattern.pipe
113119 import context.dispatcher
114120 val reconcileFuture = activeReconciliation match {
▲ Show 20 Lines • Show All 55 Lines • ▼ Show 20 Line(s)
170176
171177 res onComplete { _ =>
172178 self ! cmd.answer // unlock app
173179 }
174180
175181 res.sendAnswer(origSender, cmd)
176182 }
177183 withLockFor(runSpecId) { killTasks() }
178 })
179184
180 /**
181 * handlers for messages that unlock run specs and to retrieve running deployments
182 */
183 def sharedHandlers: Receive = {
184185 case DeploymentFinished(plan) =>
185 lockedRunSpecs --= plan.affectedRunSpecIds
186 removeLocks(plan.affectedRunSpecIds)
186187 deploymentSuccess(plan)
187188
188189 case DeploymentManager.DeploymentFailed(plan, reason) =>
189 lockedRunSpecs --= plan.affectedRunSpecIds
190 removeLocks(plan.affectedRunSpecIds)
190191 deploymentFailed(plan, reason)
191192
192 case RunSpecScaled(id) => lockedRunSpecs -= id
193 case RunSpecScaled(id) => removeLock(id)
193194
194 case TasksKilled(runSpecId, _) => lockedRunSpecs -= runSpecId
195 case TasksKilled(runSpecId, _) => removeLock(runSpecId)
195196
196197 case RetrieveRunningDeployments =>
197198 deploymentManager forward RetrieveRunningDeployments
198199 }
199200
200201 /**
201 * Waits for all the rnunSpecs affected by @plan to be unlocked
202 * and starts @plan. If it receives a CancellationTimeoutExceeded
203 * message, it will mark the deployment as failed and go into
204 * the started state.
205 *
206 * @param plan The deployment plan we are trying to execute.
207 * @param origSender The original sender of the Deploy message.
208 * @return
209 */
210 @SuppressWarnings(Array("all")) // async/await
211 def awaitCancellation(plan: DeploymentPlan, origSender: ActorRef, cancellationHandler: Cancellable): Receive =
212 sharedHandlers.andThen[Unit] { _ =>
213 if (tryDeploy(plan, origSender)) {
214 cancellationHandler.cancel()
215 }
216 } orElse {
217 case CancellationTimeoutExceeded =>
218 val reason = new TimeoutException("Exceeded timeout for canceling conflicting deployments.")
219 async { // linter:ignore UnnecessaryElseBranch
220 await(deploymentFailed(plan, reason))
221 origSender ! CommandFailed(Deploy(plan, force = true), reason)
222 unstashAll()
223 context.become(started)
224 }
225 case _ => stash()
226 }
227
228 /**
229 * If all required runSpecs are unlocked, start the deployment,
230 * unstash all messages and put actor in started state
231 *
232 * @param plan The deployment plan that has been sent with force=true
233 * @param origSender The original sender of the deployment
234 */
235 def tryDeploy(plan: DeploymentPlan, origSender: ActorRef): Boolean = {
236 val affectedRunSpecs = plan.affectedRunSpecIds
237 if (!lockedRunSpecs.exists(affectedRunSpecs)) {
238 deploy(origSender, Deploy(plan, force = false))
239 unstashAll()
240 context.become(started)
241 true
242 } else {
243 false
244 }
245 }
246
247 /**
248202 * Tries to acquire the lock for the given runSpecIds.
249203 * If it succeeds it executes the given function,
250204 * otherwise the result will contain an LockingFailedException.
251205 */
252206 def withLockFor[A](runSpecIds: Set[PathId])(f: => A): Try[A] = {
253207 // there's no need for synchronization here, because this is being
254208 // executed inside an actor, i.e. single threaded
255 val conflicts = lockedRunSpecs intersect runSpecIds
256 if (conflicts.isEmpty) {
257 lockedRunSpecs ++= runSpecIds
209 if (noConflictsWith(runSpecIds)) {
210 addLocks(runSpecIds)
258211 Try(f)
259212 } else {
260213 Failure(new LockingFailedException("Failed to acquire locks."))
261214 }
262215 }
263216
217 def noConflictsWith(runSpecIds: Set[PathId]): Boolean = {
218 val conflicts = lockedRunSpecs.keySet intersect runSpecIds
219 conflicts.isEmpty
220 }
221
222 def removeLocks(runSpecIds: Set[PathId]): Unit = runSpecIds.foreach(removeLock)
223 def removeLock(runSpecId: PathId): Unit = {
meichstedtUnsubmitted
Not Done

not high prio for this PR, but couldn't we extract all that lock stuff into something that only handles those, thus moving all of these helpers out of this actor?

224 if (lockedRunSpecs.contains(runSpecId)) {
225 val locks = lockedRunSpecs(runSpecId) - 1
226 if (locks <= 0) lockedRunSpecs -= runSpecId else lockedRunSpecs(runSpecId) -= 1
227 }
228 }
229
230 def addLocks(runSpecIds: Set[PathId]): Unit = runSpecIds.foreach(addLock)
231 def addLock(runSpecId: PathId): Unit = lockedRunSpecs(runSpecId) += 1
232
264233 /**
265234 * Tries to acquire the lock for the given runSpecId.
266235 * If it succeeds it executes the given function,
267236 * otherwise the result will contain an AppLockedException.
268237 */
269238 def withLockFor[A](runSpecId: PathId)(f: => A): Try[A] =
270239 withLockFor(Set(runSpecId))(f)
271240
272241 // there has to be a better way...
273242 @SuppressWarnings(Array("OptionGet"))
274243 def driver: SchedulerDriver = marathonSchedulerDriverHolder.driver.get
275244
276245 def deploy(origSender: ActorRef, cmd: Deploy): Unit = {
277246 val plan = cmd.plan
278 val ids = plan.affectedRunSpecIds
247 val runSpecIds = plan.affectedRunSpecIds
aquamatthiasUnsubmitted
Done

We try and that can fail.
So the force flag is not evaluated in the DeploymentManager?

zen-dogAuthorUnsubmitted
Not Done

It is.

279248
280 val res = withLockFor(ids) {
281 deploy(driver, plan)
249 // If there are no conflicting locks or the deployment is forced we lock passed runSpecIds.
250 // Afterwards the deployment plan is sent to DeploymentManager. It will take care of cancelling
251 // conflicting deployments, scheduling new one or (if there were conflicts but the deployment
252 // is not forced) send to the original sender and AppLockedException with conflicting deployments.
253 //
254 // If a deployment is forced (and there exists an old one):
aquamatthiasUnsubmitted
Done

Please add a sentence to force:
If a deployment is forced

  • the new deployment will be started
  • the old deployment will be cancelled and release all claimed locks
  • only in this case, one RunSpec can have 2 locks!
255 // - the new deployment will be started
256 // - the old deployment will be cancelled and release all claimed locks
257 // - only in this case, one RunSpec can have 2 locks
258 if (noConflictsWith(runSpecIds) || cmd.force) {
259 addLocks(runSpecIds)
282260 }
283
284 res match {
285 case Success(f) =>
286 f.map(_ => if (origSender != Actor.noSender) origSender ! cmd.answer)
287 case Failure(e: LockingFailedException) if cmd.force =>
288 deploymentManager ! CancelConflictingDeployments(plan)
289 val cancellationHandler = context.system.scheduler.scheduleOnce(
290 cancellationTimeout,
291 self,
292 CancellationTimeoutExceeded)
293
294 context.become(awaitCancellation(plan, origSender, cancellationHandler))
295 case Failure(e: LockingFailedException) =>
296 deploymentManager.ask(RetrieveRunningDeployments)(2.seconds)
297 .mapTo[RunningDeployments]
298 .foreach {
299 case RunningDeployments(plans) =>
300 def intersectsWithNewPlan(existingPlan: DeploymentPlan): Boolean = {
301 existingPlan.affectedRunSpecIds.intersect(plan.affectedRunSpecIds).nonEmpty
261 deploymentManager ! StartDeployment(plan, origSender, cmd.force)
302262 }
303 val relatedDeploymentIds: Seq[String] = plans.collect {
304 case DeploymentStepInfo(p, _, _, _) if intersectsWithNewPlan(p) => p.id
305 }
306 origSender ! CommandFailed(cmd, AppLockedException(relatedDeploymentIds))
307 }
308 }
309 }
310263
311 def deploy(driver: SchedulerDriver, plan: DeploymentPlan): Future[Unit] = {
264 def deploymentSuccess(plan: DeploymentPlan): Unit = {
aquamatthiasUnsubmitted
Done

Mmmh - shared responsibilities make it very unclear. If we want to maintain the lock in this actor, perhaps we need the information to StartDeployment(plan, sender, force, lockAquired: Boolean) ??

zen-dogAuthorUnsubmitted
Not Done

I simplified the calling logic and added a comment to it.

aquamatthiasUnsubmitted
Done

Oversimplified. Now this is possible:

  • Deployment A starts --> lock is acquired by A
  • Deployment B starts with force=true --> lock is acquired by B
  • Deployment A fails --> this will remove the lock acquired by B Deployment B now is in progress and no lock for all apps of A and B is available.
zen-dogAuthorUnsubmitted
Not Done

Right. Thanks for the catch. I've made lockedRunSpecs a Map[PathId, Int] to save the number of locks being held. This makes it better but the solution is not very robust. Since DeploymentManager has all the information about running deployments I would suggest removing the locks from MarathonSchedulerActor completely and ask DeploymentManager instead. This should be fairly simple to implement but I would wait with it until our ITs are green again.

312 deploymentRepository.store(plan).map { _ =>
265 log.info(s"Deployment ${plan.id}:${plan.version} of ${plan.target.id} finished")
aquamatthiasUnsubmitted
Not Done

Is this the responsibility of the MarathonSchedulerActor?
Can the DeploymentManager handle this?

zen-dogAuthorUnsubmitted
Not Done

I left it there because this way the even publishing happens strictly after the lock is removed. Otherwise there is no guarantee for that.

313 deploymentManager ! PerformDeployment(driver, plan)
314 }
315 }
316
317 def deploymentSuccess(plan: DeploymentPlan): Future[Unit] = {
318 log.info(s"Deployment ${plan.id}:${plan.version} of ${plan.target.id} successful")
319266 eventBus.publish(DeploymentSuccess(plan.id, plan))
320 deploymentRepository.delete(plan.id).map(_ => ())
321267 }
322268
323 def deploymentFailed(plan: DeploymentPlan, reason: Throwable): Future[Unit] = {
269 def deploymentFailed(plan: DeploymentPlan, reason: Throwable): Unit = {
aquamatthiasUnsubmitted
Not Done

Is this the responsibility of the MarathonSchedulerActor?
Can the DeploymentManager handle this?

zen-dogAuthorUnsubmitted
Not Done

I left it there because this way the purging of the launch queue happens strictly after the lock is removed. Otherwise there is no guarantee for that.

324270 log.error(reason, s"Deployment ${plan.id}:${plan.version} of ${plan.target.id} failed")
325271 plan.affectedRunSpecIds.foreach(runSpecId => launchQueue.purge(runSpecId))
326272 eventBus.publish(DeploymentFailed(plan.id, plan))
327 reason match {
328 case _: DeploymentCanceledException =>
329 deploymentRepository.delete(plan.id).map(_ => ())
330 case _ =>
331 Future.successful(())
332273 }
333274}
334}
335275
336276object MarathonSchedulerActor {
337277 @SuppressWarnings(Array("MaxParameters"))
338278 def props(
339279 createSchedulerActions: ActorRef => SchedulerActions,
340280 deploymentManagerProps: SchedulerActions => Props,
341281 historyActorProps: Props,
342 deploymentRepository: DeploymentRepository,
343282 healthCheckManager: HealthCheckManager,
344283 killService: KillService,
345284 launchQueue: LaunchQueue,
346285 marathonSchedulerDriverHolder: MarathonSchedulerDriverHolder,
347286 electionService: ElectionService,
348287 eventBus: EventStream)(implicit mat: Materializer): Props = {
349288 Props(new MarathonSchedulerActor(
350289 createSchedulerActions,
351290 deploymentManagerProps,
352291 historyActorProps,
353 deploymentRepository,
354292 healthCheckManager,
355293 killService,
356294 launchQueue,
357295 marathonSchedulerDriverHolder,
358296 electionService,
359297 eventBus
360298 ))
361299 }
362300
363 case class RecoverDeployments(deployments: Seq[DeploymentPlan])
301 case class LoadedDeploymentsOnLeaderElection(deployments: Seq[DeploymentPlan])
364302
365303 sealed trait Command {
366304 def answer: Event
367305 }
368306
369307 case object ReconcileTasks extends Command {
370308 def answer: Event = TasksReconciled
371309 }
▲ Show 20 Lines • Show All 233 Lines • Show Last 20 Lines
View Options

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

11package mesosphere.marathon.upgrade
22
33import akka.actor._
44import akka.event.EventStream
55import mesosphere.marathon.core.event.DeploymentStatus
66import mesosphere.marathon.core.launchqueue.LaunchQueue
77import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
88import mesosphere.marathon.core.task.tracker.InstanceTracker
99import mesosphere.marathon.state.RunSpec
1010import mesosphere.marathon.{ AppStartCanceledException, SchedulerActions }
11import org.apache.mesos.SchedulerDriver
1211import org.slf4j.LoggerFactory
1312
1413import scala.concurrent.Promise
1514import scala.util.control.NonFatal
1615
1716class AppStartActor(
1817 val deploymentManager: ActorRef,
1918 val status: DeploymentStatus,
20 val driver: SchedulerDriver,
2119 val scheduler: SchedulerActions,
2220 val launchQueue: LaunchQueue,
2321 val instanceTracker: InstanceTracker,
2422 val eventBus: EventStream,
2523 val readinessCheckExecutor: ReadinessCheckExecutor,
2624 val runSpec: RunSpec,
2725 val scaleTo: Int,
2826 promise: Promise[Unit]) extends Actor with StartingBehavior {
Show All 27 Lines
5654 }
5755}
5856
5957object AppStartActor {
6058 @SuppressWarnings(Array("MaxParameters"))
6159 def props(
6260 deploymentManager: ActorRef,
6361 status: DeploymentStatus,
64 driver: SchedulerDriver,
6562 scheduler: SchedulerActions,
6663 launchQueue: LaunchQueue,
6764 taskTracker: InstanceTracker,
6865 eventBus: EventStream,
6966 readinessCheckExecutor: ReadinessCheckExecutor,
7067 runSpec: RunSpec,
7168 scaleTo: Int,
7269 promise: Promise[Unit]): Props = {
73 Props(new AppStartActor(deploymentManager, status, driver, scheduler, launchQueue, taskTracker, eventBus,
70 Props(new AppStartActor(deploymentManager, status, scheduler, launchQueue, taskTracker, eventBus,
7471 readinessCheckExecutor, runSpec, scaleTo, promise))
7572 }
7673}
View Options

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

11package mesosphere.marathon
22package upgrade
33
44import java.net.URL
55
66import akka.actor._
77import akka.event.EventStream
8import mesosphere.marathon.SchedulerActions
98import mesosphere.marathon.core.event.{ DeploymentStatus, DeploymentStepFailure, DeploymentStepSuccess }
109import mesosphere.marathon.core.health.HealthCheckManager
1110import mesosphere.marathon.core.instance.Instance
1211import mesosphere.marathon.core.launchqueue.LaunchQueue
1312import mesosphere.marathon.core.pod.PodDefinition
1413import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1514import mesosphere.marathon.core.task.termination.{ KillReason, KillService }
1615import mesosphere.marathon.core.task.tracker.InstanceTracker
1716import mesosphere.marathon.io.storage.StorageProvider
1817import mesosphere.marathon.state.{ AppDefinition, RunSpec }
1918import mesosphere.marathon.upgrade.DeploymentManager.{ DeploymentFailed, DeploymentFinished, DeploymentStepInfo }
2019import mesosphere.mesos.Constraints
21import org.apache.mesos.SchedulerDriver
2220import com.typesafe.scalalogging.StrictLogging
2321
2422import scala.concurrent.duration._
25
2623import scala.concurrent.{ Future, Promise }
2724import scala.util.{ Failure, Success }
2825
2926private class DeploymentActor(
3027 deploymentManager: ActorRef,
3128 receiver: ActorRef,
32 driver: SchedulerDriver,
3329 killService: KillService,
3430 scheduler: SchedulerActions,
3531 plan: DeploymentPlan,
3632 instanceTracker: InstanceTracker,
3733 launchQueue: LaunchQueue,
3834 storage: StorageProvider,
3935 healthCheckManager: HealthCheckManager,
4036 eventBus: EventStream,
▲ Show 20 Lines • Show All 89 Lines • ▼ Show 20 Line(s)
130126 }
131127 }
132128 }
133129
134130 // scalastyle:on
135131
136132 def startRunnable(runnableSpec: RunSpec, scaleTo: Int, status: DeploymentStatus): Future[Unit] = {
137133 val promise = Promise[Unit]()
138 context.actorOf(AppStartActor.props(deploymentManager, status, driver, scheduler, launchQueue, instanceTracker,
134 context.actorOf(AppStartActor.props(deploymentManager, status, scheduler, launchQueue, instanceTracker,
139135 eventBus, readinessCheckExecutor, runnableSpec, scaleTo, promise))
140136 promise.future
141137 }
142138
143139 def scaleRunnable(runnableSpec: RunSpec, scaleTo: Int,
144140 toKill: Option[Seq[Instance]],
145141 status: DeploymentStatus): Future[Unit] = {
146142 logger.debug("Scale runnable {}", runnableSpec)
Show All 14 Lines
161157 killService.killInstances(tasks, KillReason.DeploymentScaling).map(_ => ())
162158 }
163159 }
164160
165161 def startTasksIfNeeded: Future[Unit] = {
166162 tasksToStart.fold(Future.successful(())) { tasksToStart =>
167163 logger.debug(s"Start next $tasksToStart tasks")
168164 val promise = Promise[Unit]()
169 context.actorOf(TaskStartActor.props(deploymentManager, status, driver, scheduler, launchQueue, instanceTracker, eventBus,
165 context.actorOf(TaskStartActor.props(deploymentManager, status, scheduler, launchQueue, instanceTracker, eventBus,
170166 readinessCheckExecutor, runnableSpec, scaleTo, promise))
171167 promise.future
172168 }
173169 }
174170
175171 killTasksIfNeeded.flatMap(_ => startTasksIfNeeded)
176172 }
177173
178174 def stopRunnable(runnableSpec: RunSpec): Future[Unit] = {
179175 val tasks = instanceTracker.specInstancesLaunchedSync(runnableSpec.id)
180176 // TODO: the launch queue is purged in stopRunnable, but it would make sense to do that before calling kill(tasks)
181177 killService.killInstances(tasks, KillReason.DeletingApp).map(_ => ()).andThen {
182178 case Success(_) => scheduler.stopRunSpec(runnableSpec)
183179 }
184180 }
185181
186182 def restartRunnable(run: RunSpec, status: DeploymentStatus): Future[Unit] = {
187183 if (run.instances == 0) {
188184 Future.successful(())
189185 } else {
190186 val promise = Promise[Unit]()
191 context.actorOf(TaskReplaceActor.props(deploymentManager, status, driver, killService,
187 context.actorOf(TaskReplaceActor.props(deploymentManager, status, killService,
192188 launchQueue, instanceTracker, eventBus, readinessCheckExecutor, run, promise))
193189 promise.future
194190 }
195191 }
196192
197193 def resolveArtifacts(urls: Map[URL, String]): Future[Unit] = {
198194 val promise = Promise[Boolean]()
199195 context.actorOf(ResolveArtifactsActor.props(urls, promise, storage))
Show All 16 Lines
216212 case object Shutdown
217213
218214 val GracefulDeploymentShutdownTimeout = 30000.milliseconds
219215
220216 @SuppressWarnings(Array("MaxParameters"))
221217 def props(
222218 deploymentManager: ActorRef,
223219 receiver: ActorRef,
224 driver: SchedulerDriver,
225220 killService: KillService,
226221 scheduler: SchedulerActions,
227222 plan: DeploymentPlan,
228223 taskTracker: InstanceTracker,
229224 launchQueue: LaunchQueue,
230225 storage: StorageProvider,
231226 healthCheckManager: HealthCheckManager,
232227 eventBus: EventStream,
233228 readinessCheckExecutor: ReadinessCheckExecutor): Props = {
234229
235230 Props(new DeploymentActor(
236231 deploymentManager,
237232 receiver,
238 driver,
239233 killService,
240234 scheduler,
241235 plan,
242236 taskTracker,
243237 launchQueue,
244238 storage,
245239 healthCheckManager,
246240 eventBus,
247241 readinessCheckExecutor
248242 ))
249243 }
250244}
View Options

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

11package mesosphere.marathon
22package upgrade
33
44import akka.actor.SupervisorStrategy.Stop
55import akka.actor._
66import akka.event.EventStream
7import mesosphere.marathon.MarathonSchedulerActor.{ RetrieveRunningDeployments, RunningDeployments }
7import akka.stream.Materializer
8import mesosphere.marathon.MarathonSchedulerActor.{ CommandFailed, DeploymentStarted, LoadedDeploymentsOnLeaderElection, RetrieveRunningDeployments, RunningDeployments }
89import mesosphere.marathon.core.health.HealthCheckManager
910import mesosphere.marathon.core.launchqueue.LaunchQueue
1011import mesosphere.marathon.core.readiness.{ ReadinessCheckExecutor, ReadinessCheckResult }
1112import mesosphere.marathon.core.task.Task
1213import mesosphere.marathon.core.task.termination.KillService
1314import mesosphere.marathon.core.task.tracker.InstanceTracker
1415import mesosphere.marathon.io.storage.StorageProvider
15import mesosphere.marathon.state.{ RootGroup, PathId, Timestamp }
16import mesosphere.marathon.upgrade.DeploymentActor.Cancel
17import mesosphere.marathon.{ ConcurrentTaskUpgradeException, DeploymentCanceledException, SchedulerActions }
18import org.apache.mesos.SchedulerDriver
16import mesosphere.marathon.state.PathId
17import mesosphere.marathon.storage.repository.DeploymentRepository
18import mesosphere.marathon.stream.Sink
19import org.slf4j.LoggerFactory
1920
21import scala.async.Async.{ async, await }
2022import scala.collection.immutable.Seq
2123import scala.collection.mutable
2224import scala.concurrent.{ Future, Promise }
2325import scala.util.control.NonFatal
26import scala.util.{ Failure, Success }
2427
28// format: OFF
29 /*
30 * Basic deployment message flow:
31 * ===========================================
32 * 1. Every deployment starts with a StartDeployment message. In the simplest case when there are no conflicts
33 * deployment is added to the list of running deployments and saved in the repository.
34 * 2. On completion of the store operation (which returns a Future) we proceed with the deployment by sending
35 * ourselves a LaunchDeploymentActor message.
36 * 3. Upon receiving a LaunchDeploymentActor we check if the deployment is still scheduled. It can happen that
37 * while we're were waiting for the deployment to be stored another (forced) deployment canceled (and deleted
38 * from the repository) this one. In this case the deployment is discarded.
39 * 4. When the deployment is finished, DeploymentActor sends a FinishedDeployment message which will remove it
40 * from the list of running deployment and delete from repository.
41 *
42 * Basic flow visualised:
43 *
44 * -> StartDeployment (1) | - save deployment and mark it as [Scheduling]
45 * | -> repository.store(plan).onComplete (2)
46 * | <- LaunchDeploymentActor
47 * |
48 * -> LaunchDeploymentActor | - mark deployment as [Deploying]
49 * | - spawn DeploymentActor (3)
50 * | <- FinishedDeployment
51 * |
52 * -> FinishedDeployment (4) | - remove from runningDeployments
53 * | - delete from repository
54 * ˅
55 *
56 * Deployment message flow with conflicts:
aquamatthiasUnsubmitted
Not Done

Thanks for great documentation!

57 * ===========================================
58 * This is a more complicated flow since there are multiple steps that are performed asynchronously and each individually
59 * can fail. Imagine a situation where we have already deployments A, B and C in process and a new deployment D arrives
60 * which has conflicts with all 3 of existing ones. In this case we have to:
61 *
62 * - repository.delete(A, B and C)
63 * - cancel(A, B and C)
64 * - repository.store(D)
65 * - deploy(D)
66 *
67 * Repository store() and delete() operations may fail individually hence we have here:
68 * 3 deletes + 1 store = 4 operations that may fail. This is unlikely but still possible.
69 *
70 * The matter is even more complicated by the fact that during any of those operations a new and forced deployment E
71 * my arrive for which all of existing ones are conflicts. Additionally current leader may abdicate at any moment and
72 * the new leader has to find a consistent state in the repository from which it should be able to reconcile.
73 *
74 * Due to the fact that deployments are stored/deleted individually and that we're only saving the desired (future)
75 * state makes a full and complete recovery very hard if possible. Thus deployment manager tries to leave the repository
76 * consistent (no conflicting deployments are stored) so even in a case of failure it should be possible for:
77 * - the new master to reconcile deployments and end in a valid (but might be incomplete) state
78 * - framework user to resend the deployment (force = true) to achieve desired stat
79 *
80 * Current implementation:
81 * ===========================================
82 * 1. Find all conflicting deployments and mark new plan as [Scheduled]
83 * 2. Delete all conflicts from the repository first. This way even if the master crashes we shouldn't have any
84 * conflicts stored. However in the worst case (a crash after delete() and before store() we could end up with old
85 * deployments canceled and new one not existing. This is where framework user has to resend the deployment. On
86 * competition of the delete send self a CancelDeletedConflicts message.
87 * In case of a failure the next forced deployment will go through conflicts and try to delete them again.
88 * 3. Cancel all conflicting deployment by spawning a StopActor. The cancellation futures are saved and the deployments
89 * are marked as [Cancelling]. Store the new plan in the repository. On completion of the store operation send self
90 * a WaitForCanceledConflicts.
91 * Note: Only after the deployment is stored we can send the original sender a positive response.
92 * 4. If the deployment is still scheduled we wait for all the cancellation futures of the conflicts to complete.
93 * When that's the case a LaunchDeploymentActor message is sent to ourselves and the rest of the deployment is
94 * equal to the basic flow.
95 * If any of the steps fail we send recover by sending self a FailedRepositoryOperation message. If the deployment is
96 * still scheduled and not being canceled by the next one than we remove it from internal state. Otherwise the next
97 * deployment will take care of it.
98 *
99 * Handling conflicts visualised:
100 * ===========================================
101 *
102 * -> StartDeployment (1, 2) | - mark deployment as [Scheduled]
103 * | - repository.delete(conflicts)
104 * | <- CancelDeletedConflicts
105 * |
106 * -> CancelDeletedConflicts (3) | - cancel conflicting deployments with StopActor
107 * | - mark them as [Canceled] and save the cancellation future
108 * | - repository.store(plan)
109 * | <- WaitForCanceledConflicts
110 * |
111 * -> WaitForCanceledConflicts (4) | -> Future.sequence(cancellations).onComplete (2.5)
112 * | <- LaunchDeploymentActor
113 * ... ˅
114 *
115 * Note: deployment repository serializes operations by the key (deployment id). This allows us to keep store/delete
116 * operations of the dependant deployments in order.
117 *
118 * Future work:
119 * ===========================================
120 * - Master should be able to reconcile completely from repository state because both current and desired
121 * state are stored
122 * - Multiple store/delete operations might be replaced by storing one deployment graph similar to how
123 * we handle groups
124 * - Move conflict resolution/dependencies/other upgrade logic out into a separate layer
125 */
126// format: ON
25127class DeploymentManager(
26128 taskTracker: InstanceTracker,
27129 killService: KillService,
28130 launchQueue: LaunchQueue,
29131 scheduler: SchedulerActions,
30132 storage: StorageProvider,
31133 healthCheckManager: HealthCheckManager,
32134 eventBus: EventStream,
33 readinessCheckExecutor: ReadinessCheckExecutor) extends Actor with ActorLogging {
135 readinessCheckExecutor: ReadinessCheckExecutor,
136 deploymentRepository: DeploymentRepository,
137 deploymentActorProps: (ActorRef, ActorRef, KillService, SchedulerActions, DeploymentPlan, InstanceTracker, LaunchQueue, StorageProvider, HealthCheckManager, EventStream, ReadinessCheckExecutor) => Props = DeploymentActor.props)(implicit val mat: Materializer) extends Actor {
34138 import context.dispatcher
35139 import mesosphere.marathon.upgrade.DeploymentManager._
36140
37 val runningDeployments: mutable.Map[String, DeploymentInfo] = mutable.Map.empty[String, DeploymentInfo]
38 val deploymentStatus: mutable.Map[String, DeploymentStepInfo] = mutable.Map.empty[String, DeploymentStepInfo]
141 private[this] val log = LoggerFactory.getLogger(getClass)
39142
143 val runningDeployments: mutable.Map[String, DeploymentInfo] = mutable.Map.empty
144 val deploymentStatus: mutable.Map[String, DeploymentStepInfo] = mutable.Map.empty
145
40146 override def supervisorStrategy: SupervisorStrategy = OneForOneStrategy() {
41147 case NonFatal(e) => Stop
42148 }
43149
44 def receive: Receive = {
150 def receive: Receive = suspended
aquamatthiasUnsubmitted
Done

This operation will fail if we are not leader. This needs to be handled.

zen-dogAuthorUnsubmitted
Not Done

I'm not getting it if we're not a leader (waiting for marathon scheduler actor to wake us on leader election, ignoring everything else)

aquamatthiasUnsubmitted
Not Done

The DeploymentManager is started with Marathon even if we are not elected as leader.
That is the reason why this is an Option.
You can not call get safely here.

45 case CancelConflictingDeployments(plan) =>
46 val conflictingDeployments = for {
47 info <- runningDeployments.values
48 if info.plan.isAffectedBy(plan)
49 } yield info
50151
51 val cancellations = conflictingDeployments.map { info =>
52 stopActor(info.ref, new DeploymentCanceledException("The upgrade has been cancelled"))
152 // TODO (AD): suspend-on-leader-abdication behavior should be implemented with the help of
153 // TODO (AD): LeadershipModule.startWhenLeader() when moved this class to a core component.
aquamatthiasUnsubmitted
Done

Marathon has a Leadership Actor, which will start actors onElected and stop onDefeated.
This is probably the right choice here.

zen-dogAuthorUnsubmitted
Not Done

I would suggest to do this after DeploymentManager is moved to a core component. Its' much easier to implement then. I'll leave a TODO.

154 def suspended: Receive = {
155 // Should only be sent by MarathonSchedulerActor on leader election. Reads all deployments from the
156 // repository and sends them to sender. MarathonSchedulerActor needs to create locks for all running
157 // deployments before they can be started.
158 case LoadDeploymentsOnLeaderElection =>
159 log.info("Recovering all deployments on leader election")
160 val recipient = sender()
161 deploymentRepository.all().runWith(Sink.seq).onComplete {
162 case Success(deployments) => recipient ! LoadedDeploymentsOnLeaderElection(deployments)
163 case Failure(t) =>
164 log.error(s"Failed to recover deployments from repository: $t")
165 recipient ! LoadedDeploymentsOnLeaderElection(Nil)
53166 }
54
55 Future.sequence(cancellations).onComplete { _ =>
56 log.info(s"Conflicting deployments for deployment ${plan.id} have been canceled")
57 scheduler.schedulerActor ! ConflictingDeploymentsCanceled(
58 plan.id,
59 if (conflictingDeployments.nonEmpty) {
60 conflictingDeployments.map(_.plan).to[Seq]
61 } else Seq(plan))
167 context.become(started)
168 case _ =>
169 // All messages are ignored until master reelection
62170 }
63171
64 case StopAllDeployments =>
65 for ((_, DeploymentInfo(ref, _)) <- runningDeployments)
172 def started: Receive = {
173 // Should only be sent by MarathonSchedulerActor on leader abdication. All the deployments are stopped
174 // and the context becomes suspended. It's important not to receive DeploymentFinished messages from
175 // running DeploymentActors because that will delete stored deployments from the repository.
176 case ShutdownDeployments =>
177 log.info("Shutting down all deployments on leader abdication")
178 for ((_, DeploymentInfo(Some(ref), _, _, _)) <- runningDeployments)
66179 ref ! DeploymentActor.Shutdown
67180 runningDeployments.clear()
68181 deploymentStatus.clear()
182 context.become(suspended)
69183
70184 case CancelDeployment(id) =>
71 val origSender = sender()
185 cancelDeployment(sender(), id)
aquamatthiasUnsubmitted
Done

I'm a fan of calling a method in the receive function, if it is more than 2 lines. So it easy to distinguish behaviour and communication message. But this is a taste thing..

aquamatthiasUnsubmitted
Done

If there is no running deployment, a DeploymentCanceledException is thrown? Sounds wrong.

zen-dogAuthorUnsubmitted
Not Done

Same logic as it was before - didn't change it. And since you can see "canceling" as an "upgrade operation" it can also fail... I guess there is some logic somewhere there.

72186
187 case DeploymentFinished(plan) =>
188 log.info(s"Removing ${plan.id} from list of running deployments")
aquamatthiasUnsubmitted
Done

The extracted ref is not used.

189 runningDeployments -= plan.id
190 deploymentStatus -= plan.id
191 deploymentRepository.delete(plan.id)
192
193 case LaunchDeploymentActor(plan, origSender) if isScheduledDeployment(plan.id) =>
194 log.info(s"Launching DeploymentActor for ${plan.id}")
195 startDeployment(plan, origSender)
196
197 case LaunchDeploymentActor(plan, _) =>
198 log.info(s"Deployment ${plan.id} was already canceled or overridden by another one. Not proceeding with it")
aquamatthiasUnsubmitted
Done

When can this happen? Should we log this on warn?

zen-dogAuthorUnsubmitted
Not Done

This can happen when we get 2 forced deployments in a row, second one effectively canceling the first one out. I don't think this is worth a warning - at the end this is user triggered.

aquamatthiasUnsubmitted
Not Done

Just for clarity: please add: if !isScheduledDeployment(plan.id)

199
200 case stepInfo: DeploymentStepInfo => deploymentStatus += stepInfo.plan.id -> stepInfo
201
202 case ReadinessCheckUpdate(id, result) => deploymentStatus.get(id).foreach { info =>
203 deploymentStatus += id -> info.copy(readinessChecks = info.readinessChecks.updated(result.taskId, result))
204 }
205
206 case RetrieveRunningDeployments =>
207 sender() ! RunningDeployments(deploymentStatus.values.to[Seq])
208
209 case StartDeployment(plan, origSender, force) =>
210 val conflicts = conflictingDeployments(plan)
211 val hasConflicts = conflicts.nonEmpty
212 val recipient = sender()
213
meichstedtUnsubmitted
Not Done

If storing the plan fails, it is still marked as scheduled. It should only be marked as scheduled if storing succeeded, otherwise the actor's state and the repo are out of sync.

I also wonder why you wait for the plan to be stored and then send a message to self. Wouldn't it be clearer to

  1. store new plan and map into that future with self ! LaunchDeploymentActor(plan, recipient)
  2. Do the internal state mutation on receiving that message, after the plan has been persisted:
case LaunchDeploymentActor(plan, recipient) => {
  markScheduled(plan)
  origSender ! DeploymentStarted(plan)
  startDeployment(plan, origSender)

?

zen-dogAuthorUnsubmitted
Not Done

Consider following scenario with two deployments (same pathId) A and B where B is forced and ZK storing is slow:

  • StartDeployment A is coming in
  • A is stored (wait for store to succeed)
  • StartDeployment B is coming in and is forced
  • B is stored too since it's not in the internal state yet and therefore there are no conflicts
  • LaunchDeploymentActor A comes in and proceeds
  • LaunchDeploymentActor B comes in and... proceeds too!

The whole point of saving deployments in the internal state first was to avoid such conflicts.
The case where storing fails is not handled (and wasn't handled in the previous version) since then master fails and the new master picks it up from there with a new state. This feels like a longer discussion so maybe we should sit down together and talk about it.

meichstedtUnsubmitted
Not Done

I don't see why starting DeploymentActor B couldn't mean stopping DeploymentActor A when DeploymentPlan B overrules A. But I see that this probably complicates things on another level, I'm open for discussion.

It might be that failing futures re persisting the plans wasn't handled before, but that's no excuse :P
And no – Marathon does not fail over just because of a failed write. It fails over when the zk connection is lost, not when a read or write operation fails. This should definitely be handled here, no matter in which order the steps are performed.

214 if (!hasConflicts) startNonConflictingDeployment(plan, origSender, recipient)
215 else if (hasConflicts && !force) giveUpConflictingDeployment(plan, origSender, force)
216 else if (hasConflicts && force) startConflictingDeployment(plan, conflicts, origSender, recipient)
217
218 case CancelDeletedConflicts(plan, conflicts, recipient, origSender) if isScheduledDeployment(plan.id) =>
219 cancelDeletedConflicts(plan, conflicts, recipient, origSender)
aquamatthiasUnsubmitted
Done

Why do we need to check for noSender?

zen-dogAuthorUnsubmitted
Not Done

Because reconciled deployments are sent with context.system.deadLetters as sender.

meichstedtUnsubmitted
Not Done

could we stop doing that then? if there is one place in the code where we send with deadletters as sender, we basically need to check for noSender everywhere, as it's really hard to track originalSender's throughout the codebase.

220
221 case WaitForCanceledConflicts(plan, conflicts, origSender) if isScheduledDeployment(plan.id) =>
222 waitForCanceledConflicts(plan, conflicts, origSender)
223
224 case FailedRepositoryOperation(plan, recipient, reason) if isScheduledDeployment(plan.id) =>
225 recipient ! DeploymentFailed(plan, reason)
226 runningDeployments.remove(plan.id)
227 }
228
229 private def giveUpConflictingDeployment(plan: DeploymentPlan, origSender: ActorRef, force: Boolean): Unit = {
230 log.info(s"Received new deployment plan ${plan.id}. Conflicts are detected and it is not forced, so it will not start")
231 origSender ! CommandFailed(
232 MarathonSchedulerActor.Deploy(plan, force),
233 AppLockedException(conflictingDeployments(plan).map(_.plan.id)))
aquamatthiasUnsubmitted
Done

Not necessary but easier to understand: please add if hasConflicts(plan) && force

meichstedtUnsubmitted
Not Done

It's easier to understand that way now, but it's easy to mess up with permutations in the future. The current state is exhaustive wrt StartDeployment. If one of these cases is slightly changed, there are chances the the PF doesn't catch certain messages anymore. How about catching the message in 1 case and calling distinct methods from there? Note all 3 cases will compute conflicting deployments to check whether the case matches, and in 2 of them the conflicting deployments will be computed again in the function body. When matching once you can store the result of this computation.

Also, case matches with more than 2 lines prevent readability imo.

234 }
235
236 @SuppressWarnings(Array("all")) // async/await
237 private def startNonConflictingDeployment(plan: DeploymentPlan, origSender: ActorRef, recipient: ActorRef): Unit = {
238 log.info(s"Received new deployment plan ${plan.id}, no conflicts detected")
239 markScheduled(plan)
240
241 async {
242 await(deploymentRepository.store(plan))
243 log.info(s"Stored new deployment plan ${plan.id}")
244
245 if (origSender != Actor.noSender) origSender ! DeploymentStarted(plan)
246
247 self ! LaunchDeploymentActor(plan, recipient)
248 }.recover {
249 case NonFatal(e) =>
meichstedtUnsubmitted
Done

NonFatal(e)

250 log.error(s"Couldn't start deployment ${plan.id}. Repository store failed with: $e")
aquamatthiasUnsubmitted
Done

I misinterpreted Save as store. But this is only in memory. Perhaps make the comment + method name more clear.

251 self ! FailedRepositoryOperation(plan, recipient, e)
252 }
253 }
meichstedtUnsubmitted
Not Done

Same as above: the internal state is changed before persisting the change. please persist and then change state.

254
255 @SuppressWarnings(Array("all")) // async/await
256 private def startConflictingDeployment(plan: DeploymentPlan, conflicts: Seq[DeploymentInfo], origSender: ActorRef, recipient: ActorRef): Unit = {
257 log.info(s"Received new forced deployment plan ${plan.id} Proceeding with canceling conflicts ${conflicts.map(_.plan.id)}")
258
259 markScheduled(plan)
260
261 async {
262 await(Future.sequence(conflicts.map(p => deploymentRepository.delete(p.plan.id))))
263 log.info(s"Removed conflicting deployments ${conflicts.map(_.plan.id)} from the repository")
264 self ! CancelDeletedConflicts(plan, conflicts, recipient, origSender)
265 }.recover {
266 case NonFatal(e) =>
meichstedtUnsubmitted
Done

NonFatal(e)

267 log.info(s"Failed to start deployment ${plan.id}. Repository delete failed with: $e")
268 self ! FailedRepositoryOperation(plan, recipient, e)
269 }
270 }
271
272 @SuppressWarnings(Array("all")) // async/await
273 private def waitForCanceledConflicts(plan: DeploymentPlan, conflicts: Seq[DeploymentInfo], origSender: ActorRef) = {
274 val toCancel = conflicts.filter(_.status == DeploymentStatus.Canceling)
275 val cancellations: Seq[Future[Boolean]] = toCancel.flatMap(_.cancel)
276
277 async {
278 await(Future.sequence(cancellations))
279
280 log.info(s"Conflicting deployments ${toCancel.map(_.plan.id)} for deployment ${plan.id} have been canceled")
281 self ! LaunchDeploymentActor(plan, origSender)
282 }
283 }
284
285 @SuppressWarnings(Array("all")) // async/await
aquamatthiasUnsubmitted
Done

Why do we need filter?

zen-dogAuthorUnsubmitted
Not Done

It's an optimization really. So we don't "wait" for already successfully cancellations that are still in the list. But since those will succeed immediately it's not necessary.

286 private def cancelDeletedConflicts(plan: DeploymentPlan, conflicts: Seq[DeploymentInfo], recipient: ActorRef, origSender: ActorRef): Unit = {
287 // [Scheduled] - remove from internal state (they haven't been started yet, so there is nothing to cancel),
288 // and tell MarathonSchedulerActor that it was canceled since it needs to remove the lock.
289 // [Deploying] - cancel by spawning a StopActor and marking as [Canceling]
290 // [Canceling] - Nothing to do here since this deployment is already being canceled
291 conflicts.foreach{
292 case DeploymentInfo(_, p, DeploymentStatus.Scheduled, _) => runningDeployments.remove(p.id).map(info =>
293 recipient ! DeploymentFailed(info.plan, new DeploymentCanceledException("The upgrade has been cancelled")))
aquamatthiasUnsubmitted
Done

Sounds like it would make sense...
The user starts a deployment - the old deployment needs to be canceled, before the new one can be started.
Only if the deployment is started, we can signal to the user a result.

zen-dogAuthorUnsubmitted
Not Done

This was the case with the old MarathonSchedulerActor. The new DeploymentManager cancels conflicting deployment, saves the new one, marking it as Scheduled and is ready to handle the next one. This timeout was needed in the old version, because MarathonSchedulerActor was waiting for the deployment to be canceled while being effectively blocked, stashing all other messages. This is not the case anymore.
I would argue we don't it anymore but wanted to hear another opinion - hence the TODO.

294 case DeploymentInfo(_, p, DeploymentStatus.Deploying, _) => stopDeployment(p.id)
295 case DeploymentInfo(_, _, DeploymentStatus.Canceling, _) => // Nothing to do here - this deployment is already being canceled
296 }
297
298 async {
299 await(deploymentRepository.store(plan))
300 log.info(s"Stored new deployment plan ${plan.id}")
301
302 if (origSender != Actor.noSender) origSender ! DeploymentStarted(plan)
303
304 self ! WaitForCanceledConflicts(plan, conflicts, recipient)
305 }.recover{
306 case NonFatal(e) =>
meichstedtUnsubmitted
Done

NonFatal(e)

307 log.error(s"Couldn't start deployment ${plan.id}. Repository store failed with: $e")
308 self ! FailedRepositoryOperation(plan, recipient, e)
309 }
310 }
311
312 private def cancelDeployment(sender: ActorRef, id: String) = {
73313 runningDeployments.get(id) match {
74 case Some(info) =>
75 info.ref ! Cancel(new DeploymentCanceledException("The upgrade has been cancelled"))
314 case Some(DeploymentInfo(_, _, DeploymentStatus.Scheduled, _)) =>
315 log.info(s"Canceling scheduled deployment $id.")
316 runningDeployments.remove(id).map(info => sender ! DeploymentFailed(info.plan, new DeploymentCanceledException("The upgrade has been cancelled")))
317
318 case Some(DeploymentInfo(Some(_), _, DeploymentStatus.Deploying, _)) =>
319 log.info(s"Canceling deployment $id which is already in progress.")
320 stopDeployment(id)
321
322 case Some(DeploymentInfo(_, _, DeploymentStatus.Canceling, _)) =>
323 log.warn(s"The deployment $id is already being canceled.")
aquamatthiasUnsubmitted
Done

You not only mark this deployment, but also start a deployment actor.
Suggestion: startDeployment??

324
325 case Some(_) =>
326 // This means we have a deployment with a [Deploying] status which has no DeploymentActor to cancel it.
327 // This is clearly an invalid state and should never happen.
328 log.error(s"Failed to cancel an invalid deployment ${runningDeployments.get(id)}")
329
76330 case None =>
77 origSender ! DeploymentFailed(
78 DeploymentPlan(id, RootGroup.empty, RootGroup.empty, Nil, Timestamp.now()),
331 sender ! DeploymentFailed(
332 DeploymentPlan.empty.copy(id = id),
79333 new DeploymentCanceledException("The upgrade has been cancelled"))
80334 }
335 }
81336
82 case msg @ DeploymentFinished(plan) =>
83 log.info(s"Removing ${plan.id} from list of running deployments")
84 runningDeployments -= plan.id
85 deploymentStatus -= plan.id
337 /** Method saves new DeploymentInfo with status = [Scheduled] */
338 private def markScheduled(plan: DeploymentPlan) = {
339 runningDeployments += plan.id -> DeploymentInfo(plan = plan, status = DeploymentStatus.Scheduled)
340 }
86341
87 case PerformDeployment(driver, plan) if !runningDeployments.contains(plan.id) =>
342 /** Method spawns a DeploymentActor for the passed plan and saves new DeploymentInfo with status = [Scheduled] */
343 private def startDeployment(plan: DeploymentPlan, origSender: ActorRef) = {
88344 val ref = context.actorOf(
89 DeploymentActor.props(
345 deploymentActorProps(
90346 self,
91 sender(),
92 driver,
347 origSender,
93348 killService,
94349 scheduler,
95350 plan,
96351 taskTracker,
97352 launchQueue,
98353 storage,
99354 healthCheckManager,
100355 eventBus,
101356 readinessCheckExecutor
102357 ),
103358 plan.id
104359 )
105 runningDeployments += plan.id -> DeploymentInfo(ref, plan)
106
107 case stepInfo: DeploymentStepInfo => deploymentStatus += stepInfo.plan.id -> stepInfo
108
109 case ReadinessCheckUpdate(id, result) => deploymentStatus.get(id).foreach { info =>
110 deploymentStatus += id -> info.copy(readinessChecks = info.readinessChecks.updated(result.taskId, result))
360 runningDeployments.update(plan.id, runningDeployments(plan.id).copy(ref = Some(ref), status = DeploymentStatus.Deploying))
111361 }
112362
113 case _: PerformDeployment =>
114 sender() ! Status.Failure(new ConcurrentTaskUpgradeException("Deployment is already in progress"))
115
116 case RetrieveRunningDeployments =>
117 sender() ! RunningDeployments(deploymentStatus.values.to[Seq])
363 /** Method spawns a StopActor for the passed plan Id and saves new DeploymentInfo with status = [Canceling] */
364 @SuppressWarnings(Array("OptionGet"))
365 private def stopDeployment(id: String) = {
366 val info = runningDeployments(id)
367 val stopFuture = stopActor(info.ref.get, new DeploymentCanceledException("The upgrade has been cancelled"))
368 runningDeployments.update(id, info.copy(status = DeploymentStatus.Canceling, cancel = Some(stopFuture)))
118369 }
119370
120371 def stopActor(ref: ActorRef, reason: Throwable): Future[Boolean] = {
121372 val promise = Promise[Boolean]()
122373 context.actorOf(Props(classOf[StopActor], ref, promise, reason))
123374 promise.future
124375 }
376
377 def isScheduledDeployment(id: String): Boolean = {
378 runningDeployments.contains(id) && runningDeployments(id).status == DeploymentStatus.Scheduled
125379 }
126380
381 def hasConflicts(plan: DeploymentPlan): Boolean = {
382 conflictingDeployments(plan).nonEmpty
383 }
384
385 /**
386 * Methods return all deployments that are conflicting with passed plan.
387 */
388 def conflictingDeployments(thisPlan: DeploymentPlan): Seq[DeploymentInfo] = {
389 def intersectsWith(thatPlan: DeploymentPlan): Boolean = {
390 thatPlan.affectedRunSpecIds.intersect(thisPlan.affectedRunSpecIds).nonEmpty
391 }
392 runningDeployments.values.filter(info => intersectsWith(info.plan)).to[Seq]
393 }
394}
395
127396object DeploymentManager {
128 case class PerformDeployment(driver: SchedulerDriver, plan: DeploymentPlan)
397 case class StartDeployment(plan: DeploymentPlan, origSender: ActorRef, force: Boolean = false)
129398 case class CancelDeployment(id: String)
130 case object StopAllDeployments
131 case class CancelConflictingDeployments(plan: DeploymentPlan)
399 case object ShutdownDeployments
400 case class WaitForCanceledConflicts(plan: DeploymentPlan, conflicts: Seq[DeploymentInfo], origSender: ActorRef)
401 case class CancelDeletedConflicts(plan: DeploymentPlan, conflicts: Seq[DeploymentInfo], recipient: ActorRef, origSender: ActorRef)
402 case class DeploymentFinished(plan: DeploymentPlan)
403 case class DeploymentFailed(plan: DeploymentPlan, reason: Throwable)
404 case class ReadinessCheckUpdate(deploymentId: String, result: ReadinessCheckResult)
405 case class LaunchDeploymentActor(plan: DeploymentPlan, origSender: ActorRef)
406 case class FailedRepositoryOperation(plan: DeploymentPlan, recipient: ActorRef, reason: Throwable)
407 case object LoadDeploymentsOnLeaderElection
132408
133409 case class DeploymentStepInfo(
134410 plan: DeploymentPlan,
135411 step: DeploymentStep,
136412 nr: Int,
137413 readinessChecks: Map[Task.Id, ReadinessCheckResult] = Map.empty) {
138414 lazy val readinessChecksByApp: Map[PathId, Seq[ReadinessCheckResult]] = {
139415 readinessChecks.values.groupBy(_.taskId.runSpecId).mapValues(_.to[Seq]).withDefaultValue(Seq.empty)
140416 }
141417 }
142418
143 case class DeploymentFinished(plan: DeploymentPlan)
144 case class DeploymentFailed(plan: DeploymentPlan, reason: Throwable)
145 case class AllDeploymentsCanceled(plans: Seq[DeploymentPlan])
146 case class ConflictingDeploymentsCanceled(id: String, deployments: Seq[DeploymentPlan])
147 case class ReadinessCheckUpdate(deploymentId: String, result: ReadinessCheckResult)
148
149419 case class DeploymentInfo(
150 ref: ActorRef,
151 plan: DeploymentPlan)
420 ref: Option[ActorRef] = None, // An ActorRef to the DeploymentActor if status = [Deploying]
421 plan: DeploymentPlan, // Deployment plan
422 status: DeploymentStatus, // Status can be [Scheduled], [Canceling] or [Deploying]
423 cancel: Option[Future[Boolean]] = None) // Cancellation future if status = [Canceling]
152424
425 sealed trait DeploymentStatus
426 object DeploymentStatus {
aquamatthiasUnsubmitted
Done

Please move all objects into the companion object.

427 case object Scheduled extends DeploymentStatus
428 case object Canceling extends DeploymentStatus
429 case object Deploying extends DeploymentStatus
430 }
431
432 @SuppressWarnings(Array("MaxParameters"))
153433 def props(
154434 taskTracker: InstanceTracker,
155435 killService: KillService,
156436 launchQueue: LaunchQueue,
157437 scheduler: SchedulerActions,
158438 storage: StorageProvider,
159439 healthCheckManager: HealthCheckManager,
160440 eventBus: EventStream,
161 readinessCheckExecutor: ReadinessCheckExecutor): Props = {
441 readinessCheckExecutor: ReadinessCheckExecutor,
442 deploymentRepository: DeploymentRepository,
443 deploymentActorProps: (ActorRef, ActorRef, KillService, SchedulerActions, DeploymentPlan, InstanceTracker, LaunchQueue, StorageProvider, HealthCheckManager, EventStream, ReadinessCheckExecutor) => Props = DeploymentActor.props)(implicit mat: Materializer): Props = {
162444 Props(new DeploymentManager(taskTracker, killService, launchQueue,
163 scheduler, storage, healthCheckManager, eventBus, readinessCheckExecutor))
445 scheduler, storage, healthCheckManager, eventBus, readinessCheckExecutor, deploymentRepository, deploymentActorProps))
164446 }
165447
166448}
View Options

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

11package mesosphere.marathon.upgrade
22
33import akka.actor.Actor
44import akka.event.EventStream
55import mesosphere.marathon.SchedulerActions
6import mesosphere.marathon.core.condition.Condition.Terminal
67import mesosphere.marathon.core.event.{ InstanceChanged, InstanceHealthChanged }
78import mesosphere.marathon.core.instance.Instance
8import mesosphere.marathon.core.condition.Condition.Terminal
99import mesosphere.marathon.core.launchqueue.LaunchQueue
1010import mesosphere.marathon.core.task.tracker.InstanceTracker
11import org.apache.mesos.SchedulerDriver
1211import org.slf4j.LoggerFactory
1312
1413import scala.concurrent.duration._
1514
1615trait StartingBehavior extends ReadinessBehavior { this: Actor =>
1716 import context.dispatcher
1817 import mesosphere.marathon.upgrade.StartingBehavior._
1918
2019 def eventBus: EventStream
2120 def scaleTo: Int
2221 def nrToStart: Int
2322 def launchQueue: LaunchQueue
24 def driver: SchedulerDriver
2523 def scheduler: SchedulerActions
2624 def instanceTracker: InstanceTracker
2725
2826 def initializeStart(): Unit
2927
3028 private[this] val log = LoggerFactory.getLogger(getClass)
3129
3230 final override def preStart(): Unit = {
▲ Show 20 Lines • Show All 52 Lines • Show Last 20 Lines
View Options

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

11package mesosphere.marathon.upgrade
22
33import akka.actor._
44import akka.event.EventStream
55import mesosphere.marathon._
66import mesosphere.marathon.core.event._
77import mesosphere.marathon.core.instance.Instance
88import mesosphere.marathon.core.instance.Instance.Id
99import mesosphere.marathon.core.launchqueue.LaunchQueue
1010import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1111import mesosphere.marathon.core.task.termination.{ KillReason, KillService }
1212import mesosphere.marathon.core.task.tracker.InstanceTracker
1313import mesosphere.marathon.core.task.termination.InstanceChangedPredicates.considerTerminal
1414import mesosphere.marathon.state.RunSpec
1515import mesosphere.marathon.upgrade.TaskReplaceActor._
16import org.apache.mesos.SchedulerDriver
1716import org.slf4j.LoggerFactory
1817
1918import scala.collection.{ SortedSet, mutable }
2019import scala.concurrent.Promise
2120
2221class TaskReplaceActor(
2322 val deploymentManager: ActorRef,
2423 val status: DeploymentStatus,
25 val driver: SchedulerDriver,
2624 val killService: KillService,
2725 val launchQueue: LaunchQueue,
2826 val instanceTracker: InstanceTracker,
2927 val eventBus: EventStream,
3028 val readinessCheckExecutor: ReadinessCheckExecutor,
3129 val runSpec: RunSpec,
3230 promise: Promise[Unit]) extends Actor with ReadinessBehavior with ActorLogging {
3331
▲ Show 20 Lines • Show All 134 Lines • ▼ Show 20 Line(s)
168166
169167object TaskReplaceActor {
170168 private[this] val log = LoggerFactory.getLogger(getClass)
171169
172170 //scalastyle:off
173171 def props(
174172 deploymentManager: ActorRef,
175173 status: DeploymentStatus,
176 driver: SchedulerDriver,
177174 killService: KillService,
178175 launchQueue: LaunchQueue,
179176 instanceTracker: InstanceTracker,
180177 eventBus: EventStream,
181178 readinessCheckExecutor: ReadinessCheckExecutor,
182179 app: RunSpec,
183180 promise: Promise[Unit]): Props = Props(
184 new TaskReplaceActor(deploymentManager, status, driver, killService, launchQueue, instanceTracker, eventBus,
181 new TaskReplaceActor(deploymentManager, status, killService, launchQueue, instanceTracker, eventBus,
185182 readinessCheckExecutor, app, promise)
186183 )
187184
188185 /** Encapsulates the logic how to get a Restart going */
189186 private[upgrade] case class RestartStrategy(nrToKillImmediately: Int, maxCapacity: Int)
190187
191188 private[upgrade] def computeRestartStrategy(runSpec: RunSpec, runningInstancesCount: Int): RestartStrategy = {
192189 // in addition to a spec which passed validation, we require:
Show All 35 Lines
View Options

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

11
22package mesosphere.marathon.upgrade
33
44import akka.actor.{ Actor, ActorLogging, ActorRef, Props }
55import akka.event.EventStream
6import mesosphere.marathon.core.event.DeploymentStatus
67import mesosphere.marathon.core.launchqueue.LaunchQueue
78import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
89import mesosphere.marathon.core.task.tracker.InstanceTracker
9import mesosphere.marathon.core.event.DeploymentStatus
1010import mesosphere.marathon.state.RunSpec
1111import mesosphere.marathon.{ SchedulerActions, TaskUpgradeCanceledException }
12import org.apache.mesos.SchedulerDriver
1312
1413import scala.concurrent.Promise
1514
1615class TaskStartActor(
1716 val deploymentManager: ActorRef,
1817 val status: DeploymentStatus,
19 val driver: SchedulerDriver,
2018 val scheduler: SchedulerActions,
2119 val launchQueue: LaunchQueue,
2220 val instanceTracker: InstanceTracker,
2321 val eventBus: EventStream,
2422 val readinessCheckExecutor: ReadinessCheckExecutor,
2523 val runSpec: RunSpec,
2624 val scaleTo: Int,
2725 promise: Promise[Unit]) extends Actor with ActorLogging with StartingBehavior {
Show All 28 Lines
5654 }
5755}
5856
5957object TaskStartActor {
6058 @SuppressWarnings(Array("MaxParameters"))
6159 def props(
6260 deploymentManager: ActorRef,
6361 status: DeploymentStatus,
64 driver: SchedulerDriver,
6562 scheduler: SchedulerActions,
6663 launchQueue: LaunchQueue,
6764 instanceTracker: InstanceTracker,
6865 eventBus: EventStream,
6966 readinessCheckExecutor: ReadinessCheckExecutor,
7067 runSpec: RunSpec,
7168 scaleTo: Int,
7269 promise: Promise[Unit]): Props = {
73 Props(new TaskStartActor(deploymentManager, status, driver, scheduler, launchQueue, instanceTracker,
70 Props(new TaskStartActor(deploymentManager, status, scheduler, launchQueue, instanceTracker,
7471 eventBus, readinessCheckExecutor, runSpec, scaleTo, promise)
7572 )
7673 }
7774}
View Options

src/test/scala/mesosphere/marathon/MarathonSchedulerActorTest.scala

11package mesosphere.marathon
22
3import java.util.concurrent.TimeoutException
4
53import akka.Done
64import akka.actor.{ ActorRef, Props }
75import akka.event.EventStream
86import akka.stream.scaladsl.Source
97import akka.testkit._
108import akka.util.Timeout
11import mesosphere.Unstable
129import mesosphere.marathon.MarathonSchedulerActor._
1310import mesosphere.marathon.core.condition.Condition
1411import mesosphere.marathon.core.election.{ ElectionService, LocalLeadershipEvent }
1512import mesosphere.marathon.core.event._
1613import mesosphere.marathon.core.health.HealthCheckManager
1714import mesosphere.marathon.core.history.impl.HistoryActor
1815import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
1916import mesosphere.marathon.core.launcher.impl.LaunchQueueTestHelper
▲ Show 20 Lines • Show All 433 Lines • ▼ Show 20 Line(s)
453450 deploymentRepo.store(plan) returns Future.successful(Done)
454451 instanceTracker.specInstancesLaunchedSync(app.id) returns Seq.empty[Instance]
455452
456453 val schedulerActor = system.actorOf(
457454 MarathonSchedulerActor.props(
458455 schedulerActions,
459456 deploymentManagerProps,
460457 historyActorProps,
461 deploymentRepo,
462458 hcManager,
463459 killService,
464460 queue,
465461 holder,
466462 electionService,
467463 system.eventStream
468464 ))
469465
Show All 12 Lines
482478 }
483479
484480 test("Forced deployment") {
485481 val f = new Fixture
486482 import f._
487483 val app = AppDefinition(id = PathId("app1"), cmd = Some("cmd"), instances = 2, upgradeStrategy = UpgradeStrategy(0.5))
488484 val rootGroup = createRootGroup(groups = Set(createGroup(PathId("/foo/bar"), Map(app.id -> app))))
489485
490 val plan = DeploymentPlan(createRootGroup(), rootGroup)
486 val plan = DeploymentPlan(createRootGroup(), rootGroup, id = Some("d1"))
491487
492488 appRepo.store(any) returns Future.successful(Done)
493489 appRepo.get(app.id) returns Future.successful(None)
494490 instanceTracker.specInstancesLaunchedSync(app.id) returns Seq.empty[Instance]
491 instanceTracker.specInstancesSync(app.id) returns Seq.empty[Instance]
495492 appRepo.delete(app.id) returns Future.successful(Done)
496493
497494 val schedulerActor = createActor()
498495 try {
499496 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
500497 schedulerActor ! Deploy(plan)
501498
502499 expectMsgType[DeploymentStarted](10.seconds)
503500
504 schedulerActor ! Deploy(plan, force = true)
501 schedulerActor ! Deploy(plan.copy(id = "d2"), force = true)
505502
506503 expectMsgType[DeploymentStarted]
507504
508505 } finally {
509506 stopActor(schedulerActor)
510507 }
511508 }
512509
513 // TODO: Fix this test...
514 test("Cancellation timeout - this test is really racy and fails intermittently.", Unstable) {
515 val f = new Fixture
516 import f._
517 val app = AppDefinition(id = PathId("app1"), cmd = Some("cmd"), instances = 2, upgradeStrategy = UpgradeStrategy(0.5))
518 val rootGroup = createRootGroup(Map(app.id -> app), groups = Set(createGroup(PathId("/foo/bar"))))
519
520 val plan = DeploymentPlan(createRootGroup(), rootGroup)
521
522 appRepo.store(any) returns Future.successful(Done)
523 appRepo.get(app.id) returns Future.successful(None)
524 instanceTracker.specInstancesLaunchedSync(app.id) returns Seq.empty[Instance]
525 appRepo.delete(app.id) returns Future.successful(Done)
526
527 val schedulerActor = TestActorRef[MarathonSchedulerActor](
528 MarathonSchedulerActor.props(
529 schedulerActions,
530 deploymentManagerProps,
531 historyActorProps,
532 deploymentRepo,
533 hcManager,
534 killService,
535 queue,
536 holder,
537 electionService,
538 system.eventStream
539 )
540 )
541 try {
542 val probe = TestProbe()
543 schedulerActor.tell(LocalLeadershipEvent.ElectedAsLeader, probe.testActor)
544 schedulerActor.tell(Deploy(plan), probe.testActor)
545
546 probe.expectMsgType[DeploymentStarted]
547
548 schedulerActor.tell(Deploy(plan, force = true), probe.testActor)
549
550 val answer = probe.expectMsgType[CommandFailed]
551
552 answer.reason.isInstanceOf[TimeoutException] should be(true)
553 answer.reason.getMessage should be
554
555 // this test has more messages sometimes!
556 // needs: probe.expectNoMsg()
557 } finally {
558 stopActor(schedulerActor)
559 }
560 }
561
562510 test("Do not run reconciliation concurrently") {
aquamatthiasUnsubmitted
Done

yes.

563511 val f = new Fixture
564512 import f._
565513 val actions = mock[SchedulerActions]
566514 val actionsFactory: ActorRef => SchedulerActions = _ => actions
567515 val schedulerActor = createActor(Some(actionsFactory))
568516
569517 val reconciliationPromise = Promise[Status]()
570518 actions.reconcileTasks(any) returns reconciliationPromise.future
▲ Show 20 Lines • Show All 64 Lines • ▼ Show 20 Line(s)
635583 val deploymentManagerProps: SchedulerActions => Props = schedulerActions => Props(new DeploymentManager(
636584 instanceTracker,
637585 killService,
638586 queue,
639587 schedulerActions,
640588 storage,
641589 hcManager,
642590 system.eventStream,
643 readinessCheckExecutor
591 readinessCheckExecutor,
592 deploymentRepo
644593 ))
645594
646595 val historyActorProps: Props = Props(new HistoryActor(system.eventStream, taskFailureEventRepository))
647596
648597 def createActor(overrideActions: Option[(ActorRef) => SchedulerActions] = None) = {
649598 val actions = overrideActions.getOrElse(schedulerActions)
650599 system.actorOf(
651600 MarathonSchedulerActor.props(
652601 actions,
653602 deploymentManagerProps,
654603 historyActorProps,
655 deploymentRepo,
656604 hcManager,
657605 killService,
658606 queue,
659607 holder,
660608 electionService,
661609 system.eventStream
662610 )
663611 )
Show All 21 Lines
View Options

src/test/scala/mesosphere/marathon/integration/AppDeployWithLeaderAbdicationIntegrationTest.scala

Show First 20 Lines • Show All 49 Lines • ▼ Show 20 Line(s)
5050 upgradeStrategy = Some(UpgradeStrategy(minimumHealthCapacity = 1.0))))
5151
5252 And("new and updated task is started successfully")
5353 val updated = waitForTasks(appId, 2, maxWait = 90.seconds) //make sure, the new task has really started
5454
5555 val updatedTask = updated.diff(started.value).head
5656 val updatedTaskIds: List[String] = updated.map(_.id).diff(startedTaskIds)
5757
58 And("service mock is responding")
59 val serviceFacade = new ServiceMockFacade(updatedTask)
60 WaitTestSupport.waitUntil("ServiceMock is up", 30.seconds){ Try(serviceFacade.plan()).isSuccess }
61
5862 log.info(s"Updated app: ${marathon.app(appId).entityPrettyJsonString}")
5963
6064 When("marathon leader is abdicated")
6165 val leader = marathon.leader().value
6266 marathon.abdicate().code should be (200)
6367
6468 And("a new leader is elected")
6569 WaitTestSupport.waitUntil("the leader changes", 30.seconds) { marathon.leader().value != leader }
6670
6771 And("the updated task becomes healthy")
68 val serviceFacade = new ServiceMockFacade(updatedTask)
69 WaitTestSupport.waitUntil("ServiceMock is up", 30.seconds){ Try(serviceFacade.plan()).isSuccess }
7072 // This would move the service mock from "InProgress" [HTTP 503] to "Complete" [HTTP 200]
7173 serviceFacade.continue()
7274 waitForEvent("health_status_changed_event")
7375
7476 Then("the app should have only 1 task launched")
7577 waitForTasks(appId, 1) should have size 1
7678
7779 And("app was deployed successfully")
▲ Show 20 Lines • Show All 47 Lines • Show Last 20 Lines
View Options

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

11package mesosphere.marathon.upgrade
22
33import akka.testkit.{ TestActorRef, TestProbe }
44import mesosphere.marathon.core.condition.Condition
55import mesosphere.marathon.core.event.{ DeploymentStatus, InstanceChanged, InstanceHealthChanged }
66import mesosphere.marathon.core.health.{ MarathonHttpHealthCheck, PortReference }
77import mesosphere.marathon.core.instance.Instance
88import mesosphere.marathon.core.launchqueue.LaunchQueue
99import mesosphere.marathon.core.leadership.AlwaysElectedLeadershipModule
1010import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1111import mesosphere.marathon.core.task.tracker.InstanceTracker
1212import mesosphere.marathon.state.{ AppDefinition, PathId }
1313import mesosphere.marathon.test.{ MarathonActorSupport, MarathonSpec, MarathonTestHelper, Mockito }
1414import mesosphere.marathon.{ AppStartCanceledException, SchedulerActions }
15import org.apache.mesos.SchedulerDriver
1615import org.scalatest.{ BeforeAndAfter, Matchers }
1716
1817import scala.concurrent.duration._
1918import scala.concurrent.{ Await, Future, Promise }
2019
2120class AppStartActorTest
2221 extends MarathonActorSupport
2322 with MarathonSpec
▲ Show 20 Lines • Show All 82 Lines • ▼ Show 20 Line(s)
106105 Await.result(promise.future, 5.seconds)
107106
108107 verify(f.scheduler).startRunSpec(app.copy(instances = 0))
109108 expectTerminated(ref)
110109 }
111110
112111 class Fixture {
113112
114 val driver: SchedulerDriver = mock[SchedulerDriver]
115113 val scheduler: SchedulerActions = mock[SchedulerActions]
116114 val launchQueue: LaunchQueue = mock[LaunchQueue]
117115 val taskTracker: InstanceTracker = MarathonTestHelper.createTaskTracker(AlwaysElectedLeadershipModule.forActorSystem(system))
118116 val deploymentManager: TestProbe = TestProbe()
119117 val deploymentStatus: DeploymentStatus = mock[DeploymentStatus]
120118 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
121119
122120 def instanceChanged(app: AppDefinition, condition: Condition): InstanceChanged = {
123121 val instanceId = Instance.Id.forRunSpec(app.id)
124122 val instance: Instance = mock[Instance]
125123 instance.instanceId returns instanceId
126124 InstanceChanged(instanceId, app.version, app.id, condition, instance)
127125 }
128126
129127 def healthChanged(app: AppDefinition, healthy: Boolean): InstanceHealthChanged = {
130128 InstanceHealthChanged(Instance.Id.forRunSpec(app.id), app.version, app.id, healthy = Some(healthy))
131129 }
132130
133131 def startActor(app: AppDefinition, scaleTo: Int, promise: Promise[Unit]): TestActorRef[AppStartActor] =
134 TestActorRef(AppStartActor.props(deploymentManager.ref, deploymentStatus, driver, scheduler,
132 TestActorRef(AppStartActor.props(deploymentManager.ref, deploymentStatus, scheduler,
135133 launchQueue, taskTracker, system.eventStream, readinessCheckExecutor, app, scaleTo, promise)
136134 )
137135 }
138136}
View Options

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

11package mesosphere.marathon
22package upgrade
33
44import akka.actor.{ ActorRef, ActorSystem }
55import akka.testkit.{ TestActorRef, TestProbe }
66import akka.util.Timeout
7import mesosphere.marathon.SchedulerActions
87import mesosphere.marathon.core.condition.Condition
98import mesosphere.marathon.core.event.InstanceChanged
109import mesosphere.marathon.core.health.HealthCheckManager
11import mesosphere.marathon.core.instance.Instance
10import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
1211import mesosphere.marathon.core.launchqueue.LaunchQueue
1312import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
14import mesosphere.marathon.core.task.tracker.InstanceTracker
1513import mesosphere.marathon.core.task.KillServiceMock
16import mesosphere.marathon.core.event.InstanceChanged
17import mesosphere.marathon.core.health.HealthCheckManager
18import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
14import mesosphere.marathon.core.task.tracker.InstanceTracker
1915import mesosphere.marathon.io.storage.StorageProvider
2016import mesosphere.marathon.state._
2117import mesosphere.marathon.test.{ GroupCreation, MarathonSpec, Mockito }
2218import mesosphere.marathon.upgrade.DeploymentManager.{ DeploymentFinished, DeploymentStepInfo }
23import org.apache.mesos.SchedulerDriver
24import org.mockito.Mockito.{ verifyNoMoreInteractions, when }
19import org.mockito.Mockito.when
2520import org.mockito.invocation.InvocationOnMock
2621import org.mockito.stubbing.Answer
2722import org.scalatest.{ BeforeAndAfterAll, Matchers }
2823
2924import scala.concurrent.Await
3025import scala.concurrent.duration._
3126
3227// TODO: this is NOT a unit test. the DeploymentActor create child actors that cannot be mocked in the current
▲ Show 20 Lines • Show All 193 Lines • ▼ Show 20 Line(s)
226221 plan.steps.zipWithIndex.foreach {
227222 case (step, num) => managerProbe.expectMsg(5.seconds, DeploymentStepInfo(plan, step, num + 1))
228223 }
229224
230225 managerProbe.expectMsg(5.seconds, DeploymentFinished(plan))
231226
232227 f.killService.numKilled should be (1)
233228 f.killService.killed should contain (instance1_2.instanceId)
234 verifyNoMoreInteractions(f.driver)
235229 } finally {
236230 Await.result(system.terminate(), Duration.Inf)
237231 }
238232 }
239233
240234 class Fixture {
241235 implicit val system = ActorSystem("TestSystem")
242236 val tracker: InstanceTracker = mock[InstanceTracker]
243237 val queue: LaunchQueue = mock[LaunchQueue]
244 val driver: SchedulerDriver = mock[SchedulerDriver]
245238 val killService = new KillServiceMock(system)
246239 val scheduler: SchedulerActions = mock[SchedulerActions]
247240 val storage: StorageProvider = mock[StorageProvider]
248241 val hcManager: HealthCheckManager = mock[HealthCheckManager]
249242 val config: UpgradeConfig = mock[UpgradeConfig]
250243 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
251244 config.killBatchSize returns 100
252245 config.killBatchCycle returns 10.seconds
253246
254247 def instanceChanged(app: AppDefinition, condition: Condition): InstanceChanged = {
255248 val instanceId = Instance.Id.forRunSpec(app.id)
256249 val instance: Instance = mock[Instance]
257250 instance.instanceId returns instanceId
258251 InstanceChanged(instanceId, app.version, app.id, condition, instance)
259252 }
260253
261254 def deploymentActor(manager: ActorRef, receiver: ActorRef, plan: DeploymentPlan) = TestActorRef(
262255 DeploymentActor.props(
263256 manager,
264257 receiver,
265 driver,
266258 killService,
267259 scheduler,
268260 plan,
269261 tracker,
270262 queue,
271263 storage,
272264 hcManager,
273265 system.eventStream,
274266 readinessCheckExecutor
275267 )
276268 )
277269
278270 }
279271}
View Options

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

11package mesosphere.marathon.upgrade
22
3import akka.actor.ActorRef
3import java.util.concurrent.LinkedBlockingDeque
4
5import akka.Done
6import akka.actor.{ ActorRef, Props }
47import akka.event.EventStream
8import akka.stream.scaladsl.Source
59import akka.testkit.TestActor.{ AutoPilot, NoAutoPilot }
6import akka.testkit.{ ImplicitSender, TestActorRef, TestProbe }
10import akka.testkit.{ ImplicitSender, TestActor, TestActorRef, TestProbe }
711import akka.util.Timeout
812import com.codahale.metrics.MetricRegistry
13import mesosphere.marathon.MarathonSchedulerActor.{ CommandFailed, DeploymentStarted, LoadedDeploymentsOnLeaderElection }
914import mesosphere.marathon.core.health.HealthCheckManager
1015import mesosphere.marathon.core.launchqueue.LaunchQueue
1116import mesosphere.marathon.core.leadership.AlwaysElectedLeadershipModule
1217import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1318import mesosphere.marathon.core.task.termination.KillService
1419import mesosphere.marathon.core.task.tracker.InstanceTracker
1520import mesosphere.marathon.io.storage.StorageProvider
1621import mesosphere.marathon.metrics.Metrics
1722import mesosphere.marathon.state.PathId._
1823import mesosphere.marathon.state.{ AppDefinition, PathId }
19import mesosphere.marathon.storage.repository.AppRepository
2024import mesosphere.marathon.storage.repository.legacy.AppEntityRepository
2125import mesosphere.marathon.storage.repository.legacy.store.{ InMemoryStore, MarathonStore }
26import mesosphere.marathon.storage.repository.{ AppRepository, DeploymentRepository }
2227import mesosphere.marathon.test.{ GroupCreation, MarathonActorSupport, MarathonTestHelper, Mockito }
2328import mesosphere.marathon.upgrade.DeploymentActor.Cancel
24import mesosphere.marathon.upgrade.DeploymentManager.{ CancelDeployment, DeploymentFailed, PerformDeployment, StopAllDeployments }
29import mesosphere.marathon.upgrade.DeploymentManager._
2530import mesosphere.marathon.{ MarathonConf, SchedulerActions }
2631import org.apache.mesos.SchedulerDriver
2732import org.rogach.scallop.ScallopConf
2833import org.scalatest.concurrent.Eventually
2934import org.scalatest.time.{ Seconds, Span }
3035import org.scalatest.{ BeforeAndAfter, FunSuiteLike, Matchers }
36import org.slf4j.LoggerFactory
3137
3238import scala.concurrent.duration._
33import scala.concurrent.{ Await, ExecutionContext }
39import scala.concurrent.{ Await, ExecutionContext, Future }
3440
3541class DeploymentManagerTest
3642 extends MarathonActorSupport
3743 with FunSuiteLike
3844 with Matchers
3945 with BeforeAndAfter
4046 with Mockito
4147 with Eventually
4248 with ImplicitSender
4349 with GroupCreation {
4450
45 test("deploy") {
51 private[this] val log = LoggerFactory.getLogger(getClass)
52
53 test("Deployment") {
4654 val f = new Fixture
4755 val manager = f.deploymentManager()
4856 val app = AppDefinition("app".toRootPath)
4957
5058 val oldGroup = createRootGroup()
5159 val newGroup = createRootGroup(Map(app.id -> app))
5260 val plan = DeploymentPlan(oldGroup, newGroup)
5361
54 f.launchQueue.get(app.id) returns None
55 manager ! PerformDeployment(f.driver, plan)
62 manager ! LoadDeploymentsOnLeaderElection
63 expectMsgType[LoadedDeploymentsOnLeaderElection]
5664
57 awaitCond(
58 manager.underlyingActor.runningDeployments.contains(plan.id),
59 5.seconds
60 )
65 manager ! StartDeployment(plan, ActorRef.noSender)
66
67 awaitCond(manager.underlyingActor.runningDeployments.contains(plan.id), 5.seconds)
68 manager.underlyingActor.runningDeployments(plan.id).status should be (DeploymentStatus.Deploying)
6169 }
6270
71 test("Finished deployment") {
72 val f = new Fixture
73 val manager = f.deploymentManager()
74 val app = AppDefinition("app".toRootPath)
75
76 val oldGroup = createRootGroup()
77 val newGroup = createRootGroup(Map(app.id -> app))
78 val plan = DeploymentPlan(oldGroup, newGroup)
79
80 manager ! LoadDeploymentsOnLeaderElection
81 expectMsgType[LoadedDeploymentsOnLeaderElection]
82
83 manager ! StartDeployment(plan, ActorRef.noSender)
84
85 awaitCond(manager.underlyingActor.runningDeployments.contains(plan.id), 5.seconds)
86 manager.underlyingActor.runningDeployments(plan.id).status should be (DeploymentStatus.Deploying)
87
88 manager ! DeploymentFinished(plan)
89 awaitCond(manager.underlyingActor.runningDeployments.isEmpty, 5.seconds)
90 }
91
92 test("Conflicting not forced deployment") {
93 val f = new Fixture
94 val manager = f.deploymentManager()
95 val app = AppDefinition("app".toRootPath)
96
97 val oldGroup = createRootGroup()
98 val newGroup = createRootGroup(Map(app.id -> app))
99 val plan = DeploymentPlan(oldGroup, newGroup, id = Some("d1"))
100
101 manager ! LoadDeploymentsOnLeaderElection
102 expectMsgType[LoadedDeploymentsOnLeaderElection]
103
104 manager ! StartDeployment(plan, ActorRef.noSender)
105
106 awaitCond(manager.underlyingActor.runningDeployments.contains(plan.id), 5.seconds)
107 manager.underlyingActor.runningDeployments(plan.id).status should be (DeploymentStatus.Deploying)
108
109 manager ! StartDeployment(plan.copy(id = "d2"), self, force = false)
110 expectMsgType[CommandFailed]
111 manager.underlyingActor.runningDeployments.size should be (1)
112 manager.underlyingActor.runningDeployments(plan.id).status should be (DeploymentStatus.Deploying)
113 }
114
115 test("Conflicting forced deployment") {
116 val f = new Fixture
117 val manager = f.deploymentManager()
118 val app = AppDefinition("app".toRootPath)
119
120 val oldGroup = createRootGroup()
121 val newGroup = createRootGroup(Map(app.id -> app))
122 val plan = DeploymentPlan(oldGroup, newGroup, id = Some("b1"))
123
124 manager ! LoadDeploymentsOnLeaderElection
125 expectMsgType[LoadedDeploymentsOnLeaderElection]
126
127 manager ! StartDeployment(plan, self)
128 expectMsgType[DeploymentStarted]
129
130 awaitCond(manager.underlyingActor.runningDeployments.contains(plan.id), 5.seconds)
131 manager.underlyingActor.runningDeployments(plan.id).status should be (DeploymentStatus.Deploying)
132
133 manager ! StartDeployment(plan.copy(id = "d2"), self, force = true)
134 expectMsgType[DeploymentStarted]
135 manager.underlyingActor.runningDeployments(plan.id).status should be (DeploymentStatus.Canceling)
136 eventually(manager.underlyingActor.runningDeployments("d2").status should be (DeploymentStatus.Deploying))
137 }
138
139 test("Multiple conflicting forced deployments") {
140 val f = new Fixture
141 val manager = f.deploymentManager()
142 val app = AppDefinition("app".toRootPath)
143
144 val oldGroup = createRootGroup()
145 val newGroup = createRootGroup(Map(app.id -> app))
146 val plan = DeploymentPlan(oldGroup, newGroup, id = Some("d1"))
147
148 manager ! LoadDeploymentsOnLeaderElection
149 expectMsgType[LoadedDeploymentsOnLeaderElection]
150
151 manager ! StartDeployment(plan, self)
152 expectMsgType[DeploymentStarted]
153 manager.underlyingActor.runningDeployments("d1").status should be (DeploymentStatus.Deploying)
154
155 manager ! StartDeployment(plan.copy(id = "d2"), self, force = true)
156 expectMsgType[DeploymentStarted]
157 manager.underlyingActor.runningDeployments("d1").status should be (DeploymentStatus.Canceling)
158 manager.underlyingActor.runningDeployments("d2").status should be (DeploymentStatus.Deploying)
159
160 manager ! StartDeployment(plan.copy(id = "d3"), self, force = true)
161 expectMsgType[DeploymentStarted]
162
163 // Since deployments are not really started (DeploymentActor is not spawned), DeploymentFinished event is not
164 // sent and the deployments are staying in the list of runningDeployments
165 manager.underlyingActor.runningDeployments("d1").status should be (DeploymentStatus.Canceling)
166 manager.underlyingActor.runningDeployments("d2").status should be (DeploymentStatus.Canceling)
167 manager.underlyingActor.runningDeployments("d3").status should be (DeploymentStatus.Scheduled)
168 }
169
63170 test("StopActor") {
64171 val f = new Fixture
65172 val manager = f.deploymentManager()
66173 val probe = TestProbe()
67174
68175 probe.setAutoPilot(new AutoPilot {
69176 override def run(sender: ActorRef, msg: Any): AutoPilot = msg match {
70177 case Cancel(_) =>
Show All 14 Lines
85192 val manager = f.deploymentManager()
86193 implicit val timeout = Timeout(1.minute)
87194
88195 val app = AppDefinition("app".toRootPath)
89196 val oldGroup = createRootGroup()
90197 val newGroup = createRootGroup(Map(app.id -> app))
91198 val plan = DeploymentPlan(oldGroup, newGroup)
92199
93 manager ! PerformDeployment(f.driver, plan)
200 manager ! LoadDeploymentsOnLeaderElection
201 expectMsgType[LoadedDeploymentsOnLeaderElection]
94202
95 manager ! CancelDeployment(plan.id)
203 manager ! StartDeployment(plan, self)
204 expectMsgType[DeploymentStarted]
96205
97 expectMsgType[DeploymentFailed]
206 manager ! CancelDeployment(plan.id)
207 eventually(manager.underlyingActor.runningDeployments(plan.id).status should be (DeploymentStatus.Canceling))
98208 }
99209
100 test("Stop All Deployments") {
210 test("Shutdown deployments") {
101211 val f = new Fixture
102212 val manager = f.deploymentManager()
103213 implicit val timeout = Timeout(1.minute)
104214
105215 val app1 = AppDefinition("app1".toRootPath)
106216 val app2 = AppDefinition("app2".toRootPath)
107217 val oldGroup = createRootGroup()
108 manager ! PerformDeployment(f.driver, DeploymentPlan(oldGroup, createRootGroup(Map(app1.id -> app1))))
109 manager ! PerformDeployment(f.driver, DeploymentPlan(oldGroup, createRootGroup(Map(app2.id -> app2))))
218
219 manager ! LoadDeploymentsOnLeaderElection
220 expectMsgType[LoadedDeploymentsOnLeaderElection]
221
222 manager ! StartDeployment(DeploymentPlan(oldGroup, createRootGroup(Map(app1.id -> app1))), ActorRef.noSender)
223 manager ! StartDeployment(DeploymentPlan(oldGroup, createRootGroup(Map(app2.id -> app2))), ActorRef.noSender)
110224 eventually(manager.underlyingActor.runningDeployments should have size 2)
111225
112 manager ! StopAllDeployments
226 manager ! ShutdownDeployments
113227 eventually(manager.underlyingActor.runningDeployments should have size 0)
114228 }
115229
116230 override implicit def patienceConfig: PatienceConfig = PatienceConfig(Span(3, Seconds))
117231
118232 class Fixture {
119233
120234 val driver: SchedulerDriver = mock[SchedulerDriver]
235 val deploymentRepo = mock[DeploymentRepository]
121236 val eventBus: EventStream = mock[EventStream]
122237 val launchQueue: LaunchQueue = mock[LaunchQueue]
123238 val config: MarathonConf = new ScallopConf(Seq("--master", "foo")) with MarathonConf {
124239 verify()
125240 }
126241 val metrics: Metrics = new Metrics(new MetricRegistry)
127242 val taskTracker: InstanceTracker = MarathonTestHelper.createTaskTracker(
128243 AlwaysElectedLeadershipModule.forActorSystem(system), new InMemoryStore
129244 )
130245 val taskKillService: KillService = mock[KillService]
131246 val scheduler: SchedulerActions = mock[SchedulerActions]
132247 val appRepo: AppRepository = new AppEntityRepository(
133248 new MarathonStore[AppDefinition](new InMemoryStore, metrics, () => AppDefinition(id = PathId("/test")), prefix = "app:"),
134249 0
135250 )(ExecutionContext.global, metrics)
136251 val storage: StorageProvider = mock[StorageProvider]
137252 val hcManager: HealthCheckManager = mock[HealthCheckManager]
138253 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
139254
255 // A method that returns dummy props. Used to control the deployments progress. Otherwise the tests become racy
256 // and depending on when DeploymentActor sends DeploymentFinished message.
257 val deploymentActorProps: (Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any) => Props = (_, _, _, _, _, _, _, _, _, _, _) => TestActor.props(new LinkedBlockingDeque())
258
140259 def deploymentManager(): TestActorRef[DeploymentManager] = TestActorRef (
141 DeploymentManager.props(taskTracker, taskKillService, launchQueue, scheduler, storage, hcManager, eventBus, readinessCheckExecutor)
260 DeploymentManager.props(
261 taskTracker,
262 taskKillService,
263 launchQueue,
264 scheduler,
265 storage,
266 hcManager,
267 eventBus,
268 readinessCheckExecutor,
269 deploymentRepo,
270 deploymentActorProps)
142271 )
143
272 deploymentRepo.store(any[DeploymentPlan]) returns Future.successful(Done)
273 deploymentRepo.delete(any[String]) returns Future.successful(Done)
274 deploymentRepo.all() returns Source.empty
144275 }
145276}
View Options

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

11package mesosphere.marathon
22package upgrade
33
44import akka.actor.{ Actor, Props }
55import akka.testkit.TestActorRef
66import mesosphere.marathon.core.condition.Condition
7import mesosphere.marathon.core.condition.Condition.Running
78import mesosphere.marathon.core.event._
89import mesosphere.marathon.core.health.{ MarathonHttpHealthCheck, PortReference }
9import mesosphere.marathon.core.condition.Condition.Running
1010import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
1111import mesosphere.marathon.core.launchqueue.LaunchQueue
1212import mesosphere.marathon.core.readiness.{ ReadinessCheck, ReadinessCheckExecutor, ReadinessCheckResult }
1313import mesosphere.marathon.core.task.tracker.InstanceTracker
1414import mesosphere.marathon.core.task.{ KillServiceMock, Task }
1515import mesosphere.marathon.state.PathId._
1616import mesosphere.marathon.state._
1717import mesosphere.marathon.test.MarathonActorSupport
18import org.apache.mesos.SchedulerDriver
18import org.mockito.Matchers.any
1919import org.mockito.Mockito
2020import org.mockito.Mockito._
21import org.mockito.Matchers.any
2221import org.scalatest.concurrent.Eventually
2322import org.scalatest.mockito.MockitoSugar
2423import org.scalatest.{ BeforeAndAfter, FunSuiteLike, Matchers }
2524import rx.lang.scala.Observable
2625
2726import scala.collection.immutable.Seq
2827import scala.concurrent.duration._
2928import scala.concurrent.{ Await, Promise }
▲ Show 20 Lines • Show All 549 Lines • ▼ Show 20 Line(s)
579578 eventually { f.killService.numKilled should be(1) }
580579
581580 Await.result(promise.future, 5.seconds)
582581 }
583582
584583 class Fixture {
585584 val deploymentsManager: TestActorRef[Actor] = TestActorRef[Actor](Props.empty)
586585 val deploymentStatus = DeploymentStatus(DeploymentPlan.empty, DeploymentStep(Seq.empty))
587 private[this] val driver = mock[SchedulerDriver]
588586 val killService = new KillServiceMock(system)
589587 val queue: LaunchQueue = mock[LaunchQueue]
590588 val tracker: InstanceTracker = mock[InstanceTracker]
591589 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
592590 val hostName = "host.some"
593591 val hostPorts = Seq(123)
594592
595593 def runningInstance(app: AppDefinition): Instance = {
596594 TestInstanceBuilder.newBuilder(app.id, version = app.version)
597595 .addTaskWithBuilder().taskRunning().withNetworkInfo(hostName = Some(hostName), hostPorts = hostPorts).build()
598596 .getInstance()
599597 }
600598
601599 def instanceChanged(app: AppDefinition, condition: Condition): InstanceChanged = {
602600 val instanceId = Instance.Id.forRunSpec(app.id)
603601 val instance: Instance = mock[Instance]
604602 when(instance.instanceId).thenReturn(instanceId)
605603 InstanceChanged(instanceId, app.version, app.id, condition, instance)
606604 }
607605
608606 def healthChanged(app: AppDefinition, healthy: Boolean): InstanceHealthChanged = {
609607 InstanceHealthChanged(Instance.Id.forRunSpec(app.id), app.version, app.id, healthy = Some(healthy))
610608 }
611609 def replaceActor(app: AppDefinition, promise: Promise[Unit]): TestActorRef[TaskReplaceActor] = TestActorRef(
612 TaskReplaceActor.props(deploymentsManager, deploymentStatus, driver, killService, queue,
610 TaskReplaceActor.props(deploymentsManager, deploymentStatus, killService, queue,
613611 tracker, system.eventStream, readinessCheckExecutor, app, promise)
614612 )
615613 }
616614}
View Options

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

11package mesosphere.marathon.upgrade
22
33import akka.testkit.{ TestActorRef, TestProbe }
44import com.codahale.metrics.MetricRegistry
5import mesosphere.{ IntegrationTag, Unstable }
65import mesosphere.marathon.core.condition.Condition
6import mesosphere.marathon.core.condition.Condition.{ Failed, Running }
77import mesosphere.marathon.core.event.{ DeploymentStatus, _ }
88import mesosphere.marathon.core.health.MesosCommandHealthCheck
9import mesosphere.marathon.core.condition.Condition.{ Failed, Running }
109import mesosphere.marathon.core.instance.update.InstanceUpdateOperation
1110import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
1211import mesosphere.marathon.core.launcher.impl.LaunchQueueTestHelper
1312import mesosphere.marathon.core.launchqueue.LaunchQueue
1413import mesosphere.marathon.core.leadership.AlwaysElectedLeadershipModule
1514import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1615import mesosphere.marathon.core.task.tracker.{ InstanceCreationHandler, InstanceTracker }
1716import mesosphere.marathon.metrics.Metrics
1817import mesosphere.marathon.state.PathId._
1918import mesosphere.marathon.state.{ AppDefinition, Command, Timestamp }
2019import mesosphere.marathon.storage.repository.legacy.store.InMemoryStore
2120import mesosphere.marathon.test.{ MarathonActorSupport, MarathonTestHelper, Mockito }
2221import mesosphere.marathon.{ SchedulerActions, TaskUpgradeCanceledException }
23import org.apache.mesos.SchedulerDriver
22import mesosphere.{ IntegrationTag, Unstable }
2423import org.mockito.Mockito.{ spy, when }
2524import org.scalatest.concurrent.ScalaFutures
2625import org.scalatest.{ BeforeAndAfter, FunSuiteLike, Matchers }
2726
2827import scala.concurrent.duration._
2928import scala.concurrent.{ Await, Promise }
3029
3130class TaskStartActorTest
▲ Show 20 Lines • Show All 217 Lines • ▼ Show 20 Line(s)
249248
250249 noMoreInteractions(f.launchQueue)
251250
252251 expectTerminated(ref)
253252 }
254253
255254 class Fixture {
256255
257 val driver: SchedulerDriver = mock[SchedulerDriver]
258256 val scheduler: SchedulerActions = mock[SchedulerActions]
259257 val launchQueue: LaunchQueue = mock[LaunchQueue]
260258 val metrics: Metrics = new Metrics(new MetricRegistry)
261259 val leadershipModule = AlwaysElectedLeadershipModule.forActorSystem(system)
262260 val taskTrackerModule = MarathonTestHelper.createTaskTrackerModule(
263261 leadershipModule, store = new InMemoryStore, metrics = metrics)
264262 val taskTracker: InstanceTracker = spy(taskTrackerModule.instanceTracker)
265263 val taskCreationHandler: InstanceCreationHandler = taskTrackerModule.instanceCreationHandler
266264 val deploymentManager = TestProbe()
267265 val status: DeploymentStatus = mock[DeploymentStatus]
268266 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
269267
270268 def instanceChange(app: AppDefinition, id: Instance.Id, condition: Condition): InstanceChanged = {
271269 val instance: Instance = mock[Instance]
272270 instance.instanceId returns id
273271 InstanceChanged(id, app.version, app.id, condition, instance)
274272 }
275273
276274 def healthChange(app: AppDefinition, id: Instance.Id, healthy: Boolean): InstanceHealthChanged = {
277275 InstanceHealthChanged(id, app.version, app.id, Some(healthy))
278276 }
279277
280278 def startActor(app: AppDefinition, scaleTo: Int, promise: Promise[Unit]): TestActorRef[TaskStartActor] = TestActorRef(TaskStartActor.props(
281 deploymentManager.ref, status, driver, scheduler, launchQueue, taskTracker, system.eventStream, readinessCheckExecutor, app, scaleTo, promise
279 deploymentManager.ref, status, scheduler, launchQueue, taskTracker, system.eventStream, readinessCheckExecutor, app, scaleTo, promise
282280 ))
283281 }
284282}