HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D317

Fixes #4824 - Use GroupRepository instead of AppRepository and PodRepository in MarathonSchedulerActor
ClosedAll Users
Actions

Authored by unterstein on Dec 12 2016, 4:47 PM.

Details

Reviewers
aquamatthias
timcharper
meichstedt
Commits
rMARATHONa139e513d313: Fixes #4824 - Use GroupRepository instead of AppRepository and PodRepository in…
rMARATHON784c2d4fb458: fixed tests
rMARATHON4b884b28af65: Fixes #4824 - Use GroupRepository instead of AppRepository and PodRepository in…
rMARATHON167f75c03047: Fixes #4824 - Use GroupRepository instead of AppRepository and PodRepository in…
rMARATHON7bdbaea1f268: removed unused implicit
rMARATHON2d14b3566af9: Fixes #4824 - Use GroupRepository instead of AppRepository and PodRepository in…
Summary

Using groupRepository as the only source of truth about running apps

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.
unterstein retitled this revision from to Fixes #4824 - Use GroupRepository instead of AppRepository and PodRepository in MarathonSchedulerActor.Dec 12 2016, 4:47 PM
unterstein updated this object.
unterstein edited the test plan for this revision. (Show Details)
unterstein added reviewers: timcharper, aquamatthias, meichstedt.
unterstein added a subscriber: marathon-team.
unterstein updated this revision to Diff 1231.Dec 12 2016, 5:08 PM
  • removed unused implicit
timcharper added inline comments.Dec 12 2016, 6:55 PM
src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala
397–398

I vote we change the method signature here to Unit, and then use future.foreach.

aquamatthias accepted this revision.Dec 12 2016, 7:49 PM

Thanks.
We load the whole RootGroup, take the id and then load the RunSpec by Id.
We should improve this in a follow up.

This revision is now accepted and ready to land.Dec 12 2016, 7:49 PM
meichstedt accepted this revision.Dec 12 2016, 8:07 PM
meichstedt added inline comments.
src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala
397–398

Makes sense to me too. The only usage of this method doesn't even care.

timcharper updated this revision to Diff 1236.Dec 12 2016, 8:17 PM

Future[Unit] -> Unit

timcharper accepted this revision.Dec 12 2016, 8:17 PM
timcharper added inline comments.
src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala
397–398

I went ahead and did this and updated the patch. If this was naughty behavior then I’m sorry!

Closed by commit rMARATHON167f75c03047: Fixes #4824 - Use GroupRepository instead of AppRepository and PodRepository in… (authored by unterstein, committed by timcharper). · Explain WhyDec 12 2016, 8:19 PM
This revision was automatically updated to reflect the committed changes.
unterstein added inline comments.Dec 13 2016, 11:22 AM
src/main/scala/mesosphere/marathon/MarathonSchedulerActor.scala
397–398

thanks!

Revision Contents

Diff 1237

View Options

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

Show All 14 Lines
1515import mesosphere.marathon.core.health.HealthCheckManager
1616import mesosphere.marathon.core.heartbeat._
1717import mesosphere.marathon.core.launchqueue.LaunchQueue
1818import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1919import mesosphere.marathon.core.task.termination.KillService
2020import mesosphere.marathon.core.task.tracker.InstanceTracker
2121import mesosphere.marathon.io.storage.StorageProvider
2222import mesosphere.marathon.metrics.Metrics
23import mesosphere.marathon.storage.repository.{ DeploymentRepository, GroupRepository, ReadOnlyAppRepository, ReadOnlyPodRepository }
23import mesosphere.marathon.storage.repository.{ DeploymentRepository, GroupRepository }
2424import mesosphere.marathon.upgrade.DeploymentManager
2525import mesosphere.util.state._
2626import org.apache.mesos.Scheduler
2727import org.slf4j.LoggerFactory
2828
2929import scala.concurrent.duration.FiniteDuration
3030import scala.util.control.NonFatal
3131
▲ Show 20 Lines • Show All 66 Lines • ▼ Show 20 Line(s)
9898
9999 @Named("schedulerActor")
100100 @Provides
101101 @Singleton
102102 @Inject
103103 @SuppressWarnings(Array("MaxParameters"))
104104 def provideSchedulerActor(
105105 system: ActorSystem,
106 appRepository: ReadOnlyAppRepository,
107 podRepository: ReadOnlyPodRepository,
108106 groupRepository: GroupRepository,
109107 deploymentRepository: DeploymentRepository,
110108 healthCheckManager: HealthCheckManager,
111109 instanceTracker: InstanceTracker,
112110 killService: KillService,
113111 launchQueue: LaunchQueue,
114112 driverHolder: MarathonSchedulerDriverHolder,
115113 electionService: ElectionService,
116114 storage: StorageProvider,
117115 eventBus: EventStream,
118116 readinessCheckExecutor: ReadinessCheckExecutor,
119117 @Named(ModuleNames.HISTORY_ACTOR_PROPS) historyActorProps: Props)(implicit mat: Materializer): ActorRef = {
120118 val supervision = OneForOneStrategy() {
121119 case NonFatal(_) => Restart
122120 }
123121
124122 import scala.concurrent.ExecutionContext.Implicits.global
125123
126124 def createSchedulerActions(schedulerActor: ActorRef): SchedulerActions = {
127125 new SchedulerActions(
128 appRepository,
129 podRepository,
130126 groupRepository,
131127 healthCheckManager,
132128 instanceTracker,
133129 launchQueue,
134130 eventBus,
135131 schedulerActor,
136132 killService)
137133 }
▲ Show 20 Lines • Show All 58 Lines • Show Last 20 Lines
View Options

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

11package mesosphere.marathon
22
33import akka.actor._
44import akka.event.{ EventStream, LoggingReceive }
55import akka.stream.Materializer
66import mesosphere.marathon.MarathonSchedulerActor.ScaleRunSpec
77import mesosphere.marathon.core.election.{ ElectionService, LocalLeadershipEvent }
88import mesosphere.marathon.core.event.{ AppTerminatedEvent, DeploymentFailed, DeploymentSuccess }
99import mesosphere.marathon.core.health.HealthCheckManager
1010import mesosphere.marathon.core.instance.Instance
1111import mesosphere.marathon.core.instance.Instance.AgentInfo
1212import mesosphere.marathon.core.launchqueue.LaunchQueue
1313import mesosphere.marathon.core.task.Task
1414import mesosphere.marathon.core.task.termination.{ KillReason, KillService }
1515import mesosphere.marathon.core.task.tracker.InstanceTracker
1616import mesosphere.marathon.state.{ PathId, RunSpec }
17import mesosphere.marathon.storage.repository.{ GroupRepository, ReadOnlyAppRepository, ReadOnlyPodRepository }
17import mesosphere.marathon.storage.repository.GroupRepository
1818import mesosphere.marathon.stream._
1919import mesosphere.marathon.upgrade.DeploymentManager._
2020import mesosphere.marathon.upgrade.{ DeploymentManager, DeploymentPlan, ScalingProposition }
2121import mesosphere.mesos.Constraints
2222import org.apache.mesos
2323import org.apache.mesos.Protos.{ Status, TaskState }
2424import org.apache.mesos.SchedulerDriver
2525import org.slf4j.LoggerFactory
▲ Show 20 Lines • Show All 325 Lines • ▼ Show 20 Line(s)
351351 }
352352
353353 f
354354 }
355355 }
356356}
357357
358358class SchedulerActions(
359 appRepository: ReadOnlyAppRepository,
360 podRepository: ReadOnlyPodRepository,
361359 groupRepository: GroupRepository,
362360 healthCheckManager: HealthCheckManager,
363361 instanceTracker: InstanceTracker,
364362 launchQueue: LaunchQueue,
365363 eventBus: EventStream,
366364 val schedulerActor: ActorRef,
367 val killService: KillService)(implicit ec: ExecutionContext, mat: Materializer) {
365 val killService: KillService)(implicit ec: ExecutionContext) {
368366
369367 private[this] val log = LoggerFactory.getLogger(getClass)
370368
371369 // TODO move stuff below out of the scheduler
372370
373371 def startRunSpec(runSpec: RunSpec): Unit = {
374372 log.info(s"Starting runSpec ${runSpec.id}")
375373 scale(runSpec)
Show All 15 Lines
391389 launchQueue.resetDelay(runSpec)
392390
393391 // The tasks will be removed from the InstanceTracker when their termination
394392 // was confirmed by Mesos via a task update.
395393
396394 eventBus.publish(AppTerminatedEvent(runSpec.id))
397395 }
398396 }
399397
400 def scaleRunSpec(): Future[Unit] = {
398 def scaleRunSpec(): Unit = {
timcharperUnsubmitted
Not Done

I vote we change the method signature here to Unit, and then use future.foreach.

meichstedtUnsubmitted
Not Done

Makes sense to me too. The only usage of this method doesn't even care.

timcharperUnsubmitted
Not Done

I went ahead and did this and updated the patch. If this was naughty behavior then I’m sorry!

untersteinAuthorUnsubmitted
Not Done

thanks!

401 appRepository.ids().concat(podRepository.ids()).runWith(Sink.set).andThen {
402 case Success(ids) => for (id <- ids) schedulerActor ! ScaleRunSpec(id)
403 case Failure(t) => log.warn("Failed to get task names", t)
404 }.map(_ => ())
399 groupRepository.root().foreach { root =>
400 root.transitiveRunSpecs.foreach(spec => schedulerActor ! ScaleRunSpec(spec.id))
405401 }
402 }
406403
407404 /**
408405 * Make sure all runSpecs are running the configured amount of tasks.
409406 *
410407 * Should be called some time after the framework re-registers,
411408 * to give Mesos enough time to deliver task updates.
412409 *
413410 * @param driver scheduler driver
414411 */
415412 def reconcileTasks(driver: SchedulerDriver): Future[Status] = {
416 appRepository.ids().concat(podRepository.ids()).runWith(Sink.set).flatMap { runSpecIds =>
413 groupRepository.root().flatMap { root =>
414 val runSpecIds = root.transitiveRunSpecsById.keySet
417415 instanceTracker.instancesBySpec().map { instances =>
418416 val knownTaskStatuses = runSpecIds.flatMap { runSpecId =>
419417 TaskStatusCollector.collectTaskStatusFor(instances.specInstances(runSpecId))
420418 }
421419
422420 (instances.allSpecIdsWithInstances -- runSpecIds).foreach { unknownId =>
423421 log.warn(
424422 s"RunSpec $unknownId exists in InstanceTracker, but not store. " +
▲ Show 20 Lines • Show All 72 Lines • ▼ Show 20 Line(s)
497495 def scale(runSpecId: PathId): Future[Unit] = {
498496 runSpecById(runSpecId).map {
499497 case Some(runSpec) => scale(runSpec)
500498 case _ => log.warn(s"RunSpec $runSpecId does not exist. Not scaling.")
501499 }
502500 }
503501
504502 def runSpecById(id: PathId): Future[Option[RunSpec]] = {
505 appRepository.get(id).flatMap {
506 case Some(app) => Future.successful(Some(app))
507 case None => podRepository.get(id)
508 }
503 groupRepository.root().map(_.transitiveRunSpecsById.get(id))
509504 }
510505}
511506
512507/**
513508 * Provides means to collect Mesos TaskStatus information for reconciliation.
514509 */
515510object TaskStatusCollector {
516511 def collectTaskStatusFor(instances: Seq[Instance]): Seq[mesos.Protos.TaskStatus] = {
Show All 26 Lines
View Options

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

Show All 15 Lines
1616import mesosphere.marathon.core.launcher.impl.LaunchQueueTestHelper
1717import mesosphere.marathon.core.launchqueue.LaunchQueue
1818import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
1919import mesosphere.marathon.core.task.KillServiceMock
2020import mesosphere.marathon.core.task.tracker.InstanceTracker
2121import mesosphere.marathon.io.storage.StorageProvider
2222import mesosphere.marathon.state.PathId._
2323import mesosphere.marathon.state._
24import mesosphere.marathon.storage.repository.{ AppRepository, DeploymentRepository, FrameworkIdRepository, GroupRepository, TaskFailureRepository, _ }
24import mesosphere.marathon.storage.repository.{ DeploymentRepository, FrameworkIdRepository, GroupRepository, TaskFailureRepository }
2525import mesosphere.marathon.stream._
2626import mesosphere.marathon.test.{ GroupCreation, MarathonActorSupport, MarathonSpec, Mockito }
2727import mesosphere.marathon.upgrade._
2828import org.apache.mesos.Protos.{ Status, TaskStatus }
2929import org.apache.mesos.SchedulerDriver
3030import org.scalatest.{ BeforeAndAfter, FunSuiteLike, GivenWhenThen, Matchers }
3131
3232import scala.collection.immutable.Set
Show All 27 Lines
6060 }
6161
6262 test("Reconcile orphan instance of unknown app - instance should be killed") {
6363 val f = new Fixture
6464 import f._
6565 val app = AppDefinition(id = "/deleted-app".toPath, instances = 1)
6666 val orphanedInstance = TestInstanceBuilder.newBuilder(app.id).addTaskRunning().getInstance()
6767
68 appRepo.ids() returns Source.empty[PathId]
68 groupRepo.root() returns Future.successful(createRootGroup())
6969 instanceTracker.instancesBySpec()(any[ExecutionContext]) returns Future.successful(InstanceTracker.InstancesBySpec.of(InstanceTracker.SpecInstances.forInstances(app.id, Seq(orphanedInstance))))
70 appRepo.get(app.id) returns Future.successful(None)
7170
7271 val schedulerActor = createActor()
7372 try {
7473 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
7574 schedulerActor ! ReconcileTasks
7675
7776 expectMsg(5.seconds, TasksReconciled)
7877
7978 awaitAssert({
8079 killService.killed should contain (orphanedInstance.instanceId)
8180 }, 5.seconds, 10.millis)
8281 } finally {
8382 stopActor(schedulerActor)
8483 }
8584 }
8685
8786 test("Terminal tasks should not be submitted in reconciliation") {
8887 val f = new Fixture
8988 import f._
9089 val app = AppDefinition(id = "/test-app".toPath, instances = 1)
9190 val instance = TestInstanceBuilder.newBuilder(app.id).addTaskUnreachable(containerName = Some("unreachable")).addTaskRunning().addTaskGone(containerName = Some("gone")).getInstance()
9291
93 appRepo.ids() returns Source.single(app.id)
92 groupRepo.root() returns Future.successful(createRootGroup(apps = Map(app.id -> app)))
9493 instanceTracker.instancesBySpec()(any[ExecutionContext]) returns Future.successful(InstanceTracker.InstancesBySpec.of(InstanceTracker.SpecInstances.forInstances(app.id, Seq(instance))))
95 appRepo.get(app.id) returns Future.successful(Some(app))
9694
9795 val schedulerActor = createActor()
9896 try {
9997 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
10098 schedulerActor ! ReconcileTasks
10199
102100 expectMsg(5.seconds, TasksReconciled)
103101
Show All 19 Lines
123121 .addTaskFailed(containerName = Some("failed"))
124122 .addTaskFinished(containerName = Some("finished"))
125123 .addTaskKilled(containerName = Some("killed"))
126124 .addTaskGone(containerName = Some("gone"))
127125 .addTaskDropped(containerName = Some("dropped"))
128126 .addTaskUnknown(containerName = Some("unknown"))
129127 .getInstance()
130128
131 appRepo.ids() returns Source.single(app.id)
129 groupRepo.root() returns Future.successful(createRootGroup(apps = Map(app.id -> app)))
132130 instanceTracker.instancesBySpec()(any[ExecutionContext]) returns Future.successful(InstanceTracker.InstancesBySpec.of(InstanceTracker.SpecInstances.forInstances("nope".toPath, Seq(instance))))
133 appRepo.get(app.id) returns Future.successful(Some(app))
134131
135132 val schedulerActor = createActor()
136133 try {
137134 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
138135 schedulerActor ! ReconcileTasks
139136
140137 expectMsg(5.seconds, TasksReconciled)
141138
Show All 20 Lines
162159 .addTaskCreated(containerName = Some("created"))
163160 .addTaskKilling(containerName = Some("killing"))
164161 .addTaskRunning(containerName = Some("running"))
165162 .addTaskStaging(containerName = Some("staging"))
166163 .addTaskStarting(containerName = Some("starting"))
167164 .addTaskUnreachable(containerName = Some("unreachable"))
168165 .getInstance()
169166
170 appRepo.ids() returns Source.single(app.id)
167 groupRepo.root() returns Future.successful(createRootGroup(apps = Map(app.id -> app)))
171168 instanceTracker.instancesBySpec()(any[ExecutionContext]) returns Future.successful(InstanceTracker.InstancesBySpec.of(InstanceTracker.SpecInstances.forInstances("nope".toPath, Seq(instance))))
172 appRepo.get(app.id) returns Future.successful(Some(app))
173169
174170 val schedulerActor = createActor()
175171 try {
176172 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
177173 schedulerActor ! ReconcileTasks
178174
179175 expectMsg(5.seconds, TasksReconciled)
180176
Show All 13 Lines
194190 stopActor(schedulerActor)
195191 }
196192 }
197193
198194 test("ScaleApps") {
199195 val f = new Fixture
200196 import f._
201197 val app: AppDefinition = AppDefinition(id = "/test-app".toPath, instances = 1)
198
202199 val instances = Seq(TestInstanceBuilder.newBuilder(app.id).addTaskRunning().getInstance())
203200
204201 queue.get(app.id) returns Some(LaunchQueueTestHelper.zeroCounts)
205 appRepo.ids() returns Source.single(app.id)
206202 instanceTracker.specInstancesSync(app.id) returns Seq.empty[Instance]
207203 instanceTracker.instancesBySpecSync returns InstanceTracker.InstancesBySpec.of(InstanceTracker.SpecInstances.forInstances("nope".toPath, instances))
208204 instanceTracker.specInstancesSync("nope".toPath) returns instances
209 appRepo.get(app.id) returns Future.successful(Some(app))
205 groupRepo.root() returns Future.successful(createRootGroup(apps = Map(app.id -> app)))
210206 instanceTracker.countLaunchedSpecInstancesSync(app.id) returns 0
207 groupRepo.root() returns Future.successful(createRootGroup(apps = Map(app.id -> app)))
211208
212209 val schedulerActor = createActor()
213210 try {
214211 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
215212 schedulerActor ! ScaleRunSpecs
216213
217214 awaitAssert(verify(queue).add(app, 1), 5.seconds, 10.millis)
218215 } finally {
219216 stopActor(schedulerActor)
220217 }
221218 }
222219
223220 test("ScaleApp") {
224221 val f = new Fixture
225222 import f._
226223 val app = AppDefinition(id = "test-app".toPath, instances = 1)
227224
228225 queue.get(app.id) returns Some(LaunchQueueTestHelper.zeroCounts)
229 appRepo.ids() returns Source.single(app.id)
226 groupRepo.root() returns Future.successful(createRootGroup(apps = Map(app.id -> app)))
230227 instanceTracker.specInstancesSync(app.id) returns Seq.empty[Instance]
231228
232 appRepo.get(app.id) returns Future.successful(Some(app))
233229 instanceTracker.countLaunchedSpecInstancesSync(app.id) returns 0
234230
235231 val schedulerActor = createActor()
236232 try {
237233 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
238234 schedulerActor ! ScaleRunSpec("test-app".toPath)
239235
240236 awaitAssert(verify(queue).add(app, 1), 5.seconds, 10.millis)
Show All 16 Lines
257253 instance.runSpecId,
258254 Condition.Failed,
259255 instance
260256 )
261257
262258 f.killService.customStatusUpdates.put(instance.instanceId, Seq(instanceChangedEvent))
263259
264260 queue.get(app.id) returns Some(LaunchQueueTestHelper.zeroCounts)
265 appRepo.ids() returns Source.single(app.id)
261 groupRepo.root() returns Future.successful(createRootGroup(apps = Map(app.id -> app)))
266262 instanceTracker.specInstancesLaunchedSync(app.id) returns Seq(instance)
267263
268 appRepo.get(app.id) returns (Future.successful(Some(app)), Future.successful(Some(app.copy(instances = 0))))
269264 instanceTracker.specInstancesSync(org.mockito.Matchers.eq(app.id)) returns Seq()
270265
271 appRepo.store(any) returns Future.successful(Done)
272
273266 val schedulerActor = createActor()
274267 try {
275268 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
276269 schedulerActor ! KillTasks(app.id, Seq(instance))
277270
278271 expectMsg(5.seconds, TasksKilled(app.id, Seq(instance.instanceId)))
279272
280273 val Some(taskFailureEvent) = TaskFailure.FromInstanceChangedEvent(instanceChangedEvent)
281274 awaitAssert(verify(taskFailureEventRepository, times(1)).store(taskFailureEvent), 5.seconds, 10.millis)
282275
283276 // KillTasks does no longer scale
284 verify(appRepo, times(0)).store(any[AppDefinition])
277 f.killService.numKilled shouldBe 1 // 1 kill was scheduled a few lines above
285278 } finally {
286279 stopActor(schedulerActor)
287280 }
288281 }
289282
290283 test("Kill tasks") {
291284 val f = new Fixture
292285 import f._
293286 val app = AppDefinition(id = "/test-app".toPath, instances = 1)
294287 val instanceA = TestInstanceBuilder.newBuilderWithLaunchedTask(app.id).getInstance()
295288
296289 queue.get(app.id) returns Some(LaunchQueueTestHelper.zeroCounts)
297 appRepo.ids() returns Source.single(app.id)
290 groupRepo.root() returns Future.successful(createRootGroup(apps = Map(app.id -> app)))
298291 instanceTracker.specInstancesLaunchedSync(app.id) returns Seq(instanceA)
299292
300 appRepo.get(app.id) returns (
301 Future.successful(Some(app)),
302 Future.successful(Some(app.copy(instances = 0))))
303293 instanceTracker.countLaunchedSpecInstancesSync(app.id) returns 0
304294 instanceTracker.specInstancesSync(org.mockito.Matchers.eq(app.id)) returns Seq()
305 appRepo.store(any) returns Future.successful(Done)
306295
307296 val schedulerActor = createActor()
308297 try {
309298 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
310299 schedulerActor ! KillTasks(app.id, Seq(instanceA))
311300
312301 expectMsg(5.seconds, TasksKilled(app.id, List(instanceA.instanceId)))
313302
▲ Show 20 Lines • Show All 88 Lines • ▼ Show 20 Line(s)
402391 instances = 2,
403392 upgradeStrategy = UpgradeStrategy(0.5),
404393 versionInfo = VersionInfo.forNewConfig(Timestamp(0))
405394 )
406395 val rootGroup = createRootGroup(groups = Set(createGroup(PathId("/foo/bar"), Map(app.id -> app))))
407396
408397 val plan = DeploymentPlan(createRootGroup(), rootGroup)
409398
410 appRepo.store(any) returns Future.successful(Done)
411 appRepo.get(app.id) returns Future.successful(None)
412399 instanceTracker.specInstancesLaunchedSync(app.id) returns Seq.empty[Instance]
413400 instanceTracker.specInstancesSync(app.id) returns Seq.empty[Instance]
414 appRepo.delete(app.id) returns Future.successful(Done)
401 groupRepo.root() returns Future.successful(rootGroup)
415402
416403 val schedulerActor = createActor()
417404 try {
418405 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
419406 schedulerActor ! Deploy(plan)
420407
421408 expectMsgType[DeploymentStarted]
422409
▲ Show 20 Lines • Show All 57 Lines • ▼ Show 20 Line(s)
480467 test("Forced deployment") {
481468 val f = new Fixture
482469 import f._
483470 val app = AppDefinition(id = PathId("app1"), cmd = Some("cmd"), instances = 2, upgradeStrategy = UpgradeStrategy(0.5))
484471 val rootGroup = createRootGroup(groups = Set(createGroup(PathId("/foo/bar"), Map(app.id -> app))))
485472
486473 val plan = DeploymentPlan(createRootGroup(), rootGroup, id = Some("d1"))
487474
488 appRepo.store(any) returns Future.successful(Done)
489 appRepo.get(app.id) returns Future.successful(None)
490475 instanceTracker.specInstancesLaunchedSync(app.id) returns Seq.empty[Instance]
491476 instanceTracker.specInstancesSync(app.id) returns Seq.empty[Instance]
492 appRepo.delete(app.id) returns Future.successful(Done)
477 groupRepo.root() returns Future.successful(rootGroup)
493478
494479 val schedulerActor = createActor()
495480 try {
496481 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
497482 schedulerActor ! Deploy(plan)
498483
499484 expectMsgType[DeploymentStarted](10.seconds)
500485
Show All 10 Lines
511496 val f = new Fixture
512497 import f._
513498 val actions = mock[SchedulerActions]
514499 val actionsFactory: ActorRef => SchedulerActions = _ => actions
515500 val schedulerActor = createActor(Some(actionsFactory))
516501
517502 val reconciliationPromise = Promise[Status]()
518503 actions.reconcileTasks(any) returns reconciliationPromise.future
519 appRepo.ids() returns Source.empty
504 groupRepo.root() returns Future.successful(createRootGroup())
520505
521506 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
522507
523508 schedulerActor ! MarathonSchedulerActor.ReconcileTasks
524509 schedulerActor ! MarathonSchedulerActor.ReconcileTasks
525510
526511 reconciliationPromise.success(Status.DRIVER_RUNNING)
527512
Show All 9 Lines
537522 test("Concurrent reconciliation check is not preventing sequential calls") {
538523 val f = new Fixture
539524 import f._
540525 val actions = mock[SchedulerActions]
541526 val actionsFactory: ActorRef => SchedulerActions = _ => actions
542527 val schedulerActor = createActor(Some(actionsFactory))
543528
544529 actions.reconcileTasks(any) returns Future.successful(Status.DRIVER_RUNNING)
545 appRepo.ids() returns Source.empty
530 groupRepo.root() returns Future.successful(createRootGroup())
546531
547532 schedulerActor ! LocalLeadershipEvent.ElectedAsLeader
548533
549534 schedulerActor ! MarathonSchedulerActor.ReconcileTasks
550535 expectMsg(MarathonSchedulerActor.TasksReconciled)
551536
552537 schedulerActor ! MarathonSchedulerActor.ReconcileTasks
553538 expectMsg(MarathonSchedulerActor.TasksReconciled)
554539
555540 schedulerActor ! MarathonSchedulerActor.ReconcileTasks
556541 expectMsg(MarathonSchedulerActor.TasksReconciled)
557542
558543 verify(actions, times(3)).reconcileTasks(any[SchedulerDriver])
559544 }
560545
561546 class Fixture {
562 val appRepo: AppRepository = mock[AppRepository]
563 val podRepo: ReadOnlyPodRepository = mock[ReadOnlyPodRepository]
564547 val groupRepo: GroupRepository = mock[GroupRepository]
565548 val deploymentRepo: DeploymentRepository = mock[DeploymentRepository]
566549 val hcManager: HealthCheckManager = mock[HealthCheckManager]
567550 val instanceTracker: InstanceTracker = mock[InstanceTracker]
568551 val killService = new KillServiceMock(system)
569552 val queue: LaunchQueue = mock[LaunchQueue]
570553 val frameworkIdRepo: FrameworkIdRepository = mock[FrameworkIdRepository]
571554 val driver: SchedulerDriver = mock[SchedulerDriver]
572555 val holder: MarathonSchedulerDriverHolder = new MarathonSchedulerDriverHolder
573556 holder.driver = Some(driver)
574557 val storage: StorageProvider = mock[StorageProvider]
575558 val taskFailureEventRepository: TaskFailureRepository = mock[TaskFailureRepository]
576559 val electionService: ElectionService = mock[ElectionService]
577560 val schedulerActions: ActorRef => SchedulerActions = ref => {
578561 new SchedulerActions(
579 appRepo, podRepo, groupRepo, hcManager, instanceTracker, queue, new EventStream(system), ref, killService)(system.dispatcher, mat)
562 groupRepo, hcManager, instanceTracker, queue, new EventStream(system), ref, killService)(system.dispatcher)
580563 }
581564 val conf: UpgradeConfig = mock[UpgradeConfig]
582565 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
583566 val deploymentManagerProps: SchedulerActions => Props = schedulerActions => Props(new DeploymentManager(
584567 instanceTracker,
585568 killService,
586569 queue,
587570 schedulerActions,
Show All 26 Lines
614597 watch(ref)
615598 system.stop(ref)
616599 expectTerminated(ref)
617600 }
618601
619602 deploymentRepo.store(any) returns Future.successful(Done)
620603 deploymentRepo.delete(any) returns Future.successful(Done)
621604 deploymentRepo.all() returns Source.empty
622 appRepo.all() returns Source.empty
623 podRepo.ids() returns Source.empty[PathId]
624605 groupRepo.root() returns Future.successful(createRootGroup())
625606 queue.get(any[PathId]) returns None
626607 instanceTracker.countLaunchedSpecInstancesSync(any[PathId]) returns 0
627608 conf.killBatchCycle returns 1.seconds
628609 conf.killBatchSize returns 100
629610 }
630611
631612 implicit val defaultTimeout: Timeout = 30.seconds
632613}
View Options

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

11package mesosphere.marathon
22
3import akka.Done
4import akka.stream.scaladsl.Source
53import akka.testkit.TestProbe
64import mesosphere.marathon.core.base.ConstantClock
75import mesosphere.marathon.core.condition.Condition
86import mesosphere.marathon.core.health.HealthCheckManager
97import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
108import mesosphere.marathon.core.launcher.impl.LaunchQueueTestHelper
119import mesosphere.marathon.core.launchqueue.LaunchQueue
1210import mesosphere.marathon.core.task.termination.{ KillReason, KillService }
1311import mesosphere.marathon.core.task.tracker.InstanceTracker
1412import mesosphere.marathon.core.task.tracker.InstanceTracker.{ InstancesBySpec, SpecInstances }
15import mesosphere.marathon.state.{ AppDefinition, PathId, Timestamp }
16import mesosphere.marathon.storage.repository.{ AppRepository, GroupRepository, ReadOnlyPodRepository }
13import mesosphere.marathon.state.{ AppDefinition, RootGroup, PathId, Timestamp }
14import mesosphere.marathon.storage.repository.GroupRepository
1715import mesosphere.marathon.stream._
1816import mesosphere.marathon.test.{ MarathonActorSupport, MarathonSpec, MarathonTestHelper, Mockito }
1917import org.apache.mesos.SchedulerDriver
2018import org.mockito.Mockito.verifyNoMoreInteractions
2119import org.scalatest.concurrent.{ PatienceConfiguration, ScalaFutures }
2220import org.scalatest.time.{ Millis, Span }
2321import org.scalatest.{ GivenWhenThen, Matchers }
2422
2523import scala.concurrent.Future
2624import scala.concurrent.duration._
2725
2826class SchedulerActionsTest
2927 extends MarathonActorSupport
3028 with MarathonSpec
3129 with Matchers
3230 with Mockito
3331 with ScalaFutures
3432 with GivenWhenThen {
3533
3634 test("Reset rate limiter if application is stopped") {
3735 val f = new Fixture
3836 val app = AppDefinition(id = PathId("/myapp"))
3937
40 f.appRepo.delete(app.id) returns Future.successful(Done)
4138 f.instanceTracker.specInstances(eq(app.id))(any) returns Future.successful(Seq.empty[Instance])
4239
4340 f.scheduler.stopRunSpec(app).futureValue(1.second)
4441
4542 verify(f.queue).purge(app.id)
4643 verify(f.queue).resetDelay(app)
4744 verifyNoMoreInteractions(f.queue)
4845 }
4946
5047 test("Task reconciliation sends known running and staged tasks and empty list") {
5148 val f = new Fixture
5249 val app = AppDefinition(id = PathId("/myapp"))
50 val rootGroup: RootGroup = RootGroup(apps = Map((app.id, app)))
5351 val runningInstance = TestInstanceBuilder.newBuilder(app.id).addTaskRunning().getInstance()
5452 val stagedInstance = TestInstanceBuilder.newBuilder(app.id).addTaskStaged().getInstance()
5553
5654 val stagedInstanceWithSlaveId = TestInstanceBuilder.newBuilder(app.id)
5755 .addTaskWithBuilder().taskStaged().build()
5856 .withAgentInfo(agentId = Some("slave 1"))
5957 .getInstance()
6058
6159 val instances = Seq(runningInstance, stagedInstance, stagedInstanceWithSlaveId)
6260 f.instanceTracker.instancesBySpec() returns Future.successful(InstancesBySpec.of(SpecInstances.forInstances(app.id, instances)))
63 f.appRepo.ids() returns Source.single(app.id)
61 f.groupRepo.root() returns Future.successful(rootGroup)
6462
6563 f.scheduler.reconcileTasks(f.driver).futureValue(5.seconds)
6664
6765 verify(f.driver).reconcileTasks(Set(
6866 runningInstance,
6967 stagedInstance,
7068 stagedInstanceWithSlaveId
7169 ).flatMap(_.tasksMap.values).flatMap(_.status.mesosStatus))
7270 verify(f.driver).reconcileTasks(java.util.Arrays.asList())
7371 }
7472
7573 test("Task reconciliation only one empty list, when no tasks are present in Marathon") {
7674 val f = new Fixture
7775
7876 f.instanceTracker.instancesBySpec() returns Future.successful(InstancesBySpec.empty)
79 f.appRepo.ids() returns Source.empty
77 f.groupRepo.root() returns Future.successful(RootGroup())
8078
8179 f.scheduler.reconcileTasks(f.driver).futureValue
8280
8381 verify(f.driver, times(1)).reconcileTasks(java.util.Arrays.asList())
8482 }
8583
8684 test("Kill orphaned task") {
8785 val f = new Fixture
8886 val app = AppDefinition(id = PathId("/myapp"))
8987 val orphanedApp = AppDefinition(id = PathId("/orphan"))
9088 val instance = TestInstanceBuilder.newBuilder(app.id).addTaskRunning().getInstance()
9189 val orphanedInstance = TestInstanceBuilder.newBuilder(orphanedApp.id).addTaskRunning().getInstance()
9290
9391 val tasksOfApp = SpecInstances.forInstances(app.id, Seq(instance))
9492 val tasksOfOrphanedApp = SpecInstances.forInstances(orphanedApp.id, Seq(orphanedInstance))
9593
9694 f.instanceTracker.instancesBySpec() returns Future.successful(InstancesBySpec.of(tasksOfApp, tasksOfOrphanedApp))
97 f.appRepo.ids() returns Source.single(app.id)
95 val rootGroup: RootGroup = RootGroup(apps = Map((app.id, app)))
96 f.groupRepo.root() returns Future.successful(rootGroup)
9897
9998 f.scheduler.reconcileTasks(f.driver).futureValue(5.seconds)
10099
101100 verify(f.killService, times(1)).killInstance(orphanedInstance, KillReason.Orphaned)
102101 }
103102
104103 test("Scale up correctly in case of lost tasks (active queue)") {
105104 val f = new Fixture
▲ Show 20 Lines • Show All 179 Lines • ▼ Show 20 Line(s)
285284
286285 import scala.language.implicitConversions
287286 implicit def durationToPatienceConfigTimeout(d: FiniteDuration): PatienceConfiguration.Timeout = {
288287 PatienceConfiguration.Timeout(Span(d.toMillis, Millis))
289288 }
290289
291290 class Fixture {
292291 val queue = mock[LaunchQueue]
293 val appRepo = mock[AppRepository]
294 val podRepo: ReadOnlyPodRepository = mock[ReadOnlyPodRepository]
292 val groupRepo = mock[GroupRepository]
295293 val instanceTracker = mock[InstanceTracker]
296294 val driver = mock[SchedulerDriver]
297295 val killService = mock[KillService]
298296 val clock = ConstantClock()
299297
300 podRepo.ids() returns Source.empty[PathId]
301
302298 val scheduler = new SchedulerActions(
303 appRepo,
304 podRepo,
305 mock[GroupRepository],
299 groupRepo,
306300 mock[HealthCheckManager],
307301 instanceTracker,
308302 queue,
309303 system.eventStream,
310304 TestProbe().ref,
311305 killService
312306 )
313307 }
314308
315309}