HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D314

Remove additional condition from InstanceChangedEventsGenerator.events
ClosedAll Users
Actions

Authored by meichstedt on Dec 12 2016, 11:29 AM.

Details

Reviewers
aquamatthias
unterstein
jeschkies
Commits
rMARATHON83634c26d7b3: Remove additional condition from InstanceChangedEventsGenerator.events
rMARATHONad881100cb0b: Remove additional condition from InstanceChangedEventsGenerator.events
rMARATHON280747f6e78e: Remove additional condition from InstanceChangedEventsGenerator.events
Summary
  • remove the condition that was previously passed in addition to the instance
  • replace instanceChanged flag with previousCondition and let the generator figure out whether the condition has changed
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.
meichstedt retitled this revision from to Remove additional condition from InstanceChangedEventsGenerator.events.Dec 12 2016, 11:29 AM
meichstedt updated this object.
meichstedt edited the test plan for this revision. (Show Details)
meichstedt added reviewers: jeschkies, unterstein.
meichstedt added a subscriber: marathon-team.
aquamatthias accepted this revision.Dec 12 2016, 2:21 PM
This revision is now accepted and ready to land.Dec 12 2016, 2:21 PM
unterstein accepted this revision.Dec 12 2016, 2:22 PM
This comment was removed by unterstein.
Closed by commit rMARATHONad881100cb0b: Remove additional condition from InstanceChangedEventsGenerator.events (authored by meichstedt). · Explain WhyDec 12 2016, 2:24 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents

Diff 1221

View Options

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

Show First 20 Lines • Show All 66 Lines • ▼ Show 20 Line(s)
6767 op match {
6868 case InstanceUpdateOperation.MesosUpdate(instance, status, mesosStatus, now) =>
6969 val taskId = Task.Id(mesosStatus.getTaskId)
7070 tasksMap.get(taskId).map { task =>
7171 val taskEffect = task.update(TaskUpdateOperation.MesosUpdate(status, mesosStatus, now))
7272 taskEffect match {
7373 case TaskUpdateEffect.Update(updatedTask) =>
7474 val updated: Instance = updatedInstance(updatedTask, now)
75 val events = eventsGenerator.events(updated, Some(updatedTask), now, updated.state.condition != this.state.condition)
75 val events = eventsGenerator.events(updated, Some(updatedTask), now, previousCondition = Some(state.condition))
7676 if (updated.tasksMap.values.forall(_.isTerminal)) {
7777 Instance.log.info("all tasks of {} are terminal, requesting to expunge", updated.instanceId)
7878 InstanceUpdateEffect.Expunge(updated, events)
7979 } else {
8080 InstanceUpdateEffect.Update(updated, oldState = Some(this), events)
8181 }
8282
8383 // We might still become UnreachableInactive.
8484 case TaskUpdateEffect.Noop if status == Condition.Unreachable && this.state.condition != Condition.UnreachableInactive =>
8585 val updated: Instance = updatedInstance(task, now)
8686 if (updated.state.condition == Condition.UnreachableInactive) {
87 val events = eventsGenerator.events(updated, Some(task), now, updated.state.condition != this.state.condition)
87 val events = eventsGenerator.events(updated, Some(task), now, previousCondition = Some(state.condition))
8888 InstanceUpdateEffect.Update(updated, oldState = Some(this), events)
8989 } else {
9090 InstanceUpdateEffect.Noop(instance.instanceId)
9191 }
9292
9393 case TaskUpdateEffect.Noop =>
9494 InstanceUpdateEffect.Noop(instance.instanceId)
9595
Show All 17 Lines
113113 val updated = this.copy(
114114 state = state.copy(
115115 condition = Condition.Staging,
116116 since = timestamp
117117 ),
118118 tasksMap = tasksMap.updated(task.taskId, updatedTask),
119119 runSpecVersion = newRunSpecVersion
120120 )
121 val events = eventsGenerator.events(updated, task = None, timestamp, instanceChanged = updated.state.condition != this.state.condition)
121 val events = eventsGenerator.events(updated, task = None, timestamp, previousCondition = Some(state.condition))
122122 InstanceUpdateEffect.Update(updated, oldState = Some(this), events)
123123
124124 case _ =>
125125 InstanceUpdateEffect.Failure(s"Unexpected taskUpdateEffect $taskEffect")
126126 }
127127 } else {
128128 InstanceUpdateEffect.Failure("LaunchOnReservation can only be applied to a reserved instance")
129129 }
130130
131131 case InstanceUpdateOperation.ReservationTimeout(_) =>
132132 if (this.isReserved) {
133 // TODO(cleanup): don't use Killed - use Obsolete or similar
134 val updatedInstance = copy(
135 state = state.copy(condition = Condition.Killed)
136 )
133137 // TODO(PODS): don#t use Timestamp.now()
134 val events = eventsGenerator.events(this, task = None, Timestamp.now(), instanceChanged = true)
138 val events = eventsGenerator.events(updatedInstance, task = None, Timestamp.now(), previousCondition = Some(state.condition))
135139 InstanceUpdateEffect.Expunge(this, events)
136140 } else {
137141 InstanceUpdateEffect.Failure("ReservationTimeout can only be applied to a reserved instance")
138142 }
139
140143 case InstanceUpdateOperation.LaunchEphemeral(instance) =>
141144 InstanceUpdateEffect.Failure("LaunchEphemeral cannot be passed to an existing instance")
142145
143146 case InstanceUpdateOperation.Reserve(_) =>
144147 InstanceUpdateEffect.Failure("Reserve cannot be passed to an existing instance")
145148
146149 case InstanceUpdateOperation.Revert(oldState) =>
147150 InstanceUpdateEffect.Failure("Revert cannot be passed to an existing instance")
▲ Show 20 Lines • Show All 343 Lines • Show Last 20 Lines
View Options

src/main/scala/mesosphere/marathon/core/instance/update/InstanceChangedEventsGenerator.scala

11package mesosphere.marathon.core.instance.update
22
33import mesosphere.marathon.core.condition.Condition
44import mesosphere.marathon.core.event.{ InstanceChanged, MarathonEvent, MesosStatusUpdateEvent }
55import mesosphere.marathon.core.instance.Instance
66import mesosphere.marathon.core.task.Task
77import mesosphere.marathon.state.Timestamp
88import org.apache.mesos.Protos.TaskState
99
1010import scala.collection.immutable.Seq
1111
1212object InstanceChangedEventsGenerator {
13 def events(instance: Instance, task: Option[Task], now: Timestamp, instanceChanged: Boolean): Seq[MarathonEvent] = {
14 events(instance.state.condition, instance, task, now, instanceChanged)
15 }
16
17 def events(condition: Condition, instance: Instance, task: Option[Task], now: Timestamp, instanceChanged: Boolean): Seq[MarathonEvent] = {
13 def events(instance: Instance, task: Option[Task], now: Timestamp, previousCondition: Option[Condition]): Seq[MarathonEvent] = {
14 val stateChanged = previousCondition.fold(true)(_ != instance.state.condition)
1815 val runSpecId = instance.runSpecId
1916 val version = instance.runSpecVersion
2017
21 val instanceEvent: Seq[MarathonEvent] = if (instanceChanged) {
18 val instanceEvent: Seq[MarathonEvent] = if (stateChanged) {
2219 Seq(InstanceChanged(
2320 id = instance.instanceId,
2421 runSpecVersion = version,
2522 runSpecId = runSpecId,
26 condition = condition,
23 condition = instance.state.condition,
2724 instance = instance
2825 ))
2926 } else Nil
3027
3128 task.fold(instanceEvent) { task =>
3229 val maybeTaskStatus = task.status.mesosStatus
3330 val ports = task.status.networkInfo.hostPorts
3431 val host = instance.agentInfo.host
Show All 23 Lines
View Options

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

Show First 20 Lines • Show All 42 Lines • ▼ Show 20 Line(s)
4343 implicit
4444 ec: ExecutionContext): Future[InstanceUpdateEffect] = {
4545 directInstanceTracker.instance(instanceId).map {
4646 case Some(existingInstance) =>
4747 InstanceUpdateEffect.Failure( //
4848 new IllegalStateException(s"$instanceId of app [${instanceId.runSpecId}] already exists"))
4949
5050 case None =>
51 val events = eventsGenerator.events(updatedInstance, task = None, clock.now(), instanceChanged = true)
51 val events = eventsGenerator.events(updatedInstance, task = None, clock.now(), previousCondition = None)
5252 InstanceUpdateEffect.Update(updatedInstance, oldState = None, events)
5353 }
5454 }
5555
5656 private[this] def updateExistingInstance(op: InstanceUpdateOperation) //
5757 (implicit ec: ExecutionContext): Future[InstanceUpdateEffect] = {
5858 directInstanceTracker.instance(op.instanceId).map {
5959 case Some(existingInstance) =>
6060 existingInstance.update(op)
6161
6262 case None =>
6363 val id = op.instanceId
6464 InstanceUpdateEffect.Failure(new IllegalStateException(s"$id of app [${id.runSpecId}] does not exist"))
6565 }
6666 }
6767
6868 private[this] def expungeInstance(id: Instance.Id)(implicit ec: ExecutionContext): Future[InstanceUpdateEffect] = {
6969 directInstanceTracker.instance(id).map {
7070 case Some(existingInstance: Instance) =>
71 val events = eventsGenerator.events(Condition.Killed, existingInstance, task = None, clock.now(), instanceChanged = true)
71 val updatedInstance = existingInstance.copy(
72 state = existingInstance.state.copy(
73 // TODO(cleanup): what would be a correct state here? ForceExpunged, Obsolete?
74 condition = Condition.Killed // Using Killed for now to report a terminal state
75 )
76 )
77 val events = eventsGenerator.events(
78 updatedInstance, task = None, clock.now(), previousCondition = Some(existingInstance.state.condition))
7279 InstanceUpdateEffect.Expunge(existingInstance, events)
7380
7481 case None =>
7582 log.info("Ignoring ForceExpunge for [{}], task does not exist", id)
7683 InstanceUpdateEffect.Noop(id)
7784 }
7885 }
7986}
View Options

src/test/scala/mesosphere/marathon/core/task/KillServiceMock.scala

Show All 28 Lines
2929 override def killInstances(instances: Seq[Instance], reason: KillReason): Future[Done] = {
3030 instances.foreach { instance =>
3131 killInstance(instance, reason)
3232 }
3333 Future.successful(Done)
3434 }
3535 override def killInstance(instance: Instance, reason: KillReason): Future[Done] = {
3636 val id = instance.instanceId
37 val events = customStatusUpdates.getOrElse(id, eventsGenerator.events(Condition.Killed, instance, task = None, now = clock.now(), instance.state.condition != Condition.Killed))
37 val updatedInstance = instance.copy(state = instance.state.copy(condition = Condition.Killed))
38 val events = customStatusUpdates.getOrElse(id, eventsGenerator.events(updatedInstance, task = None, now = clock.now(), previousCondition = Some(instance.state.condition)))
3839 events.foreach(system.eventStream.publish)
3940 numKilled += 1
4041 killed += id
4142 Future.successful(Done)
4243 }
4344
4445 override def killUnknownTask(taskId: Id, reason: KillReason): Unit = {
4546 val instance = mock[Instance]
4647 instance.instanceId returns taskId.instanceId
4748 killInstance(instance, reason)
4849 }
4950}
5051
View Options

src/test/scala/mesosphere/marathon/core/task/tracker/impl/InstanceTrackerActorTest.scala

Show First 20 Lines • Show All 103 Lines • ▼ Show 20 Line(s)
104104 )
105105 f.taskLoader.load() returns Future.successful(appDataMap)
106106
107107 When("staged task gets deleted")
108108 val probe = TestProbe()
109109 val helper = TaskStatusUpdateTestHelper.killed(stagedInstance)
110110 val operation = helper.operation.asInstanceOf[InstanceUpdateOperation.MesosUpdate]
111111 val stagedUpdate = helper.effect
112 val (stagedTaskId, stagedTask) = stagedInstance.tasksMap.head
112 val stagedTaskId = stagedInstance.firstTask.taskId
113113 val expectedTask = TestTaskBuilder.Helper.killedTask(stagedTaskId)
114114 val stagedAck = InstanceTrackerActor.Ack(probe.ref, stagedUpdate)
115 val c = helper.wrapped.condition
116 val events = f.eventsGenerator.events(helper.wrapped.condition, helper.wrapped.instance, Some(expectedTask), operation.now, stagedInstance.state.condition != helper.wrapped.condition)
115 val events = f.eventsGenerator.events(helper.wrapped.instance, Some(expectedTask), operation.now, previousCondition = Some(stagedInstance.state.condition))
117116
118117 probe.send(f.taskTrackerActor, InstanceTrackerActor.StateChanged(stagedAck))
119118 probe.expectMsg(InstanceUpdateEffect.Expunge(helper.wrapped.instance, events))
120119
121120 Then("it will have set the correct metric counts")
122121 f.actorMetrics.runningCount.getValue should be(2)
123122 f.actorMetrics.stagedCount.getValue should be(0)
124123
▲ Show 20 Lines • Show All 106 Lines • Show Last 20 Lines
View Options

src/test/scala/mesosphere/marathon/core/task/tracker/impl/InstanceUpdateOpResolverTest.scala

Show First 20 Lines • Show All 147 Lines • ▼ Show 20 Line(s)
148148 val expectedState = f.existingInstance.copy(
149149 state = f.existingInstance.state.copy(
150150 condition = TaskCondition(stateOp.mesosStatus),
151151 since = stateOp.now
152152 ),
153153 tasksMap = updatedTasksMap
154154 )
155155
156 val events = f.eventsGenerator.events(expectedState.state.condition, expectedState, Some(updatedTask), stateOp.now, expectedState.state.condition != f.existingInstance.state.condition)
156 val events = f.eventsGenerator.events(expectedState, Some(updatedTask), stateOp.now, previousCondition = Some(f.existingInstance.state.condition))
157157 stateChange shouldEqual InstanceUpdateEffect.Expunge(expectedState, events)
158158
159159 And("there are no more interactions")
160160 f.verifyNoMoreInteractions()
161161 }
162162 }
163163
164164 for (
▲ Show 20 Lines • Show All 364 Lines • Show Last 20 Lines