| zen-dog | |
| timcharper | |
| jeschkies | |
| jenkins |
During the replacement of GroupManagerActor to GroupManagerImpl, the following behavior was introduced:
When taking over leadership, a marathon instance will not load current root group from zookeeper, instead the old (probably outdated) root group will used to operate. This leads wrong behavior, because this root group is not in sync with the current zookeeper state.
This fix introduced a refresh cache implementation which is called during starting after taking leadership.
sbt integration:test
| Automatic diff as part of commit; lint not applicable. |
| Automatic diff as part of commit; unit tests not applicable. |
| src/main/scala/mesosphere/marathon/MarathonSchedulerService.scala | ||
|---|---|---|
| 199 | We do have two. The first one is quite fast which is refreshGroupCache with an infinite time out. The second one could probably mutates to a long calculation during do leadership preparation. You are able to define a dedicated timeout for the latter, therefore I would not combine those two to one async/await. | |
| src/main/scala/mesosphere/marathon/core/group/GroupManager.scala | ||
| 153 | sure =) | |
| src/main/scala/mesosphere/marathon/core/group/impl/GroupManagerImpl.scala | ||
| 202 | Unfortunately yes :( | |
| src/test/scala/mesosphere/marathon/integration/LeaderIntegrationTest.scala | ||
| 183 | ok | |
| 216 | clone and owned this from the other few tests in this file which are doing this the same way ¯\_(ツ)_/¯ | |
| src/main/scala/mesosphere/marathon/storage/repository/GroupRepositoryImpl.scala | ||
|---|---|---|
| 246 | I think there's an implicit assumption here that refreshGroupCache will never be invoked at the same time as any other func in this class. If that's true then please clearly document this fact. If that's not the long term intent, then please protect rootFuture here with a lock like everywhere else in this class. | |
@unterstein can you explain more the motivation for refreshGroupCache? It seems like this method is only called when we are elected leader?
Could this same thing be expressed by making some promise in rootGroup, and then postponing all requests to load state until after this promise is completed? This way the order of operations dependencies are explicit?
I am not as familiar with this code and did not dive in deeply, so, I could be barking up the wrong tree :)
| Commit | Tree | Parents | Author | Summary | Date |
|---|---|---|---|---|---|
| 0f60e74c637e | f489ebe08a2d | ed656160b6cd | Johannes Unterstein | feedback | Jun 12 2017, 5:28 PM |
| ed656160b6cd | 2eff540ad85f | 76b0386d1257 | Johannes Unterstein | @SuppressWarnings(Array("all")) // async/await | Jun 12 2017, 12:45 PM |
| 76b0386d1257 | a9982c14028f | aeb118aabbe0 | Johannes Unterstein | adapted feedback | Jun 12 2017, 11:50 AM |
| aeb118aabbe0 | 0503c1efc85f | b8e70ae00202 | Johannes Unterstein | more SI | Jun 9 2017, 7:27 PM |
| b8e70ae00202 | 9f7eab7fab38 | df0578fe910e | Johannes Unterstein | Future[Done] | Jun 9 2017, 6:41 PM |
| df0578fe910e | 8c72454ecbaa | 42613e85bbea | Johannes Unterstein | updated integration test | Jun 9 2017, 12:03 PM |
| 42613e85bbea | b5afe9ac4347 | 56cb0cc17daf | Johannes Unterstein | added SI test | Jun 9 2017, 11:54 AM |
| 56cb0cc17daf | 12713f5613a5 | b10e45b1a033 | Johannes Unterstein | [WIP MARATHON-7458] reset root group cache when elected as leader (Show More…) | Jun 9 2017, 9:29 AM |
| b10e45b1a033 | 51151285abe1 | 7bd25aec816e | Johannes Unterstein | added test | Jun 9 2017, 9:18 AM |
| 7bd25aec816e | ee5384ccf0ec | 774657076c2f | Johannes Unterstein | reset root group cache when elected as leader | Jun 9 2017, 8:31 AM |
| Show First 20 Lines • Show All 190 Lines • ▼ Show 20 Line(s) | ||||
| 191 | 191 | // signals that the driver was stopped manually (as opposed to crashing mid-process) | ||
|---|---|---|---|---|
| 192 | 192 | driver = None | ||
| 193 | 193 | } | ||
| 194 | 194 | |||
| 195 | 195 | //End Service interface | ||
| 196 | 196 | |||
| 197 | 197 | //Begin ElectionCandidate interface | ||
| 198 | 198 | |||
| 199 | 199 | override def startLeadership(): Unit = synchronized { | ||
We do have two. The first one is quite fast which is refreshGroupCache with an infinite time out. The second one could probably mutates to a long calculation during do leadership preparation. You are able to define a dedicated timeout for the latter, therefore I would not combine those two to one async/await. | ||||
| 200 | 200 | log.info("As new leader running the driver") | ||
| 201 | 201 | |||
| 202 | // refresh group repository cache | |||
| 203 | Await.result(groupManager.refreshGroupCache(), Duration.Inf) | |||
| 204 | ||||
| 202 | 205 | // execute tasks, only the leader is allowed to | ||
| 203 | 206 | migration.migrate() | ||
| 204 | 207 | |||
| 205 | 208 | // run all pre-driver callbacks | ||
| 206 | 209 | log.info(s"""Call preDriverStarts callbacks on ${prePostDriverCallbacks.mkString(", ")}""") | ||
| 207 | 210 | Await.result( | ||
| 208 | 211 | Future.sequence(prePostDriverCallbacks.map(_.preDriverStarts)), | ||
| 209 | 212 | config.onElectedPrepareTimeout().millis | ||
| ▲ Show 20 Lines • Show All 94 Lines • Show Last 20 Lines | ||||
| 1 | 1 | package mesosphere.marathon | ||
|---|---|---|---|---|
| 2 | 2 | package core.group | ||
| 3 | 3 | |||
| 4 | 4 | import java.time.OffsetDateTime | ||
| 5 | 5 | |||
| 6 | import akka.NotUsed | |||
| 6 | import akka.{ Done, NotUsed } | |||
| 7 | 7 | import akka.stream.scaladsl.Source | ||
| 8 | 8 | import mesosphere.marathon.core.instance.Instance | ||
| 9 | 9 | import mesosphere.marathon.core.pod.PodDefinition | ||
| 10 | 10 | import mesosphere.marathon.state.{ AppDefinition, Group, PathId, RootGroup, RunSpec, Timestamp } | ||
| 11 | 11 | import mesosphere.marathon.core.deployment.DeploymentPlan | ||
| 12 | 12 | |||
| 13 | 13 | import scala.concurrent.Future | ||
| 14 | 14 | import scala.collection.immutable.Seq | ||
| ▲ Show 20 Lines • Show All 130 Lines • ▼ Show 20 Line(s) | ||||
| 145 | 145 | def updatePod( | ||
| 146 | 146 | podId: PathId, | ||
| 147 | 147 | fn: Option[PodDefinition] => PodDefinition, | ||
| 148 | 148 | version: Timestamp = Timestamp.now(), | ||
| 149 | 149 | force: Boolean = false, | ||
| 150 | 150 | toKill: Seq[Instance] = Seq.empty | ||
| 151 | 151 | ): Future[DeploymentPlan] = updateRoot(podId.parent, _.updatePod(podId, fn, version), version, force, Map(podId -> toKill)) | ||
| 152 | 152 | |||
| 153 | /** | |||
| 154 | * Refresh the internal root group cache. When calling this function, the internal hold cached root group will be dropped | |||
| 155 | * and loaded when accessing the next time. | |||
| 156 | * | |||
| 157 | * @return Done if refresh was successful | |||
| 158 | */ | |||
| 159 | def refreshGroupCache(): Future[Done] | |||
| 153 | 160 | } | ||
| 1 | 1 | package mesosphere.marathon | ||
|---|---|---|---|---|
| 2 | 2 | package core.group.impl | ||
| 3 | 3 | |||
| 4 | 4 | import java.time.OffsetDateTime | ||
| 5 | 5 | import javax.inject.Provider | ||
| 6 | 6 | |||
| 7 | import akka.NotUsed | |||
| 8 | 7 | import akka.event.EventStream | ||
| 9 | 8 | import akka.stream.scaladsl.Source | ||
| 9 | import akka.{Done, NotUsed} | |||
| 10 | 10 | import com.typesafe.scalalogging.StrictLogging | ||
| 11 | 11 | import mesosphere.marathon.api.v2.Validation | ||
| 12 | 12 | import mesosphere.marathon.core.deployment.DeploymentPlan | ||
| 13 | import mesosphere.marathon.core.event.{ GroupChangeFailed, GroupChangeSuccess } | |||
| 14 | import mesosphere.marathon.core.group.{ GroupManager, GroupManagerConfig } | |||
| 13 | import mesosphere.marathon.core.event.{GroupChangeFailed, GroupChangeSuccess} | |||
| 14 | import mesosphere.marathon.core.group.{GroupManager, GroupManagerConfig} | |||
| 15 | 15 | import mesosphere.marathon.core.instance.Instance | ||
| 16 | 16 | import mesosphere.marathon.core.pod.PodDefinition | ||
| 17 | 17 | import mesosphere.marathon.state._ | ||
| 18 | 18 | import mesosphere.marathon.storage.repository.GroupRepository | ||
| 19 | 19 | import mesosphere.marathon.upgrade.GroupVersioningUtil | ||
| 20 | import mesosphere.marathon.util.{ LockedVar, WorkQueue } | |||
| 20 | import mesosphere.marathon.util.{LockedVar, WorkQueue} | |||
| 21 | 21 | |||
| 22 | 22 | import scala.async.Async._ | ||
| 23 | 23 | import scala.collection.immutable.Seq | ||
| 24 | 24 | import scala.collection.mutable | ||
| 25 | import scala.concurrent.{ ExecutionContext, Future } | |||
| 25 | import scala.concurrent.{ExecutionContext, Future} | |||
| 26 | 26 | import scala.util.control.NonFatal | ||
| 27 | import scala.util.{ Failure, Success } | |||
| 27 | import scala.util.{Failure, Success} | |||
| 28 | 28 | |||
| 29 | 29 | class GroupManagerImpl( | ||
| 30 | 30 | config: GroupManagerConfig, | ||
| 31 | 31 | initialRoot: RootGroup, | ||
| 32 | 32 | groupRepository: GroupRepository, | ||
| 33 | 33 | deploymentService: Provider[DeploymentService])(implicit eventStream: EventStream, ctx: ExecutionContext) extends GroupManager with StrictLogging { | ||
| 34 | 34 | /** | ||
| 35 | 35 | * All updates to root() should go through this workqueue and the maxConcurrent should always be "1" | ||
| ▲ Show 20 Lines • Show All 150 Lines • ▼ Show 20 Line(s) | ||||
| 186 | 186 | app.copy( | ||
| 187 | 187 | portDefinitions = mergeServicePortsAndPortDefinitions(app.portDefinitions, app.servicePorts) | ||
| 188 | 188 | ) | ||
| 189 | 189 | } | ||
| 190 | 190 | |||
| 191 | 191 | dynamicApps.foldLeft(to) { (rootGroup, app) => | ||
| 192 | 192 | rootGroup.updateApp(app.id, _ => app, app.version) | ||
| 193 | 193 | } | ||
| 194 | } | |||
| 195 | ||||
| 196 | @SuppressWarnings(Array("all")) // async/await | |||
| 197 | override def refreshGroupCache(): Future[Done] = async { | |||
| 198 | // propagation of reset group caches on repository is needed, | |||
| 199 | // because manager and repository are holding own caches | |||
| 200 | await(groupRepository.refreshGroupCache()) | |||
| 201 | val currentRoot = await(groupRepository.root()) | |||
| 202 | root := currentRoot | |||
| 203 | Done | |||
| 194 | 204 | } | ||
| 195 | 205 | } | ||
| Show First 20 Lines • Show All 174 Lines • ▼ Show 20 Line(s) | ||||
| 175 | 175 | get -> "wait" on the future, if it fails, create a new promise for it and actually fetch the root, | ||
|---|---|---|---|---|
| 176 | 176 | completing the promise with the fetch result. | ||
| 177 | 177 | set -> create a new promise for the root. If store succeeds, go update it, if it doesn't | ||
| 178 | 178 | complete the new future with the result of the previous root future. | ||
| 179 | 179 | |||
| 180 | 180 | This gives us read-after-write consistency. | ||
| 181 | 181 | */ | ||
| 182 | 182 | private val lock = RichLock() | ||
| 183 | private var rootFuture = Future.failed[RootGroup](new Exception("Root not yet loaded")) | |||
| 183 | private val rootNotLoaded: Future[RootGroup] = Future.failed[RootGroup](new Exception("Root not yet loaded")) | |||
| 184 | private var rootFuture: Future[RootGroup] = rootNotLoaded | |||
| 184 | 185 | private[storage] var beforeStore = Option.empty[(StoredGroup) => Future[Done]] | ||
| 185 | 186 | private val versionCache = TrieMap.empty[OffsetDateTime, Group] | ||
| 186 | 187 | |||
| 187 | 188 | private val storedRepo = { | ||
| 188 | 189 | @tailrec | ||
| 189 | 190 | def leafStore(store: PersistenceStore[K, C, S]): PersistenceStore[K, C, S] = store match { | ||
| 190 | 191 | case s: BasePersistenceStore[K, C, S] => s | ||
| 191 | 192 | case s: LoadTimeCachingPersistenceStore[K, C, S] => leafStore(s.store) | ||
| ▲ Show 20 Lines • Show All 42 Lines • ▼ Show 20 Line(s) | ||||
| 234 | 235 | await(storeRoot(root, Nil, Nil, Nil, Nil)) | ||
| 235 | 236 | root | ||
| 236 | 237 | } | ||
| 237 | 238 | promise.success(newRoot) | ||
| 238 | 239 | newRoot | ||
| 239 | 240 | case Success(root) => | ||
| 240 | 241 | root | ||
| 241 | 242 | } | ||
| 243 | } | |||
| 244 | ||||
| 245 | override def refreshGroupCache(): Future[Done] = { | |||
| 246 | rootFuture = rootNotLoaded | |||
I think there's an implicit assumption here that refreshGroupCache will never be invoked at the same time as any other func in this class. If that's true then please clearly document this fact. If that's not the long term intent, then please protect rootFuture here with a lock like everywhere else in this class. | ||||
| 247 | Future.successful(Done) | |||
| 242 | 248 | } | ||
| 243 | 249 | |||
| 244 | 250 | override def rootVersions(): Source[OffsetDateTime, NotUsed] = | ||
| 245 | 251 | storedRepo.versions(RootId) | ||
| 246 | 252 | |||
| 247 | 253 | @SuppressWarnings(Array("all")) // async/await | ||
| 248 | 254 | override def rootVersion(version: OffsetDateTime): Future[Option[RootGroup]] = { | ||
| 249 | 255 | async { | ||
| ▲ Show 20 Lines • Show All 127 Lines • Show Last 20 Lines | ||||
| Show All 27 Lines | |||
| 28 | 28 | ||
|---|---|---|---|
| 29 | 29 | trait GroupRepository { | |
| 30 | 30 | /** Fetch the root, returns an empty root if the root doesn't yet exist */ | |
| 31 | 31 | def root(): Future[RootGroup] | |
| 32 | 32 | /** List previous versions of the root */ | |
| 33 | 33 | def rootVersions(): Source[OffsetDateTime, NotUsed] | |
| 34 | 34 | /** Fetch a previous version of the root */ | |
| 35 | 35 | def rootVersion(version: OffsetDateTime): Future[Option[RootGroup]] | |
| 36 | /** Resets cached root group */ | ||
| 37 | def refreshGroupCache(): Future[Done] | ||
| 36 | 38 | ||
| 37 | 39 | /** | |
| 38 | 40 | * Store the root, new/updated apps and delete apps. fails if it could not | |
| 39 | 41 | * update the apps or the root, but deletion errors are ignored. | |
| 40 | 42 | */ | |
| 41 | 43 | def storeRoot(rootGroup: RootGroup, updatedApps: Seq[AppDefinition], deletedApps: Seq[PathId], | |
| 42 | 44 | updatedPods: Seq[PodDefinition], deletedPods: Seq[PathId]): Future[Done] | |
| 43 | 45 | ||
| ▲ Show 20 Lines • Show All 306 Lines • Show Last 20 Lines | |||
| 1 | 1 | package mesosphere.marathon | |
|---|---|---|---|
| 2 | 2 | ||
| 3 | 3 | import java.util.{ Timer, TimerTask } | |
| 4 | 4 | ||
| 5 | import akka.Done | ||
| 5 | 6 | import akka.actor.ActorRef | |
| 6 | 7 | import akka.testkit.TestProbe | |
| 7 | 8 | import mesosphere.AkkaUnitTest | |
| 8 | 9 | import mesosphere.chaos.http.HttpConf | |
| 9 | 10 | import mesosphere.marathon.Protos.StorageVersion | |
| 10 | 11 | import mesosphere.marathon.core.base.RichRuntime | |
| 11 | 12 | import mesosphere.marathon.core.deployment.DeploymentManager | |
| 12 | 13 | import mesosphere.marathon.core.election.ElectionService | |
| ▲ Show 20 Lines • Show All 56 Lines • ▼ Show 20 Line(s) | |||
| 69 | 70 | val taskTracker: InstanceTracker = mock[InstanceTracker] | |
| 70 | 71 | val marathonScheduler: MarathonScheduler = mock[MarathonScheduler] | |
| 71 | 72 | val migration: Migration = mock[Migration] | |
| 72 | 73 | val schedulerActor: ActorRef = probe.ref | |
| 73 | 74 | val heartbeatActor: ActorRef = heartbeatProbe.ref | |
| 74 | 75 | val prePostDriverCallbacks: Seq[PrePostDriverCallback] = Seq.empty | |
| 75 | 76 | val mockTimer: Timer = mock[Timer] | |
| 76 | 77 | val deploymentManager: DeploymentManager = mock[DeploymentManager] | |
| 78 | |||
| 79 | groupManager.refreshGroupCache() returns Future.successful(Done) | ||
| 77 | 80 | } | |
| 78 | 81 | ||
| 79 | 82 | def driverFactory[T](provide: => SchedulerDriver): SchedulerDriverFactory = { | |
| 80 | 83 | new SchedulerDriverFactory { | |
| 81 | 84 | override def createDriver(): SchedulerDriver = provide | |
| 82 | 85 | } | |
| 83 | 86 | } | |
| 84 | 87 | ||
| ▲ Show 20 Lines • Show All 217 Lines • Show Last 20 Lines | |||
| 1 | 1 | package mesosphere.marathon | ||
|---|---|---|---|---|
| 2 | 2 | package integration | ||
| 3 | 3 | |||
| 4 | 4 | import mesosphere.AkkaIntegrationTest | ||
| 5 | 5 | import mesosphere.marathon.integration.facades.MarathonFacade | ||
| 6 | import mesosphere.marathon.integration.facades.MarathonFacade.extractDeploymentIds | |||
| 6 | 7 | import mesosphere.marathon.integration.setup._ | ||
| 8 | import mesosphere.marathon.raml.App | |||
| 9 | import mesosphere.marathon.state.PathId._ | |||
| 7 | 10 | |||
| 8 | 11 | import scala.concurrent.duration._ | ||
| 9 | 12 | |||
| 10 | 13 | /** | ||
| 11 | 14 | * Do not add tests to this class. See the notes in [[NonDestructiveLeaderIntegrationTest]]. | ||
| 12 | 15 | */ | ||
| 13 | 16 | abstract class LeaderIntegrationTest extends AkkaIntegrationTest with MarathonClusterTest { | ||
| 14 | 17 | |||
| ▲ Show 20 Lines • Show All 128 Lines • ▼ Show 20 Line(s) | ||||
| 143 | 146 | } | ||
| 144 | 147 | |||
| 145 | 148 | // allow ZK session for former leader to timeout before proceeding | ||
| 146 | 149 | Thread.sleep((zkTimeout * 2.5).toLong) | ||
| 147 | 150 | |||
| 148 | 151 | And("the old leader should restart just fine") | ||
| 149 | 152 | leadingProcess.start().futureValue | ||
| 150 | 153 | } | ||
| 154 | } | |||
| 155 | } | |||
| 156 | } | |||
| 157 | ||||
| 158 | // Regression test for MARATHON-7458 | |||
| 159 | @IntegrationTest | |||
| 160 | class KeepAppsRunningDuringAbdicationIntegrationTest extends LeaderIntegrationTest { | |||
| 161 | ||||
| 162 | val zkTimeout = 2000L | |||
| 163 | override val marathonArgs: Map[String, String] = Map( | |||
| 164 | "zk_timeout" -> s"$zkTimeout" | |||
| 165 | ) | |||
| 166 | ||||
| 167 | override val numAdditionalMarathons = 2 | |||
| 168 | ||||
| 169 | "Abdicating a leader" should { | |||
| 170 | "keep all running apps alive" in { | |||
| 171 | ||||
| 172 | Given("a leader") | |||
| 173 | WaitTestSupport.waitUntil("a leader has been elected") { firstRunningProcess.client.leader().code == 200 } | |||
| 174 | ||||
| 175 | // pick the leader to communicate with because it's the only known survivor | |||
| 176 | val leader = firstRunningProcess.client.leader().value | |||
| 177 | val leadingProcess: LocalMarathon = leadingServerProcess(leader.leader) | |||
| 178 | val client = leadingProcess.client | |||
| 179 | ||||
| 180 | val app = App("/keepappsrunningduringabdicationintegrationtest", cmd = Some("sleep 1000")) | |||
| 181 | val result = marathon.createAppV2(app) | |||
| 182 | result.code should be(201) //Created | |||
| 183 | extractDeploymentIds(result) should have size 1 withClue "Deployment was not triggered" | |||
| 184 | waitForDeployment(result) | |||
| 185 | val oldInstances = client.tasks(app.id.toPath).value | |||
| 186 | oldInstances should have size 1 withClue "Required instance was not started" | |||
| 187 | ||||
| 188 | When("calling DELETE /v2/leader") | |||
| 189 | val abdicateResult = client.abdicate() | |||
| 190 | ||||
| 191 | Then("the request should be successful") | |||
| 192 | abdicateResult.code should be (200) withClue "Leader was not abdicated" | |||
| 193 | (abdicateResult.entityJson \ "message").as[String] should be ("Leadership abdicated") | |||
| 194 | ||||
| 195 | And("the leader must have died") | |||
| 196 | WaitTestSupport.waitUntil("the former leading marathon process dies", 30.seconds) { !leadingProcess.isRunning() } | |||
| 197 | leadingProcess.stop() // already stopped, but still need to clear old state | |||
| 198 | ||||
| 199 | And("the leader must have changed") | |||
| 200 | WaitTestSupport.waitUntil("the leader changes") { | |||
| 201 | val result = firstRunningProcess.client.leader() | |||
| 202 | result.code == 200 && result.value != leader | |||
| 203 | } | |||
| 204 | ||||
| 205 | val newLeader = firstRunningProcess.client.leader().value | |||
| 206 | val newLeadingProcess: LocalMarathon = leadingServerProcess(newLeader.leader) | |||
| 207 | val newClient = newLeadingProcess.client | |||
| 208 | ||||
| 209 | // we should have one survived instance | |||
| 210 | newClient.app(app.id.toPath).value.app.instances should be(1) withClue "Previously started app did not survive the abdication" | |||
| 211 | val newInstances = newClient.tasks(app.id.toPath).value | |||
| 212 | newInstances should have size 1 withClue "Previously started one instance did not survive the abdication" | |||
| 213 | newInstances(0).id should be (oldInstances(0).id) withClue "During abdication we started a new instance, instead keeping the old one." | |||
| 214 | ||||
| 215 | // allow ZK session for former leader to timeout before proceeding | |||
| 216 | Thread.sleep((zkTimeout * 2.5).toLong) | |||
| 151 | 217 | } | ||
| 152 | 218 | } | ||
| 153 | 219 | } | ||
| 1 | 1 | """ Test using root marathon. | ||
|---|---|---|---|---|
| 2 | 2 | This test suite imports all common tests found in marathon_common.py which are | ||
| 3 | 3 | to be tested on root marathon and MoM. | ||
| 4 | 4 | In addition it contains tests which are specific to root marathon, specifically | ||
| 5 | 5 | tests round dcos services registration and control and security. | ||
| 6 | 6 | """ | ||
| 7 | 7 | import common | ||
| 8 | 8 | import shakedown | ||
| 9 | import uuid | |||
| 9 | 10 | |||
| 10 | 11 | # this is intentional import * | ||
| 11 | 12 | # it imports all the common test_ methods which are to be tested on root and mom | ||
| 12 | 13 | from dcos_service_marathon_tests import * | ||
| 13 | 14 | from marathon_common_tests import * | ||
| 14 | 15 | from marathon_auth_common_tests import * | ||
| 15 | 16 | from marathon_pods_tests import * | ||
| 16 | 17 | |||
| Show All 38 Lines | ||||
| 55 | 56 | @retrying.retry(stop_max_attempt_number=30) | ||
| 56 | 57 | def marathon_leadership_changed(): | ||
| 57 | 58 | current_leader = shakedown.marathon_leader_ip() | ||
| 58 | 59 | print('leader: {}'.format(current_leader)) | ||
| 59 | 60 | assert original_leader != current_leader | ||
| 60 | 61 | |||
| 61 | 62 | marathon_leadership_changed() | ||
| 62 | 63 | |||
| 64 | @masters(3) | |||
| 65 | def test_marathon_delete_leader_and_check_apps(marathon_service_name): | |||
| 66 | ||||
| 67 | original_leader = shakedown.marathon_leader_ip() | |||
| 68 | print('leader: {}'.format(original_leader)) | |||
| 69 | ||||
| 70 | # start an app | |||
| 71 | app_def = common.app(id=uuid.uuid4().hex) | |||
| 72 | app_id = app_def['id'] | |||
| 73 | ||||
| 74 | client = marathon.create_client() | |||
| 75 | client.add_app(app_def) | |||
| 76 | shakedown.deployment_wait() | |||
| 77 | ||||
| 78 | app = client.get_app(app_id) | |||
| 79 | assert app['tasksRunning'] == 1 | |||
| 80 | ||||
| 81 | # abdicate leader after app was started successfully | |||
| 82 | common.delete_marathon_path('v2/leader') | |||
| 83 | ||||
| 84 | shakedown.wait_for_service_endpoint(marathon_service_name, timedelta(minutes=5).total_seconds()) | |||
| 85 | ||||
| 86 | @retrying.retry(stop_max_attempt_number=30) | |||
| 87 | def marathon_leadership_changed(): | |||
| 88 | current_leader = shakedown.marathon_leader_ip() | |||
| 89 | print('leader: {}'.format(current_leader)) | |||
| 90 | assert original_leader != current_leader | |||
| 91 | ||||
| 92 | # wait until leader changed | |||
| 93 | marathon_leadership_changed() | |||
| 94 | ||||
| 95 | @retrying.retry(stop_max_attempt_number=30) | |||
| 96 | def check_app_existence(expected_instances): | |||
| 97 | app = client.get_app(app_id) | |||
| 98 | assert app['tasksRunning'] == expected_instances | |||
| 99 | ||||
| 100 | # check if app definition is still there and one instance is still running after new leader was elected | |||
| 101 | check_app_existence(1) | |||
| 102 | ||||
| 103 | client.remove_app(app_id) | |||
| 104 | shakedown.deployment_wait() | |||
| 105 | ||||
| 106 | app = client.get_app(app_id) | |||
| 107 | assert app['tasksRunning'] == 0 | |||
| 108 | ||||
| 109 | # abdicate leader after app was started successfully | |||
| 110 | common.delete_marathon_path('v2/leader') | |||
| 111 | ||||
| 112 | shakedown.wait_for_service_endpoint(marathon_service_name, timedelta(minutes=5).total_seconds()) | |||
| 113 | ||||
| 114 | # wait until leader changed | |||
| 115 | marathon_leadership_changed() | |||
| 116 | ||||
| 117 | # check if app definition is still not there and no instance is running after new leader was elected | |||
| 118 | check_app_existence(0) | |||
| 119 | ||||
| 63 | 120 | |||
| 64 | 121 | @masters(3) | ||
| 65 | 122 | def test_marathon_zk_partition_leader_change(marathon_service_name): | ||
| 66 | 123 | |||
| 67 | 124 | original_leader = common.get_marathon_leader_not_on_master_leader_node() | ||
| 68 | 125 | |||
| 69 | 126 | # blocking zk on marathon leader (not master leader) | ||
| 70 | 127 | with shakedown.iptable_rules(original_leader): | ||
| ▲ Show 20 Lines • Show All 413 Lines • Show Last 20 Lines | ||||
Hm, since we have a few awaits we could wrap it in async/await. That might mess with synchronized though.