HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D529

Don't destroy persistent volumes when killing unreachable resident tasks
ClosedAll Users
Actions

Authored by timcharper on Feb 20 2017, 4:04 PM.

Details

Reviewers
unterstein
jenkins
• jasongilanfarr
meichstedt
Commits
rMARATHONfec27295be9e: Don't destroy persistent volumes when killing unreachable resident tasks
rMARATHON1455fdd9e86c: Don't destroy persistent volumes when killing unreachable resident tasks
rMARATHONc6431e2e7169: Don't destroy persistent volumes when killing unreachable resident tasks
rMARATHONef6bd779e8d8: Refine logic on how to treat instances that shall be killed.
rMARATHONeeed52b43305: Refine logic on how to treat instances that shall be killed.
rMARATHON5f34abd85438: Don't destroy persistent volumes when killing unreachable resident tasks
Summary

Killing an unreachable resident task will do nothing, rather than
destroy the reservations.

Fixes #5207

Also-By: tharper@mesosphere.com

Test Plan

sbt test

Diff Detail

Repository
rMARATHON marathon
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.
There are a very large number of changes, so older changes are hidden. Show Older Changes
timcharper added a comment.Feb 28 2017, 2:02 AM

update description

timcharper updated this revision to Diff 2039.Feb 28 2017, 2:07 AM

revert missing changes

jenkins requested changes to this revision.Feb 28 2017, 2:08 AM
This revision now requires changes to proceed.Feb 28 2017, 2:08 AM
jenkins accepted this revision.Feb 28 2017, 2:36 AM

LGTM

timcharper marked an inline comment as done.Feb 28 2017, 5:14 PM
meichstedt accepted this revision.Mar 1 2017, 7:04 PM

One minor comment, otherwise LGTM

src/main/scala/mesosphere/marathon/core/task/termination/impl/KillActionResolver.scala
52 ↗(On Diff #2039)

Thinking out loud: should we differentiate between these two cases? The comment above says

// An instance will be expunged once all tasks are terminal. Therefore, this case is
// highly unlikely. Should it ever occur, this will still expunge the instance to clean up.

isUnkillable -> Noop
allTerminal -> expunge
?

Otherwise the comment should be adjusted to reflect the reasoning and behavior.

This revision is now accepted and ready to land.Mar 1 2017, 7:04 PM
timcharper updated this revision to Diff 2074.Mar 1 2017, 9:21 PM

update comment

jenkins requested changes to this revision.Mar 1 2017, 9:21 PM
This revision now requires changes to proceed.Mar 1 2017, 9:21 PM
timcharper marked an inline comment as done.Mar 1 2017, 9:22 PM
timcharper added inline comments.
src/main/scala/mesosphere/marathon/core/task/termination/impl/KillActionResolver.scala
52 ↗(On Diff #2039)

good catch.

I think the behavior is correct (accounting for the decision to not support kill-while-unreachable for resident tasks). I've update the comment

jenkins accepted this revision.Mar 1 2017, 9:39 PM

Thanks.

This revision is now accepted and ready to land.Mar 1 2017, 9:39 PM
• jasongilanfarr accepted this revision.Mar 1 2017, 9:44 PM

Mostly minor/stylistic stuff

src/main/scala/mesosphere/marathon/core/task/termination/KillReason.scala
40 ↗(On Diff #2074)

By the user? (If so, please document)

src/main/scala/mesosphere/marathon/core/task/termination/impl/KillActionResolver.scala
10 ↗(On Diff #2074)

Could be in KillAction's companion via an apply()

src/main/scala/mesosphere/marathon/core/task/termination/impl/KillServiceActor.scala
142

may be better as Done () looks weird.

src/main/scala/mesosphere/marathon/core/task/termination/impl/ToKill.scala
18 ↗(On Diff #2074)

Does this need to stand alone?

meichstedt added a comment.Mar 2 2017, 3:06 PM

minor changes requested still approved.

  • sbt doc will fail because there's an invalid ref (see comment)
  • there's an unused import due to double package notation. I'll try to change my scala file template to add double package notation for mesosphere.marathon by default.
src/main/scala/mesosphere/marathon/core/task/termination/KillReason.scala
40 ↗(On Diff #2074)

The need for this vale has been removed, please delete it.

src/main/scala/mesosphere/marathon/core/task/termination/impl/KillAction.scala
30

The mentioned value does not exist anymore. Please delete the note.

src/main/scala/mesosphere/marathon/core/task/termination/impl/KillActionResolver.scala
10 ↗(On Diff #2074)

It's a question of preference and I'm not sure about which preference we share within the team. I am not a big fan of files containing more than one definition, especially when it mixes up contracts and logic. The combination of both, for me personally, is visual clutter. If we apply this approach generally, everything in termination.* will end up in the same file basically. In my opinion, packages act as a better container for classes, objects, and types.

src/main/scala/mesosphere/marathon/core/task/termination/impl/KillServiceActor.scala
142

agreed.

src/main/scala/mesosphere/marathon/core/task/termination/impl/ToKill.scala
4 ↗(On Diff #2074)

unused import

18 ↗(On Diff #2074)

I've extracted the class because it's now used in KillServiceActor and KillActionResolver. I don't feel too strong about this, but if we move this, we can move everything into 1 file. Not sure that's better.

I understand though that in one case class per file seems picky.

timcharper marked 12 inline comments as done.Mar 2 2017, 4:36 PM
timcharper added inline comments.
src/main/scala/mesosphere/marathon/core/task/termination/KillReason.scala
40 ↗(On Diff #2074)

Great catch.

src/main/scala/mesosphere/marathon/core/task/termination/impl/KillActionResolver.scala
10 ↗(On Diff #2074)

I'm inclined to agree with Matthias; if KillAction were only used by one class, then putting it in the companion object is optimal. However, it's used by KillActionResolved, and, putting KillActionResolver logic in the actor starts to make the file rather big, pushing it past the 500 line count. ¯\_(ツ)_/¯

src/main/scala/mesosphere/marathon/core/task/termination/impl/KillServiceActor.scala
142

gonna push back slightly. It seems that () isn't inherently weird... it's an empty tuple. Unit.

Future[Done] is much better than Future[Unit] because of the pitfalls for accidentally tossing an enclosed Future[Unit]. However, in the case of Unit, our linter is too aggressive and will expect that the caller assign it.

src/main/scala/mesosphere/marathon/core/task/termination/impl/ToKill.scala
18 ↗(On Diff #2074)

No.

• jasongilanfarr requested changes to this revision.Mar 2 2017, 4:40 PM
• jasongilanfarr added inline comments.
src/main/scala/mesosphere/marathon/core/task/termination/impl/KillActionResolver.scala
10 ↗(On Diff #2074)

If this was about the other class that stood alone, I'd agree, but it's not. This is "how do I get a KillAction" which KillAction _should_ do/be responsible for. I'm not saying it should be in the actor.

This revision now requires changes to proceed.Mar 2 2017, 4:40 PM
timcharper updated this revision to Diff 2085.Mar 2 2017, 4:40 PM
timcharper marked 2 inline comments as done.

address comments

jenkins requested changes to this revision.Mar 2 2017, 4:40 PM
This revision now requires changes to proceed.Mar 2 2017, 4:40 PM
timcharper updated this revision to Diff 2087.Mar 2 2017, 4:48 PM

use inside; wrap long lines

jenkins requested changes to this revision.Mar 2 2017, 4:57 PM
This revision now requires changes to proceed.Mar 2 2017, 4:57 PM
jenkins accepted this revision.Mar 2 2017, 4:59 PM

Ship it!

timcharper updated this revision to Diff 2088.Mar 2 2017, 5:13 PM

mark ActorOfferMatcherTest unstable

jenkins requested changes to this revision.Mar 2 2017, 5:13 PM
This revision now requires changes to proceed.Mar 2 2017, 5:13 PM
timcharper updated this revision to Diff 2091.Mar 2 2017, 5:35 PM

inline KillActionResolver logic to KillAction

  • remove usage of ToKill outside of actor
  • inline ToKill with actor companion object
timcharper marked an inline comment as done.Mar 2 2017, 5:35 PM
timcharper added inline comments.
src/main/scala/mesosphere/marathon/core/task/termination/impl/KillActionResolver.scala
10 ↗(On Diff #2074)

agreed, I'll move it

jenkins requested changes to this revision.Mar 2 2017, 5:35 PM
This revision now requires changes to proceed.Mar 2 2017, 5:35 PM
jenkins accepted this revision.Mar 2 2017, 5:53 PM

LGTM

• jasongilanfarr accepted this revision.Mar 2 2017, 11:26 PM
• jasongilanfarr added inline comments.
src/main/scala/mesosphere/marathon/core/task/termination/impl/KillAction.scala
64

minor, but apply() seems useful.

src/test/scala/mesosphere/marathon/core/task/termination/impl/KillActionTest.scala
32–40

Remove the formatting pretty please.

This revision is now accepted and ready to land.Mar 2 2017, 11:26 PM
timcharper updated this revision to Diff 2097.Mar 3 2017, 1:53 AM

rebase. make Jason happy.

jenkins requested changes to this revision.Mar 3 2017, 1:58 AM
This revision now requires changes to proceed.Mar 3 2017, 1:58 AM
jenkins accepted this revision.Mar 3 2017, 2:23 AM

(ﾉ◕ヮ◕)ﾉ*: It's Green :*ヽ(◕ヮ◕ヽ)

This revision is now accepted and ready to land.Mar 3 2017, 2:23 AM
Closed by commit rMARATHONc6431e2e7169: Don't destroy persistent volumes when killing unreachable resident tasks (authored by meichstedt, committed by timcharper). · Explain WhyMar 3 2017, 5:23 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents

Diff 2098

View Options

src/main/scala/mesosphere/marathon/core/instance/Instance.scala

Show First 20 Lines • Show All 43 Lines • ▼ Show 20 Line(s)
4444 def isStarting: Boolean = state.condition == Condition.Starting
4545 def isUnreachable: Boolean = state.condition == Condition.Unreachable
4646 def isUnreachableInactive: Boolean = state.condition == Condition.UnreachableInactive
4747 def isGone: Boolean = state.condition == Condition.Gone
4848 def isUnknown: Boolean = state.condition == Condition.Unknown
4949 def isDropped: Boolean = state.condition == Condition.Dropped
5050 def isTerminated: Boolean = state.condition.isTerminal
5151 def isActive: Boolean = state.condition.isActive
52 def hasReservation =
jdefUnsubmitted
Done

taskMap.exists((id,task) => task.reservationWithVolumes.isDefined) ?

53 tasksMap.values.exists {
meichstedtUnsubmitted
Not Done

taskMap.exists((id,task) => ...? map.values creates a new collection on the fly which is not necessary in this case. Evaluating task.reservationWithVolumes.isDefined was used before to get around type checking, which is cumbersome when mapping over a tuple.

timcharperAuthorUnsubmitted
Not Done

I think type checking is okay. All reserved tasks extend this common sealed trait.

map.values, as I understand it, does not create a new collection. (what are you seeing that has led you to believe otherwise?)

My understanding is that it creates a new Iterable, which is basically an iterator generator that refers to the Map iterator generator, and transforms the iterator to select the second tuple item.

meichstedtUnsubmitted
Not Done

You're right. I must have confused it with filter, or it returned a new coll in a previous scala version? No idea, it's fine.

54 case _: Task.ReservedTask => true
55 case _ => false
56 }
5257
5358 override def mergeFromProto(message: Protos.Json): Instance = {
5459 Json.parse(message.getJson).as[Instance]
5560 }
5661 override def mergeFromProto(bytes: Array[Byte]): Instance = {
5762 mergeFromProto(Protos.Json.parseFrom(bytes))
5863 }
5964 override def toProto: Protos.Json = {
▲ Show 20 Lines • Show All 317 Lines • Show Last 20 Lines
View Options

src/main/scala/mesosphere/marathon/core/task/termination/impl/KillAction.scala

  • This file was added.
1package mesosphere.marathon
2package core.task.termination.impl
3
4import com.typesafe.scalalogging.StrictLogging
5import mesosphere.marathon.core.condition.Condition
6import mesosphere.marathon.core.instance.Instance
7import mesosphere.marathon.core.task.Task
8
9/**
10 * Possible actions that can be chosen in order to `kill` a given instance.
11 * Depending on the instance's state this can be one of
12 * - [[KillAction.ExpungeFromState]]
13 * - [[KillAction.Noop]]
14 * - [[KillAction.IssueKillRequest]]
15 */
16private[termination] sealed trait KillAction
17
18private[termination] object KillAction extends StrictLogging {
19 /**
20 * Any normal, reachable and stateless instance will simply be killed via the scheduler driver.
21 */
22 case object IssueKillRequest extends KillAction
23
24 /**
25 * Do nothing. This is currently what we do for unreachable tasks with reservations. See #5261
jdefUnsubmitted
Done

or .. ForceReschedule?

timcharperAuthorUnsubmitted
Done

It may not be ForceReschedule. For example, we could be killing an Unreachable resident task while over provisioned.

26 */
27 case object Noop extends KillAction
28
29 /**
30 * In case of an instance being Unreachable, killing the related Mesos task is impossible.
meichstedtUnsubmitted
Done

The mentioned value does not exist anymore. Please delete the note.

31 * In order to get rid of the instance, processing this action expunges the metadata from
32 * state. If the instance is reported to be non-terminal in the future, it will be killed.
jdefUnsubmitted
Done

does this imply that we'll get a state transition error if ExpungeFromState is used in combination with a task that has reservations?

timcharperAuthorUnsubmitted
Done

No; KillActionResolver is responsible for returning the proper KillAction

33 */
34 case object ExpungeFromState extends KillAction
35
36 /* returns whether or not we can expect the task to report a terminal state after sending a kill signal */
37 private val wontRespondToKill: Condition => Boolean = {
38 import Condition._
39 Set(
40 Unknown, Unreachable, UnreachableInactive,
41 // TODO: it should be safe to remove these from this list, because
42 // 1) all taskId's should be removed at this point, because Gone & Dropped are terminal.
43 // 2) Killing a Gone / Dropped task will cause it to be in a terminal state.
44 // 3) Killing a Gone / Dropped task may result in no status change at all.
45 // 4) Either way, we end up in a terminal state.
46 // However, we didn't want to risk changing behavior in a point release. So they remain here.
47 Dropped, Gone
48 )
49 }
50
51 /**
52 * Computes the [[KillAction]] based on the instance's state.
53 *
54 * if the instance can't be reached, issuing a kill request won't cause the instance to progress towards a terminal
55 * state; Mesos will simply re-send the current state. Our current behavior, for ephemeral, is to simply delete any
56 * knowledge that the instance might be running, such that if it is reported by Mesos later we will kill it. (that
57 * could be improved).
58 *
59 * If the instance is lost _and_ has reservations, we do nothing.
60 *
61 * any other case -> issue a kill request
62 */
63 def apply(instanceId: Instance.Id, taskIds: Iterable[Task.Id], knownInstance: Option[Instance]): KillAction = {
64 val hasReservations = knownInstance.fold(false)(_.hasReservation)
jasongilanfarrUnsubmitted
Not Done

minor, but apply() seems useful.

65
66 // TODO(PODS): align this with other Terminal/Unreachable/whatever extractors
67 val maybeCondition = knownInstance.map(_.state.condition)
68 val isUnkillable = maybeCondition.fold(false)(wontRespondToKill)
69
70 // Ephemeral instances are expunged once all tasks are terminal, it's unlikely for this to be true for them.
71 // Resident tasks, however, could be in this state if scaled down, or, if kill is attempted between recovery.
72 val allTerminal: Boolean = taskIds.isEmpty
73
74 if (isUnkillable || allTerminal) {
75 val msg = if (isUnkillable)
76 s"it is ${maybeCondition.fold("unknown")(_.toString)}"
77 else
78 "none of its tasks are running"
79 if (hasReservations) {
80 logger.info(
81 s"Ignoring kill request for ${instanceId}; killing it while ${msg} is unsupported")
82 KillAction.Noop
83 } else {
84 logger.warn(s"Expunging ${instanceId} from state because ${msg}")
85 // we will eventually be notified of a taskStatusUpdate after the instance has been expunged
86 KillAction.ExpungeFromState
87 }
88 } else {
89 val knownOrNot = if (knownInstance.isDefined) "known" else "unknown"
90 logger.warn("Killing {} {} of instance {}", knownOrNot, taskIds.mkString(","), instanceId)
91 KillAction.IssueKillRequest
92 }
93 }
94
95}
View Options

src/main/scala/mesosphere/marathon/core/task/termination/impl/KillServiceActor.scala

11package mesosphere.marathon
22package core.task.termination.impl
33
44import akka.Done
55import akka.actor.{ Actor, ActorLogging, Cancellable, Props }
66import akka.stream.ActorMaterializer
77import mesosphere.marathon.core.base.Clock
88import mesosphere.marathon.core.task.termination.KillConfig
9import mesosphere.marathon.state.Timestamp
109import mesosphere.marathon.core.task.Task
1110import mesosphere.marathon.core.task.tracker.TaskStateOpProcessor
1211import mesosphere.marathon.core.event.{ InstanceChanged, UnknownInstanceTerminated }
1312import mesosphere.marathon.core.instance.Instance
1413import mesosphere.marathon.core.instance.update.InstanceUpdateOperation
1514import mesosphere.marathon.core.task.termination.InstanceChangedPredicates.considerTerminal
1615import mesosphere.marathon.core.task.Task.Id
1716import mesosphere.marathon.stream.Sink
1817
1918import scala.collection.mutable
19import mesosphere.marathon.state.Timestamp
2020import scala.concurrent.{ Future, Promise }
2121
2222/**
2323 * An actor that handles killing instances in chunks and depending on the instance state.
2424 * Lost instances will simply be expunged from state, while active instances will be killed
2525 * via the scheduler driver. There is be a maximum number of kills in flight, and
2626 * the service will only issue more kills when instances are reported terminal.
2727 *
2828 * If a kill is not acknowledged with a terminal status update within a configurable
2929 * time window, the kill is retried a configurable number of times. If the maximum
3030 * number of retries is exceeded, the instance will be expunged from state similar to a
3131 * lost instance.
3232 *
33 * For each kill request, a child [[InstanceKillProgressActor]] will be spawned, which
33 * For each kill request, a [[KillStreamWatcher]] will be created, which
3434 * is supposed to watch the progress and complete a given promise when all watched
3535 * instances are reportedly terminal.
3636 *
3737 * For pods started via the default executor, it is sufficient to kill 1 task of the group,
3838 * which will cause all tasks to be killed
3939 *
4040 * See [[KillConfig]] for configuration options.
4141 */
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Line(s)
134134
135135 if (inFlight.isEmpty) {
136136 retryTimer.cancel()
137137 } else {
138138 retryTimer.setup()
139139 }
140140 }
141141
142142 def processKill(toKill: ToKill): Unit = {
jasongilanfarrUnsubmitted
Done

may be better as Done () looks weird.

meichstedtUnsubmitted
Done

agreed.

timcharperAuthorUnsubmitted
Done

gonna push back slightly. It seems that () isn't inherently weird... it's an empty tuple. Unit.

Future[Done] is much better than Future[Unit] because of the pitfalls for accidentally tossing an enclosed Future[Unit]. However, in the case of Unit, our linter is too aggressive and will expect that the caller assign it.

143143 val instanceId = toKill.instanceId
144144 val taskIds = toKill.taskIdsToKill
145145
146 // TODO(PODS): align this with other Terminal/Unreachable/whatever extractors
147 val isLost: Boolean = toKill.maybeInstance.fold(false) { instance =>
148 instance.isGone || instance.isUnknown || instance.isDropped || instance.isUnreachable || instance.isUnreachableInactive
149 }
146 KillAction(toKill.instanceId, toKill.taskIdsToKill, toKill.maybeInstance) match {
147 case KillAction.Noop =>
148 ()
150149
151 // An instance will be expunged once all tasks are terminal. Therefore, this case is
152 // highly unlikely. Should it ever occur, this will still expunge the instance to clean up.
153 val allTerminal: Boolean = taskIds.isEmpty
154
155 if (isLost || allTerminal) {
156 val msg = if (isLost) "it is lost" else "all its tasks are terminal"
157 log.warning("Expunging {} from state because {}", instanceId, msg)
158 // we will eventually be notified of a taskStatusUpdate after the instance has been expunged
159 stateOpProcessor.process(InstanceUpdateOperation.ForceExpunge(toKill.instanceId))
160 } else {
161 val knownOrNot = if (toKill.maybeInstance.isDefined) "known" else "unknown"
162 log.warning("Killing {} {} of instance {}", knownOrNot, taskIds.mkString(","), instanceId)
150 case KillAction.IssueKillRequest =>
163151 driverHolder.driver.foreach { driver =>
meichstedtUnsubmitted
Done

I know we had this before but wouldn't InstanceUpdateOperation .TransitionToReserved be the better term? TransitionToReserved clearly transports that the instance should be transitioned to another state while Suspend sounds like actively suspending something here, in addition to @jdef arguing that Suspend might be understood as _suspended until resumed_. Not being pedantic, but having 2 terms which are probably not 100% clear is worse that having just 1.

timcharperAuthorUnsubmitted
Done

That is an interesting point. I'll push TransitionToReserved through.

164152 taskIds.map(_.mesosTaskId).foreach(driver.killTask)
165153 }
154 val attempts = inFlight.get(toKill.instanceId).fold(1)(_.attempts + 1)
155 inFlight.update(
156 toKill.instanceId, ToKill(instanceId, taskIds, toKill.maybeInstance, attempts, issued = clock.now()))
157
158 case KillAction.ExpungeFromState =>
159 stateOpProcessor.process(InstanceUpdateOperation.ForceExpunge(toKill.instanceId))
166160 }
167161
168 val attempts = inFlight.get(toKill.instanceId).fold(1)(_.attempts + 1)
169 inFlight.update(toKill.instanceId, ToKill(instanceId, taskIds, toKill.maybeInstance, attempts, issued = clock.now()))
170162 instancesToKill.remove(instanceId)
171163 }
172164
173165 def handleTerminal(instanceId: Instance.Id): Unit = {
174166 instancesToKill.remove(instanceId)
175167 inFlight.remove(instanceId)
176168 log.debug("{} is terminal. ({} kills queued, {} in flight)", instanceId, instancesToKill.size, inFlight.size)
177169 processKills()
Show All 16 Lines
194186
195187 sealed trait Request extends InternalRequest
196188 case class KillInstances(instances: Seq[Instance], promise: Promise[Done]) extends Request
197189 case class KillUnknownTaskById(taskId: Task.Id) extends Request
198190
199191 sealed trait InternalRequest
200192 case object Retry extends InternalRequest
201193
194 def props(
195 driverHolder: MarathonSchedulerDriverHolder,
196 stateOpProcessor: TaskStateOpProcessor,
197 config: KillConfig,
198 clock: Clock): Props = Props(
199 new KillServiceActor(driverHolder, stateOpProcessor, config, clock))
200
202201 /**
203202 * Metadata used to track which instances to kill and how many attempts have been made
203 *
204204 * @param instanceId id of the instance to kill
205205 * @param taskIdsToKill ids of the tasks to kill
206206 * @param maybeInstance the instance, if available
207207 * @param attempts the number of kill attempts
208208 * @param issued the time of the last issued kill request
209209 */
210210 case class ToKill(
211211 instanceId: Instance.Id,
212212 taskIdsToKill: Seq[Task.Id],
213213 maybeInstance: Option[Instance],
214214 attempts: Int,
215215 issued: Timestamp = Timestamp.zero)
216
217 def props(
218 driverHolder: MarathonSchedulerDriverHolder,
219 stateOpProcessor: TaskStateOpProcessor,
220 config: KillConfig,
221 clock: Clock): Props = Props(
222 new KillServiceActor(driverHolder, stateOpProcessor, config, clock))
223216}
224217
225218/**
226219 * Wraps a timer into an interface that hides internal mutable state behind simple setup and cancel methods
227220 */
228221private[this] trait RetryTimer {
229222 private[this] var retryTimer: Option[Cancellable] = None
230223
Show All 21 Lines
View Options

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

Show All 22 Lines
2323 stateOpResolver: InstanceUpdateOpResolver,
2424 config: InstanceTrackerConfig) extends InstanceOpProcessor {
2525 import InstanceOpProcessor._
2626
2727 private[this] val log = LoggerFactory.getLogger(getClass)
2828
2929 override def process(op: Operation)(implicit ec: ExecutionContext): Future[Unit] = {
3030 val stateChange = stateOpResolver.resolve(op.op)
31
3132 stateChange.flatMap {
3233 case change: InstanceUpdateEffect.Expunge =>
3334 // Used for task termination or as a result from a UpdateStatus action.
3435 // The expunge is propagated to the instanceTracker which informs the sender about the success (see Ack).
3536 repository.delete(change.instance.instanceId).map { _ => InstanceTrackerActor.Ack(op.sender, change) }
3637 .recoverWith(tryToRecover(op)(
3738 expectedState = None, oldState = Some(change.instance), change.events))
3839 .flatMap { ack: InstanceTrackerActor.Ack => notifyTaskTrackerActor(ack) }
▲ Show 20 Lines • Show All 80 Lines • Show Last 20 Lines
View Options

src/test/scala/mesosphere/marathon/core/instance/TestTaskBuilder.scala

Show First 20 Lines • Show All 53 Lines • ▼ Show 20 Line(s)
5454 this.copy(task = Some(TestTaskBuilder.Helper.residentReservedTask(instance.instanceId.runSpecId, taskReservationState, Seq.empty[Task.LocalVolumeId]: _*).copy(taskId = Task.Id.forInstanceId(instance.instanceId, None))))
5555 }
5656
5757 def taskResidentLaunched(localVolumeIds: Task.LocalVolumeId*) = {
5858 val instance = instanceBuilder.getInstance()
5959 this.copy(task = Some(TestTaskBuilder.Helper.residentLaunchedTask(instance.instanceId.runSpecId, localVolumeIds: _*).copy(taskId = Task.Id.forInstanceId(instance.instanceId, None))))
6060 }
6161
62 def taskResidentUnreachable(localVolumeIds: Task.LocalVolumeId*) = {
63 val instance = instanceBuilder.getInstance()
64 this.copy(task = Some(TestTaskBuilder.Helper.residentUnreachableTask(instance.instanceId.runSpecId, localVolumeIds: _*).copy(taskId = Task.Id.forInstanceId(instance.instanceId, None))))
65 }
66
6267 def taskRunning(containerName: Option[String] = None, stagedAt: Timestamp = now, startedAt: Timestamp = now) = {
6368 val instance = instanceBuilder.getInstance()
6469 this.copy(task = Some(TestTaskBuilder.Helper.runningTask(
6570 Task.Id.forInstanceId(instance.instanceId, maybeMesosContainerByName(containerName)),
6671 instance.runSpecVersion, stagedAt = stagedAt.millis, startedAt = startedAt.millis)))
6772 }
6873
6974 /**
▲ Show 20 Lines • Show All 214 Lines • ▼ Show 20 Line(s)
284289 Task.LaunchedOnReservation(
285290 taskId = Task.Id.forRunSpec(appId),
286291 runSpecVersion = now,
287292 status = Task.Status(
288293 stagedAt = now,
289294 startedAt = Some(now),
290295 mesosStatus = None,
291296 condition = Condition.Running,
297 networkInfo = NetworkInfoPlaceholder()
298 ),
299 reservation = Task.Reservation(localVolumeIds.to[Seq], Task.Reservation.State.Launched))
300 }
301
302 def residentUnreachableTask(appId: PathId, localVolumeIds: Task.LocalVolumeId*) = {
303 val now = Timestamp.now()
304 Task.LaunchedOnReservation(
305 taskId = Task.Id.forRunSpec(appId),
306 runSpecVersion = now,
307 status = Task.Status(
308 stagedAt = now,
309 startedAt = Some(now),
310 mesosStatus = None,
311 condition = Condition.Unreachable,
292312 networkInfo = NetworkInfoPlaceholder()
293313 ),
294314 reservation = Task.Reservation(localVolumeIds.to[Seq], Task.Reservation.State.Launched))
295315 }
296316
297317 def startingTaskForApp(instanceId: Instance.Id, appVersion: Timestamp = Timestamp(1), stagedAt: Long = 2, container: Option[MesosContainer] = None): Task.LaunchedEphemeral =
298318 startingTask(
299319 Task.Id.forInstanceId(instanceId, container),
▲ Show 20 Lines • Show All 123 Lines • Show Last 20 Lines
View Options

src/test/scala/mesosphere/marathon/core/instance/update/InstanceUpdateOpResolverTest.scala

11package mesosphere.marathon
22package core.instance.update
33
44import mesosphere.UnitTest
55import mesosphere.marathon.core.base.ConstantClock
66import mesosphere.marathon.core.condition.Condition
77import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
88import mesosphere.marathon.core.instance.TestInstanceBuilder._
99import mesosphere.marathon.core.task.bus.{ MesosTaskStatusTestHelper, TaskStatusUpdateTestHelper }
1010import mesosphere.marathon.core.task.state.{ NetworkInfoPlaceholder, TaskConditionMapping }
1111import mesosphere.marathon.core.task.tracker.InstanceTracker
1212import mesosphere.marathon.core.task.{ Task, TaskCondition }
1313import mesosphere.marathon.state.{ PathId, Timestamp }
1414import org.apache.mesos
15import org.scalatest.Inside
1516
1617import scala.collection.immutable.Seq
1718import scala.concurrent.Future
1819
1920/**
2021 * Some specialized tests for statusUpdate action resolving.
2122 *
2223 * More tests are in [[mesosphere.marathon.tasks.InstanceTrackerImplTest]]
2324 */
24class InstanceUpdateOpResolverTest extends UnitTest {
25class InstanceUpdateOpResolverTest extends UnitTest with Inside {
2526
2627 import scala.concurrent.ExecutionContext.Implicits.global
2728
2829 "InstanceUpdateOpResolver" should {
2930 "ForceExpunge for an unknown task" in new Fixture {
3031 instanceTracker.instance(notExistingInstanceId) returns Future.successful(None)
3132 val stateChange = updateOpResolver.resolve(InstanceUpdateOperation.ForceExpunge(notExistingInstanceId)).futureValue
3233 When("call taskTracker.task")
▲ Show 20 Lines • Show All 48 Lines • ▼ Show 20 Line(s)
8182 instanceTracker.instance(existingInstance.instanceId) returns Future.successful(Some(existingInstance))
8283 val operation = TaskStatusUpdateTestHelper.lost(reason, existingInstance).operation
8384 val effect = updateOpResolver.resolve(operation).futureValue
8485
8586 When("call taskTracker.task")
8687 verify(instanceTracker).instance(existingInstance.instanceId)
8788
8889 Then("result in an Update with the correct status")
89 effect shouldBe a[InstanceUpdateEffect.Update]
90
91 val update: InstanceUpdateEffect.Update = effect.asInstanceOf[InstanceUpdateEffect.Update]
92 update.instance.isUnreachable should be(true)
93
90 inside(effect) {
91 case update: InstanceUpdateEffect.Update =>
92 update.instance.isUnreachable shouldBe true
93 }
9494 verifyNoMoreInteractions()
9595 }
9696 }
9797
9898 for (
9999 reason <- TaskConditionMapping.Gone
100100 ) {
101101 s"TASK_LOST update with $reason indicating a task won't come" in new Fixture {
102102 instanceTracker.instance(existingInstance.instanceId) returns Future.successful(Some(existingInstance))
103 val stateOp: InstanceUpdateOperation.MesosUpdate = TaskStatusUpdateTestHelper.lost(reason, existingInstance).operation.asInstanceOf[InstanceUpdateOperation.MesosUpdate]
103 val stateOp: InstanceUpdateOperation.MesosUpdate = TaskStatusUpdateTestHelper.lost(
104 reason, existingInstance).operation.asInstanceOf[InstanceUpdateOperation.MesosUpdate]
104105 val stateChange = updateOpResolver.resolve(stateOp).futureValue
105106
106107 When("call taskTracker.task")
107108 verify(instanceTracker).instance(existingInstance.instanceId)
108109
109110 Then("result in an Expunge with the correct status")
110111 stateChange shouldBe a[InstanceUpdateEffect.Expunge]
111112
Show All 9 Lines
121122 val expectedState = existingInstance.copy(
122123 state = existingInstance.state.copy(
123124 condition = TaskCondition(stateOp.mesosStatus),
124125 since = stateOp.now
125126 ),
126127 tasksMap = updatedTasksMap
127128 )
128129
129 val events = eventsGenerator.events(expectedState, Some(updatedTask), stateOp.now, previousCondition = Some(existingInstance.state.condition))
130 val events = eventsGenerator.events(
131 expectedState, Some(updatedTask), stateOp.now, previousCondition = Some(existingInstance.state.condition))
130132 stateChange shouldEqual InstanceUpdateEffect.Expunge(expectedState, events)
131133
132134 verifyNoMoreInteractions()
133135 }
134136 }
135137
136138 for (
137139 reason <- TaskConditionMapping.Unreachable
138140 ) {
139141 s"a TASK_LOST update with an unreachable $reason but a message saying that the task is unknown to the slave " in new Fixture {
140142 instanceTracker.instance(existingTask.taskId.instanceId) returns Future.successful(Some(existingInstance))
141143 val message = "Reconciliation: Task is unknown to the slave"
142 val stateOp: InstanceUpdateOperation.MesosUpdate = TaskStatusUpdateTestHelper.lost(reason, existingInstance, Some(message)).operation.asInstanceOf[InstanceUpdateOperation.MesosUpdate]
144 val stateOp: InstanceUpdateOperation.MesosUpdate = TaskStatusUpdateTestHelper.lost(
145 reason, existingInstance, Some(message)).operation.asInstanceOf[InstanceUpdateOperation.MesosUpdate]
143146 val stateChange = updateOpResolver.resolve(stateOp).futureValue
144147
145148 When("call taskTracker.task")
146149 verify(instanceTracker).instance(existingTask.taskId.instanceId)
147150
148151 Then("result in an expunge")
149152 stateChange shouldBe a[InstanceUpdateEffect.Expunge]
150153
Show All 23 Lines
174177
175178 verifyNoMoreInteractions()
176179 }
177180
178181 "a subsequent TASK_LOST update with a message saying that the task is unknown to the slave" in new Fixture {
179182 instanceTracker.instance(unreachableInstance.instanceId) returns Future.successful(Some(unreachableInstance))
180183 val reason = mesos.Protos.TaskStatus.Reason.REASON_RECONCILIATION
181184 val maybeMessage = Some("Reconciliation: Task is unknown to the slave")
182 val stateOp: InstanceUpdateOperation.MesosUpdate = TaskStatusUpdateTestHelper.lost(reason, unreachableInstance, maybeMessage).operation.asInstanceOf[InstanceUpdateOperation.MesosUpdate]
185 val stateOp: InstanceUpdateOperation.MesosUpdate = TaskStatusUpdateTestHelper.lost(
186 reason, unreachableInstance, maybeMessage).operation.asInstanceOf[InstanceUpdateOperation.MesosUpdate]
183187 val stateChange = updateOpResolver.resolve(stateOp).futureValue
184188
185189 When("call taskTracker.task")
186190 verify(instanceTracker).instance(unreachableInstance.instanceId)
187191
188192 Then("result in an expunge")
189193 stateChange shouldBe a[InstanceUpdateEffect.Expunge]
190194
Show All 22 Lines
213217 Then("result in a Failure")
214218 stateChange shouldBe a[InstanceUpdateEffect.Failure]
215219
216220 verifyNoMoreInteractions()
217221
218222 }
219223
220224 "Processing a Reserve for an existing instanceId" in new Fixture {
221 instanceTracker.instance(existingReservedInstance.instanceId) returns Future.successful(Some(existingReservedInstance))
222 val stateChange = updateOpResolver.resolve(InstanceUpdateOperation.Reserve(existingReservedInstance)).futureValue
225 instanceTracker.instance(reservedInstance.instanceId) returns Future.successful(Some(reservedInstance))
226 val stateChange = updateOpResolver.resolve(InstanceUpdateOperation.Reserve(reservedInstance)).futureValue
223227
224228 When("call taskTracker.task")
225 verify(instanceTracker).instance(existingReservedInstance.instanceId)
229 verify(instanceTracker).instance(reservedInstance.instanceId)
226230
227231 Then("result in a Failure")
228232 stateChange shouldBe a[InstanceUpdateEffect.Failure]
229233
230234 verifyNoMoreInteractions()
231235 }
232236
233237 "Revert" in new Fixture {
234 val stateChange = updateOpResolver.resolve(InstanceUpdateOperation.Revert(existingReservedInstance)).futureValue
238 val stateChange = updateOpResolver.resolve(InstanceUpdateOperation.Revert(reservedInstance)).futureValue
235239
236240 When("result in an Update")
237 stateChange shouldEqual InstanceUpdateEffect.Update(existingReservedInstance, None, events = Nil)
241 stateChange shouldEqual InstanceUpdateEffect.Update(reservedInstance, None, events = Nil)
238242
239243 Then("not query the taskTracker all")
240244 verifyNoMoreInteractions()
241245 }
242246
243247 // Mesos 1.1 task statuses specs. See https://mesosphere.atlassian.net/browse/DCOS-9941
244248
245249 "Processing a TASK_FAILED update for running task" in new Fixture {
▲ Show 20 Lines • Show All 66 Lines • ▼ Show 20 Line(s)
312316 instanceTracker.instance(instance.instanceId) returns Future.successful(Some(instance))
313317 val update = TaskStatusUpdateTestHelper.unreachable(instance)
314318 val stateChange = updateOpResolver.resolve(update.operation).futureValue
315319
316320 When("fetch the instance")
317321 verify(instanceTracker).instance(instance.instanceId)
318322
319323 Then("result in an update")
320 stateChange shouldBe a[InstanceUpdateEffect.Update]
321 val updateEffect = stateChange.asInstanceOf[InstanceUpdateEffect.Update]
324 inside(stateChange) {
325 case updateEffect: InstanceUpdateEffect.Update =>
322326 updateEffect.events should have size 2
327 }
323328
324329 verifyNoMoreInteractions()
325330 }
326331
327332 "Processing a TASK_UNREACHABLE update for a starting task" in new Fixture {
328333 val builder = TestInstanceBuilder.newBuilder(appId)
329334 val instance = builder.addTaskStarting().getInstance()
330335 instanceTracker.instance(instance.instanceId) returns Future.successful(Some(instance))
331336 val update = TaskStatusUpdateTestHelper.unreachable(instance)
332337 val stateChange = updateOpResolver.resolve(update.operation).futureValue
333338
334339 When("fetch the instance")
335340 verify(instanceTracker).instance(instance.instanceId)
336341
337342 Then("result in an update")
338343 stateChange shouldBe a[InstanceUpdateEffect.Update]
339 val updateEffect = stateChange.asInstanceOf[InstanceUpdateEffect.Update]
344 inside(stateChange) {
345 case updateEffect: InstanceUpdateEffect.Update =>
340346 updateEffect.events should have size 2
341
347 }
342348 verifyNoMoreInteractions()
343
344349 }
345350
346351 "Processing a TASK_UNREACHABLE update for a running task" in new Fixture {
347352 val builder = TestInstanceBuilder.newBuilder(appId)
348353 val instance = builder.addTaskRunning().getInstance()
349354 instanceTracker.instance(instance.instanceId) returns Future.successful(Some(instance))
350355 val update = TaskStatusUpdateTestHelper.unreachable(instance)
351356 val stateChange = updateOpResolver.resolve(update.operation).futureValue
352357
353358 When("fetch the instance")
354359 verify(instanceTracker).instance(instance.instanceId)
355360
356361 Then("result in an update")
357 stateChange shouldBe a[InstanceUpdateEffect.Update]
358 val updateEffect = stateChange.asInstanceOf[InstanceUpdateEffect.Update]
362 inside(stateChange) {
363 case updateEffect: InstanceUpdateEffect.Update =>
359364 updateEffect.events should have size 2
365 }
360366
361367 verifyNoMoreInteractions()
362368 }
363369
364370 "Processing a TASK_UNKNOWN update for an unreachable task" in new Fixture {
365371 val builder = TestInstanceBuilder.newBuilder(appId)
366372 val instance = builder.addTaskUnreachable().getInstance()
367373 instanceTracker.instance(instance.instanceId) returns Future.successful(Some(instance))
Show All 11 Lines
379385 }
380386
381387 class Fixture {
382388 val eventsGenerator = InstanceChangedEventsGenerator
383389 val clock = ConstantClock(Timestamp.now())
384390 val instanceTracker = mock[InstanceTracker]
385391 val updateOpResolver = new InstanceUpdateOpResolver(instanceTracker, clock)
386392
387 val appId = PathId("/app")
388 val existingInstance: Instance = TestInstanceBuilder.newBuilder(appId).addTaskRunning().getInstance()
389 val existingTask: Task.LaunchedEphemeral = existingInstance.appTask
393 lazy val appId = PathId("/app")
394 lazy val existingInstance: Instance = TestInstanceBuilder.newBuilder(appId).addTaskRunning().getInstance()
395 lazy val existingTask: Task.LaunchedEphemeral = existingInstance.appTask
390396
391 val existingReservedInstance = TestInstanceBuilder.newBuilder(appId).addTaskReserved().getInstance()
392 val existingReservedTask: Task.Reserved = existingReservedInstance.appTask
393 val notExistingInstanceId = Instance.Id.forRunSpec(appId)
394 val unreachableInstance = TestInstanceBuilder.newBuilder(appId).addTaskUnreachable().getInstance()
397 lazy val reservedInstance = TestInstanceBuilder.newBuilder(appId).addTaskReserved().getInstance()
398 lazy val existingReservedTask: Task.Reserved = reservedInstance.appTask
399
400 lazy val reservedLaunchedInstance: Instance = TestInstanceBuilder.newBuilder(appId).addTaskResidentLaunched().getInstance()
401
402 lazy val notExistingInstanceId = Instance.Id.forRunSpec(appId)
403 lazy val unreachableInstance = TestInstanceBuilder.newBuilder(appId).addTaskUnreachable().getInstance()
395404
396405 def verifyNoMoreInteractions(): Unit = {
397406 noMoreInteractions(instanceTracker)
398407 }
399408 }
400409}
View Options

src/test/scala/mesosphere/marathon/core/matcher/base/util/ActorOfferMatcherTest.scala

11package mesosphere.marathon
22package core.matcher.base.util
33
4import mesosphere.Unstable
45import akka.actor.ActorRef
56import akka.testkit.TestActor.AutoPilot
67import akka.testkit.{ TestActor, TestProbe }
78import mesosphere.AkkaUnitTest
89import mesosphere.marathon.core.matcher.base.OfferMatcher.MatchedInstanceOps
910import mesosphere.marathon.state.Timestamp
1011import mesosphere.marathon.test.MarathonTestHelper
1112import mesosphere.mesos.protos.Implicits._
▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Line(s)
6465 val offerMatch: MatchedInstanceOps = offerMatcher.matchOffer(now, deadline, offer).futureValue
6566
6667 offerMatch should be(MatchedInstanceOps.noMatch(offer.getId))
6768 probe.expectNoMsg()
6869 }
6970 }
7071
7172 "the actor takes too long to process" should {
72 "receive a no match after deadline" in {
73 "receive a no match after deadline" taggedAs Unstable in {
7374 val now = Timestamp.zero
7475 val deadline = now + 60.millis
7576
7677 val probe = TestProbe()
7778 probe.setAutoPilot(new TestActor.AutoPilot {
7879 override def run(sender: ActorRef, msg: Any): AutoPilot = {
7980 msg match {
8081 case ActorOfferMatcher.MatchOffer(deadline, offer, p) =>
Show All 21 Lines
View Options

src/test/scala/mesosphere/marathon/core/task/termination/impl/KillActionTest.scala

  • This file was added.
1package mesosphere.marathon
2package core.task.termination.impl
3
4import mesosphere.UnitTest
5import mesosphere.marathon.core.base.ConstantClock
6import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
7import mesosphere.marathon.core.task.Task.LocalVolumeId
8import mesosphere.marathon.state.PathId
9import org.scalatest.prop.TableDrivenPropertyChecks
10
11class KillActionTest extends UnitTest with TableDrivenPropertyChecks {
12
13 val clock = ConstantClock()
14 val appId = PathId("/test")
15
16 lazy val localVolumeId = LocalVolumeId(appId, "unwanted-persistent-volume", "uuid1")
17 lazy val residentLaunchedInstance: Instance = TestInstanceBuilder.newBuilder(appId).
18 addTaskResidentLaunched(localVolumeId).
19 getInstance()
20
21 lazy val residentUnreachableInstance: Instance = TestInstanceBuilder.newBuilder(appId).
22 addTaskWithBuilder().
23 taskResidentUnreachable(localVolumeId).
24 build().
25 getInstance()
26
27 lazy val unreachableInstance: Instance = TestInstanceBuilder.newBuilder(appId).addTaskUnreachable().getInstance()
28 lazy val runningInstance: Instance = TestInstanceBuilder.newBuilder(appId).addTaskLaunched().getInstance()
29
30 "computeKillAction" when {
31 Table(
32 ("name", "instance", "expected"),
33 ("an unreachable reserved instance", residentUnreachableInstance, KillAction.Noop),
34 ("a running reserved instance", residentLaunchedInstance, KillAction.IssueKillRequest),
35 ("an unreachable ephemeral instance", unreachableInstance, KillAction.ExpungeFromState),
36 ("a running ephemeral instance", runningInstance, KillAction.IssueKillRequest)
37 ).
38 foreach {
39 case (name, instance, expected) =>
40 s"killing ${name}" should {
jasongilanfarrUnsubmitted
Not Done

Remove the formatting pretty please.

41 s"result in ${expected}" in {
42 KillAction(
43 instance.instanceId, instance.tasksMap.keys, Some(instance)).
44 shouldBe(expected)
45 }
46 }
47 }
48 }
49}