src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala
540–541	Can you please rename `task` to `instance`?
src/test/scala/mesosphere/marathon/MarathonSchedulerActorTest.scala
268 ↗	(On Diff #1092)	Thanks for the test! How about a test that also verifies that no instances are added to the launch queue when either the queued instance count is equal to `toStart`, or the queued instance count is bigger than `toStart`. The latter is clearly a failure scenario, probably of another component. Yet adding -3 to the launch queue wouldn't make sense so we should verify that it works for that scenario as well.

src/test/scala/mesosphere/marathon/MarathonSchedulerActorTest.scala
293 ↗	(On Diff #1109)	This means it will wait for 5 seconds to verify `queue.add` wasn't called. Following the pattern these (older) tests normally use, please write `noMoreInteractions(queue)`. Yet, why does this code test the actor anyways? the `def scale(runSpec: RunSpec): Unit` is defined in the `SchedulerActions` class, so you can should test this in the associated `SchedulerActionsTest` – sorry for not realizing this earlier. Benefit is it doesn't use an actor system.

src/test/scala/mesosphere/marathon/MarathonSchedulerActorTest.scala
293 ↗	(On Diff #1109)	Oh, I didn't see those. Yes, `SchedulerActionsTest` is a much better place for these tests.

Diff	ID	Base	Description	Created	Lint	Unit
Base			Base
Diff 1	1091	07869fe		Dec 6 2016, 11:33 AM	★	★
Diff 2	1092	07869fe	Add new MarathonSchedulerActor "ScaleApp with tasks in launch queue" test	Dec 6 2016, 12:07 PM	★	★
Diff 3	1106	07869fe	- Add "ScaleApp with too many tasks in launch queue" MarathonSchedulerActor test	Dec 6 2016, 3:48 PM	★	★
Diff 4	1108	07869fe	Fix failing SchedulerActionsTest with a forgotten mock call	Dec 6 2016, 4:12 PM	★	★
Diff 5	1109	46a86b3	Rebased on master	Dec 6 2016, 4:43 PM	★	★
Diff 6	1119	46a86b3	Moved new tests to SchedulerActionsTest	Dec 7 2016, 11:00 AM	★	★
Diff 7	1124	ef6a71d	Rebased on master	Dec 7 2016, 1:34 PM	★	★
Diff 8	1132	ef6a71d	Fix failing pod test	Dec 7 2016, 3:53 PM	★	★
Diff 9	1140	89740d0	rMARATHON4a19a8693c2ad6e166bc21ccbf7ea94583c60d6c	Dec 7 2016, 6:17 PM	★	★

Commit	Tree	Parents	Author	Summary	Date
ade03a58da8d	23d58c6b1aaa	ef6a71d2b50d	Aleksey Dukhovniy	MarathonSchedulerActor now respects launch queue tasks when scaling app (Show More…)	Dec 6 2016, 11:32 AM

Diff 1140

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

Show First 20 Lines • Show All 512 Lines • ▼ Show 20 Line(s)
513	513		}
514	514
515	515		/**
516	516		* Make sure the runSpec is running the correct number of instances
517	517		*/
518	518		// FIXME: extract computation into a function that can be easily tested
519	519		def scale(runSpec: RunSpec): Unit = {
520	520
521			val runningTasks = instanceTracker.specInstancesSync(runSpec.id).filter(_.state.condition.isActive)
	521		val runningInstances = instanceTracker.specInstancesSync(runSpec.id).filter(_.state.condition.isActive)
522	522
523	523		def killToMeetConstraints(notSentencedAndRunning: Seq[Instance], toKillCount: Int) = {
524	524		Constraints.selectInstancesToKill(runSpec, notSentencedAndRunning, toKillCount)
525	525		}
526	526
527	527		val targetCount = runSpec.instances
528	528
529			val ScalingProposition(tasksToKill, tasksToStart) = ScalingProposition.propose(
530			runningTasks, None, killToMeetConstraints, targetCount, runSpec.killSelection)
	529		val ScalingProposition(instancesToKill, instancesToStart) = ScalingProposition.propose(
	530		runningInstances, None, killToMeetConstraints, targetCount, runSpec.killSelection)
531	531
532			tasksToKill.foreach { tasks: Seq[Instance] =>
533			log.info(s"Scaling ${runSpec.id} from ${runningTasks.size} down to $targetCount instances")
	532		instancesToKill.foreach { instances: Seq[Instance] =>
	533		log.info(s"Scaling ${runSpec.id} from ${runningInstances.size} down to $targetCount instances")
534	534
535	535		launchQueue.purge(runSpec.id)
536	536
537			log.info("Killing tasks {}", tasks.map(_.instanceId))
538			killService.killInstances(tasks, KillReason.OverCapacity)
	537		log.info("Killing instances {}", instances.map(_.instanceId))
	538		killService.killInstances(instances, KillReason.OverCapacity)
539	539		}
540	540
541			tasksToStart.foreach { toQueue: Int =>
	541		instancesToStart.foreach { toStart: Int =>
		meichstedtUnsubmitted Done Can you please rename `task` to `instance`?
542			log.info(s"Need to scale ${runSpec.id} from ${runningTasks.size} up to $targetCount instances")
	542		log.info(s"Need to scale ${runSpec.id} from ${runningInstances.size} up to $targetCount instances")
	543		val leftToLaunch = launchQueue.get(runSpec.id).fold(0)(_.instancesLeftToLaunch)
	544		val toAdd = toStart - leftToLaunch
543	545
544			if (toQueue > 0) {
545			log.info(s"Queueing $toQueue new tasks for ${runSpec.id}")
546			launchQueue.add(runSpec, toQueue)
	546		if (toAdd > 0) {
	547		log.info(s"Queueing $toAdd new instances for ${runSpec.id} to the already $leftToLaunch queued ones")
	548		launchQueue.add(runSpec, toAdd)
547	549		} else {
548			log.info(s"Already queued or started ${runningTasks.size} tasks for ${runSpec.id}. Not scaling.")
	550		log.info(s"Already queued or started ${runningInstances.size} instances for ${runSpec.id}. Not scaling.")
549	551		}
550	552		}
551	553
552			if (tasksToKill.isEmpty && tasksToStart.isEmpty) {
	554		if (instancesToKill.isEmpty && instancesToStart.isEmpty) {
553	555		log.info(s"Already running ${runSpec.instances} instances of ${runSpec.id}. Not scaling.")
554	556		}
555	557		}
556	558
557	559		def scale(runSpecId: PathId): Future[Unit] = {
558	560		runSpecById(runSpecId).map {
559	561		case Some(runSpec) => scale(runSpec)
560	562		case _ => log.warn(s"RunSpec $runSpecId does not exist. Not scaling.")
▲ Show 20 Lines • Show All 42 Lines • Show Last 20 Lines

View Options

src/test/scala/mesosphere/marathon/SchedulerActionsTest.scala

1	1	package mesosphere.marathon
2	2
3	3	import akka.Done
4	4	import akka.stream.scaladsl.Source
5	5	import akka.testkit.TestProbe
6	6	import mesosphere.marathon.core.base.ConstantClock
7	7	import mesosphere.marathon.core.condition.Condition
8	8	import mesosphere.marathon.core.health.HealthCheckManager
9	9	import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
	10	import mesosphere.marathon.core.launcher.impl.LaunchQueueTestHelper
10	11	import mesosphere.marathon.core.launchqueue.LaunchQueue
11	12	import mesosphere.marathon.core.task.termination.{ KillReason, KillService }
12	13	import mesosphere.marathon.core.task.tracker.InstanceTracker
13	14	import mesosphere.marathon.core.task.tracker.InstanceTracker.{ InstancesBySpec, SpecInstances }
14	15	import mesosphere.marathon.state.{ AppDefinition, PathId, Timestamp }
15	16	import mesosphere.marathon.storage.repository.{ AppRepository, GroupRepository, ReadOnlyPodRepository }
16	17	import mesosphere.marathon.stream._
17	18	import mesosphere.marathon.test.{ MarathonActorSupport, MarathonSpec, MarathonTestHelper, Mockito }
18	19	import org.apache.mesos.SchedulerDriver
19	20	import org.mockito.Mockito.verifyNoMoreInteractions
20	21	import org.scalatest.concurrent.{ PatienceConfiguration, ScalaFutures }
21	22	import org.scalatest.time.{ Millis, Span }
22	23	import org.scalatest.{ GivenWhenThen, Matchers }
23	24
24	25	import scala.concurrent.Future
25	26	import scala.concurrent.duration._
26	27
27	28	class SchedulerActionsTest
28	29	extends MarathonActorSupport
29	30	with MarathonSpec
30	31	with Matchers
31	32	with Mockito
32	33	with ScalaFutures
33	34	with GivenWhenThen {
34		import system.dispatcher
35	35
36	36	test("Reset rate limiter if application is stopped") {
37	37	val f = new Fixture
38	38	val app = AppDefinition(id = PathId("/myapp"))
39	39
40	40	f.appRepo.delete(app.id) returns Future.successful(Done)
41		f.taskTracker.specInstances(eq(app.id))(any) returns Future.successful(Seq.empty[Instance])
	41	f.instanceTracker.specInstances(eq(app.id))(any) returns Future.successful(Seq.empty[Instance])
42	42
43	43	f.scheduler.stopRunSpec(app).futureValue(1.second)
44	44
45	45	verify(f.queue).purge(app.id)
46	46	verify(f.queue).resetDelay(app)
47	47	verifyNoMoreInteractions(f.queue)
48	48	}
49	49
50	50	test("Task reconciliation sends known running and staged tasks and empty list") {
51	51	val f = new Fixture
52	52	val app = AppDefinition(id = PathId("/myapp"))
53	53	val runningInstance = TestInstanceBuilder.newBuilder(app.id).addTaskRunning().getInstance()
54	54	val stagedInstance = TestInstanceBuilder.newBuilder(app.id).addTaskStaged().getInstance()
55	55
56	56	val stagedInstanceWithSlaveId = TestInstanceBuilder.newBuilder(app.id)
57	57	.addTaskWithBuilder().taskStaged().build()
58	58	.withAgentInfo(agentId = Some("slave 1"))
59	59	.getInstance()
60	60
61	61	val instances = Seq(runningInstance, stagedInstance, stagedInstanceWithSlaveId)
62		f.taskTracker.instancesBySpec() returns Future.successful(InstancesBySpec.of(SpecInstances.forInstances(app.id, instances)))
	62	f.instanceTracker.instancesBySpec() returns Future.successful(InstancesBySpec.of(SpecInstances.forInstances(app.id, instances)))
63	63	f.appRepo.ids() returns Source.single(app.id)
64	64
65	65	f.scheduler.reconcileTasks(f.driver).futureValue(5.seconds)
66	66
67	67	verify(f.driver).reconcileTasks(Set(
68	68	runningInstance,
69	69	stagedInstance,
70	70	stagedInstanceWithSlaveId
71	71	).flatMap(_.tasksMap.values).flatMap(_.status.mesosStatus))
72	72	verify(f.driver).reconcileTasks(java.util.Arrays.asList())
73	73	}
74	74
75	75	test("Task reconciliation only one empty list, when no tasks are present in Marathon") {
76	76	val f = new Fixture
77	77
78		f.taskTracker.instancesBySpec() returns Future.successful(InstancesBySpec.empty)
	78	f.instanceTracker.instancesBySpec() returns Future.successful(InstancesBySpec.empty)
79	79	f.appRepo.ids() returns Source.empty
80	80
81	81	f.scheduler.reconcileTasks(f.driver).futureValue
82	82
83	83	verify(f.driver, times(1)).reconcileTasks(java.util.Arrays.asList())
84	84	}
85	85
86	86	test("Kill orphaned task") {
87	87	val f = new Fixture
88	88	val app = AppDefinition(id = PathId("/myapp"))
89	89	val orphanedApp = AppDefinition(id = PathId("/orphan"))
90	90	val instance = TestInstanceBuilder.newBuilder(app.id).addTaskRunning().getInstance()
91	91	val orphanedInstance = TestInstanceBuilder.newBuilder(orphanedApp.id).addTaskRunning().getInstance()
92	92
93	93	val tasksOfApp = SpecInstances.forInstances(app.id, Seq(instance))
94	94	val tasksOfOrphanedApp = SpecInstances.forInstances(orphanedApp.id, Seq(orphanedInstance))
95	95
96		f.taskTracker.instancesBySpec() returns Future.successful(InstancesBySpec.of(tasksOfApp, tasksOfOrphanedApp))
	96	f.instanceTracker.instancesBySpec() returns Future.successful(InstancesBySpec.of(tasksOfApp, tasksOfOrphanedApp))
97	97	f.appRepo.ids() returns Source.single(app.id)
98	98
99	99	f.scheduler.reconcileTasks(f.driver).futureValue(5.seconds)
100	100
101	101	verify(f.killService, times(1)).killInstance(orphanedInstance, KillReason.Orphaned)
102	102	}
103	103
104	104	test("Scale up correctly in case of lost tasks (active queue)") {
105	105	val f = new Fixture
106	106
107	107	Given("An active queue and unreachable tasks")
108	108	val app = MarathonTestHelper.makeBasicApp().copy(instances = 15)
109	109
110	110	val unreachableInstances = Seq.fill(5)(TestInstanceBuilder.newBuilder(app.id).addTaskUnreachableInactive().getInstance())
111	111	val runnningInstances = Seq.fill(10)(TestInstanceBuilder.newBuilder(app.id).addTaskRunning().getInstance())
112		f.taskTracker.specInstancesSync(eq(app.id)) returns (unreachableInstances ++ runnningInstances)
	112	f.instanceTracker.specInstancesSync(eq(app.id)) returns (unreachableInstances ++ runnningInstances)
	113	f.queue.get(eq(app.id)) returns Some(LaunchQueueTestHelper.zeroCounts)
113	114
114	115	When("the app is scaled")
115	116	f.scheduler.scale(app)
116	117
117	118	Then("5 tasks should be placed onto the launchQueue")
118	119	verify(f.queue, times(1)).add(app, 5)
119	120	}
120	121
	122	test("Scale up with some tasks in launch queue") {
	123	val f = new Fixture
	124
	125	Given("an app with 10 instances and an active queue with 4 tasks")
	126	val app = MarathonTestHelper.makeBasicApp().copy(instances = 10)
	127	f.queue.get(app.id) returns Some(LaunchQueueTestHelper.instanceCounts(instancesLeftToLaunch = 4, finalInstanceCount = 10))
	128	f.instanceTracker.specInstancesSync(app.id) returns Seq.empty[Instance]
	129
	130	When("app is scaled")
	131	f.scheduler.scale(app)
	132
	133	Then("6 more tasks are added to the queue")
	134	verify(f.queue, times(1)).add(app, 6)
	135	}
	136
	137	test("Scale up with enough tasks in launch queue") {
	138	val f = new Fixture
	139
	140	Given("an app with 10 instances and an active queue with 10 tasks")
	141	val app = MarathonTestHelper.makeBasicApp().copy(instances = 10)
	142	f.queue.get(app.id) returns Some(LaunchQueueTestHelper.instanceCounts(instancesLeftToLaunch = 10, finalInstanceCount = 10))
	143	f.instanceTracker.specInstancesSync(app.id) returns Seq.empty[Instance]
	144
	145	When("app is scaled")
	146	f.scheduler.scale(app)
	147
	148	Then("no tasks are added to the queue")
	149	verify(f.queue, never).add(eq(app), any[Int])
	150	}
	151
	152	// This test was an explicit wish by Matthias E.
	153	test("Scale up with too many tasks in launch queue") {
	154	val f = new Fixture
	155
	156	Given("an app with 10 instances and an active queue with 10 tasks")
	157	val app = MarathonTestHelper.makeBasicApp().copy(instances = 10)
	158	f.queue.get(app.id) returns Some(LaunchQueueTestHelper.instanceCounts(instancesLeftToLaunch = 15, finalInstanceCount = 10))
	159	f.instanceTracker.specInstancesSync(app.id) returns Seq.empty[Instance]
	160
	161	When("app is scaled")
	162	f.scheduler.scale(app)
	163
	164	Then("no tasks are added to the queue")
	165	verify(f.queue, never).add(eq(app), any[Int])
	166	}
	167
121	168	// This scenario is the following:
122	169	// - There's an active queue and Marathon has 10 running + 5 staged tasks
123	170	// - Marathon receives StatusUpdates for 5 previously LOST tasks
124	171	// - A scale is initiated and Marathon realizes there are 5 tasks over capacity
125	172	// => We expect Marathon to kill the 5 staged tasks
126	173	test("Kill staged tasks in correct order in case lost tasks reappear") {
127	174	val f = new Fixture
128	175
Show All 10 Lines
139	186	stagedInstance(1L),
140	187	runningInstance(),
141	188	staged_3,
142	189	runningInstance(),
143	190	staged_2,
144	191	runningInstance()
145	192	)
146	193
147		f.taskTracker.specInstancesSync(app.id) returns tasks
	194	f.instanceTracker.specInstancesSync(app.id) returns tasks
148	195	When("the app is scaled")
149	196	f.scheduler.scale(app)
150	197
151	198	Then("the queue is purged")
152	199	verify(f.queue, times(1)).purge(app.id)
153	200
154	201	And("the youngest STAGED tasks are killed")
155	202	verify(f.killService).killInstances(List(staged_3, staged_2), KillReason.OverCapacity)
Show All 20 Lines
176	223	runningInstance(stagedAt = 1L),
177	224	runningInstance(stagedAt = 4L),
178	225	runningInstance(stagedAt = 5L),
179	226	running_6,
180	227	runningInstance(stagedAt = 2L)
181	228	)
182	229
183	230	f.queue.get(app.id) returns None
184		f.taskTracker.countSpecInstancesSync(eq(app.id), any) returns 7
185		f.taskTracker.specInstancesSync(app.id) returns instances
	231	f.instanceTracker.countSpecInstancesSync(eq(app.id), any) returns 7
	232	f.instanceTracker.specInstancesSync(app.id) returns instances
186	233	When("the app is scaled")
187	234	f.scheduler.scale(app)
188	235
189	236	Then("the queue is purged")
190	237	verify(f.queue, times(1)).purge(app.id)
191	238
192	239	And("the youngest RUNNING tasks are killed")
193	240	verify(f.killService).killInstances(List(running_7, running_6), KillReason.OverCapacity)
Show All 23 Lines
217	264	val tasks: Seq[Instance] = Seq(
218	265	runningInstance(stagedAt = 3L),
219	266	running_4,
220	267	staged_1,
221	268	runningInstance(stagedAt = 1L),
222	269	runningInstance(stagedAt = 2L)
223	270	)
224	271
225		f.taskTracker.countSpecInstancesSync(eq(app.id), any) returns 5
226		f.taskTracker.specInstancesSync(app.id) returns tasks
	272	f.instanceTracker.countSpecInstancesSync(eq(app.id), any) returns 5
	273	f.instanceTracker.specInstancesSync(app.id) returns tasks
227	274	When("the app is scaled")
228	275	f.scheduler.scale(app)
229	276
230	277	Then("the queue is purged")
231	278	verify(f.queue, times(1)).purge(app.id)
232	279
233	280	And("all STAGED tasks plus the youngest RUNNING tasks are killed")
234	281	verify(f.killService).killInstances(List(staged_1, running_4), KillReason.OverCapacity)
235	282	verifyNoMoreInteractions(f.driver)
236	283	verifyNoMoreInteractions(f.killService)
237	284	}
238	285
239	286	import scala.language.implicitConversions
240	287	implicit def durationToPatienceConfigTimeout(d: FiniteDuration): PatienceConfiguration.Timeout = {
241	288	PatienceConfiguration.Timeout(Span(d.toMillis, Millis))
242	289	}
243	290
244	291	class Fixture {
245	292	val queue = mock[LaunchQueue]
246	293	val appRepo = mock[AppRepository]
247	294	val podRepo: ReadOnlyPodRepository = mock[ReadOnlyPodRepository]
248		val taskTracker = mock[InstanceTracker]
	295	val instanceTracker = mock[InstanceTracker]
249	296	val driver = mock[SchedulerDriver]
250	297	val killService = mock[KillService]
251	298	val clock = ConstantClock()
252	299
253	300	podRepo.ids() returns Source.empty[PathId]
254	301
255	302	val scheduler = new SchedulerActions(
256	303	appRepo,
257	304	podRepo,
258	305	mock[GroupRepository],
259	306	mock[HealthCheckManager],
260		taskTracker,
	307	instanceTracker,
261	308	queue,
262	309	system.eventStream,
263	310	TestProbe().ref,
264	311	killService
265	312	)
266	313	}
267	314
268	315	}

View Options

src/test/scala/mesosphere/marathon/core/launcher/impl/LaunchQueueTestHelper.scala

1	1	package mesosphere.marathon.core.launcher.impl
2	2
3	3	import mesosphere.marathon.core.launchqueue.LaunchQueue
4	4	import mesosphere.marathon.state.{ PathId, Timestamp, AppDefinition }
5	5
6	6	object LaunchQueueTestHelper {
7	7	val zeroCounts = LaunchQueue.QueuedInstanceInfo(
8	8	runSpec = AppDefinition(PathId("/thisisignored")),
9	9	inProgress = true,
10	10	instancesLeftToLaunch = 0,
11	11	finalInstanceCount = 0,
12	12	unreachableInstances = 0,
13	13	backOffUntil = Timestamp(0),
14	14	startedAt = Timestamp(0)
15	15	)
	16
	17	def instanceCounts(instancesLeftToLaunch: Int, finalInstanceCount: Int): LaunchQueue.QueuedInstanceInfo = {
	18	LaunchQueue.QueuedInstanceInfo(
	19	runSpec = AppDefinition(PathId("/thisisignored")),
	20	inProgress = true,
	21	instancesLeftToLaunch = instancesLeftToLaunch,
	22	finalInstanceCount = finalInstanceCount,
	23	unreachableInstances = 0,
	24	backOffUntil = Timestamp(0),
	25	startedAt = Timestamp(0)
	26	)
	27	}
16	28	}

			Path	Packages
M			src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala (30 lines)
M			src/test/scala/mesosphere/marathon/SchedulerActionsTest.scala (73 lines)
M			src/test/scala/mesosphere/marathon/core/launcher/impl/LaunchQueueTestHelper.scala (12 lines)

MarathonSchedulerActor now respects launch queue tasks when scaling app
ClosedAll Users
Actions

Details

Diff Detail

Revision Contents

Diff 1140

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

src/test/scala/mesosphere/marathon/SchedulerActionsTest.scala

src/test/scala/mesosphere/marathon/core/launcher/impl/LaunchQueueTestHelper.scala

MarathonSchedulerActor now respects launch queue tasks when scaling appClosedAll UsersActions

Details

Diff Detail

Revision Contents

Diff 1140

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

src/test/scala/mesosphere/marathon/SchedulerActionsTest.scala

src/test/scala/mesosphere/marathon/core/launcher/impl/LaunchQueueTestHelper.scala

MarathonSchedulerActor now respects launch queue tasks when scaling app
ClosedAll Users
Actions