HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D338

Delete Legacy Storage (#4852)
ClosedAll Users
Actions

Authored by • jasongilanfarr on Dec 14 2016, 8:23 PM.

Details

Reviewers
aquamatthias
jdef
unterstein
timcharper
meichstedt
Commits
rMARATHON9dc73ae7bf18: Delete Legacy Storage (#4852)
Summary
  • We don't support legacy storage in 1.5+ anymore, so delete all of it.
  • We still use _some_ of MarathonState - I'd really like to get rid of that to and use a Proto typeclass or something (or just have the proto stuff live in the storage layer and not the classes).
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.
• jasongilanfarr retitled this revision from to Delete Legacy Storage (#4852).Dec 14 2016, 8:23 PM
• jasongilanfarr updated this object.
• jasongilanfarr edited the test plan for this revision. (Show Details)
• jasongilanfarr added reviewers: timcharper, unterstein, meichstedt.
timcharper accepted this revision.Dec 15 2016, 12:36 AM

＼\ ٩( ᐛ )و /／

LGTM except you probably want to run scalariform on this to reconcile formatting.

This revision is now accepted and ready to land.Dec 15 2016, 12:36 AM
• jasongilanfarr updated this revision to Diff 1317.Dec 15 2016, 7:04 PM
• jasongilanfarr updated this object.

-Rebased

jdef requested changes to this revision.Dec 15 2016, 9:02 PM
jdef added inline comments.
src/test/scala/mesosphere/marathon/state/RootGroupTest.scala
476

why is this validation test removed?

This revision now requires changes to proceed.Dec 15 2016, 9:02 PM
• jasongilanfarr added inline comments.Dec 15 2016, 9:03 PM
src/test/scala/mesosphere/marathon/state/RootGroupTest.scala
476

oh good catch, i meant to only remove the proto one.

• jasongilanfarr updated this revision to Diff 1319.Dec 15 2016, 10:11 PM
  • Undo deletion of a test case
jdef accepted this revision.Dec 15 2016, 10:31 PM
This revision is now accepted and ready to land.Dec 15 2016, 10:31 PM
Closed by commit rMARATHON9dc73ae7bf18: Delete Legacy Storage (#4852) (authored by • jasongilanfarr). · Explain WhyDec 15 2016, 11:20 PM
This revision was automatically updated to reflect the committed changes.
aquamatthias added a comment.Dec 16 2016, 9:41 AM

Even if already landed: what a great patch!

src/main/scala/mesosphere/marathon/storage/migration/Migration.scala
42

Think we should remove the storage format from the version.

unterstein added inline comments.Dec 16 2016, 11:03 AM
src/main/scala/mesosphere/marathon/storage/migration/Migration.scala
42

We should add a big block of documentation within the 1.5 changelog for this minimum requirement of 1.4 running before and that you can not directly upgrade from a version < 1.4

See https://github.com/mesosphere/marathon/issues/4864

• jasongilanfarr added inline comments.Dec 16 2016, 5:19 PM
src/main/scala/mesosphere/marathon/storage/migration/Migration.scala
42

We can't remove it unfortunately. If we read a storage format that has legacy, we can't migrate.

The 1.4 and 1.5 changelog will be very explicit about this.

Revision Contents

PathPackages
Msrc/main/scala/mesosphere/marathon/core/group/impl/GroupManagerActor.scala (2 lines)
Msrc/main/scala/mesosphere/marathon/state/Group.scala (38 lines)
Msrc/main/scala/mesosphere/marathon/state/RootGroup.scala (4 lines)
Msrc/main/scala/mesosphere/marathon/storage/StorageConf.scala (4 lines)
Msrc/main/scala/mesosphere/marathon/storage/StorageConfig.scala (158 lines)
Msrc/main/scala/mesosphere/marathon/storage/StorageModule.scala (55 lines)
Msrc/main/scala/mesosphere/marathon/storage/migration/Migration.scala (143 lines)
DMsrc/main/scala/mesosphere/marathon/storage/migration/MigrationTo1_4_PersistenceStore.scala (234 lines)
DMsrc/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_11.scala (120 lines)
DMsrc/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_13.scala (125 lines)
DMsrc/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_16.scala (67 lines)
DMsrc/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo1_1_5.scala (135 lines)
DMsrc/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo1_2.scala (69 lines)
DMsrc/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo_1_4_0.scala (111 lines)
Msrc/main/scala/mesosphere/marathon/storage/repository/Repositories.scala (75 lines)
DMsrc/main/scala/mesosphere/marathon/storage/repository/legacy/LegacyEntityRepository.scala (296 lines)
DMsrc/main/scala/mesosphere/marathon/storage/repository/legacy/store/EntityStore.scala (33 lines)
DMsrc/main/scala/mesosphere/marathon/storage/repository/legacy/store/EntityStoreCache.scala (124 lines)
DMsrc/main/scala/mesosphere/marathon/storage/repository/legacy/store/InMemoryStore.scala (53 lines)
DMsrc/main/scala/mesosphere/marathon/storage/repository/legacy/store/JavaFutureToFuture.scala (30 lines)
DMsrc/main/scala/mesosphere/marathon/storage/repository/legacy/store/MarathonStore.scala (100 lines)
DMsrc/main/scala/mesosphere/marathon/storage/repository/legacy/store/MesosStateStore.scala (95 lines)
DMsrc/main/scala/mesosphere/marathon/storage/repository/legacy/store/PersistentStore.scala (120 lines)
DMsrc/main/scala/mesosphere/marathon/storage/repository/legacy/store/ZKStore.scala (181 lines)
Msrc/main/scala/mesosphere/marathon/upgrade/DeploymentPlan.scala (46 lines)
Msrc/test/scala/mesosphere/marathon/core/storage/store/impl/zk/ZkPersistenceStoreTest.scala (22 lines)
DMsrc/test/scala/mesosphere/marathon/state/MarathonStoreTest.scala (253 lines)
Msrc/test/scala/mesosphere/marathon/state/RootGroupTest.scala (61 lines)
Msrc/test/scala/mesosphere/marathon/storage/migration/MigrationTest.scala (103 lines)
DMsrc/test/scala/mesosphere/marathon/storage/migration/MigrationTo1_1_5_Test.scala (235 lines)
DMsrc/test/scala/mesosphere/marathon/storage/migration/MigrationTo1_4_PersistenceStoreTest.scala (313 lines)
DMsrc/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_11Test.scala (156 lines)
DMsrc/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_13Test.scala (218 lines)
DMsrc/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_16Test.scala (142 lines)
DMsrc/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo1_2Test.scala (100 lines)
DMsrc/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo_1_40Test.scala (117 lines)
Msrc/test/scala/mesosphere/marathon/storage/repository/GroupRepositoryTest.scala (36 lines)
Msrc/test/scala/mesosphere/marathon/storage/repository/RepositoryTest.scala (56 lines)
Msrc/test/scala/mesosphere/marathon/storage/repository/SingletonRepositoryTest.scala (25 lines)
DMsrc/test/scala/mesosphere/marathon/storage/repository/legacy/store/EntityStoreCacheTest.scala (259 lines)
DMsrc/test/scala/mesosphere/marathon/storage/repository/legacy/store/ZKStoreTest.scala (131 lines)
Msrc/test/scala/mesosphere/marathon/tasks/InstanceTrackerImplTest.scala (8 lines)
Msrc/test/scala/mesosphere/marathon/test/MarathonTestHelper.scala (12 lines)
Msrc/test/scala/mesosphere/marathon/upgrade/DeploymentManagerTest.scala (19 lines)
Msrc/test/scala/mesosphere/marathon/upgrade/DeploymentPlanTest.scala (32 lines)
Msrc/test/scala/mesosphere/marathon/upgrade/TaskStartActorTest.scala (29 lines)
DMsrc/test/scala/mesosphere/util/state/PersistentStoreTest.scala (101 lines)
DMsrc/test/scala/mesosphere/util/state/memory/InMemoryStoreTest.scala (26 lines)
DMsrc/test/scala/mesosphere/util/state/mesos/MesosStateStoreTest.scala (33 lines)

Diff 1324

View Options

src/main/scala/mesosphere/marathon/core/group/impl/GroupManagerActor.scala

Show First 20 Lines • Show All 139 Lines • ▼ Show 20 Line(s)
140140 log.info(s"Upgrade root group version:$version with force:$force")
141141
142142 val deployment = for {
143143 from <- groupRepo.root()
144144 (toUnversioned, resolve) <- resolveStoreUrls(assignDynamicServicePorts(from, change(from)))
145145 to = GroupVersioningUtil.updateVersionInfoForChangedApps(version, from, toUnversioned)
146146 _ = validateOrThrow(to)(RootGroup.valid(config.availableFeatures))
147147 plan = DeploymentPlan(from, to, resolve, version, toKill)
148 _ = validateOrThrow(plan)(DeploymentPlan.deploymentPlanValidator(config))
148 _ = validateOrThrow(plan)(DeploymentPlan.deploymentPlanValidator())
149149 _ = log.info(s"Computed new deployment plan:\n$plan")
150150 _ <- groupRepo.storeRootVersion(plan.target, plan.createdOrUpdatedApps, plan.createdOrUpdatedPods)
151151 _ <- scheduler.deploy(plan, force)
152152 _ <- groupRepo.storeRoot(plan.target, plan.createdOrUpdatedApps,
153153 plan.deletedApps, plan.createdOrUpdatedPods, plan.deletedPods)
154154 _ = log.info(s"Updated groups/apps/pods according to deployment plan ${plan.id}")
155155 } yield plan
156156
▲ Show 20 Lines • Show All 124 Lines • Show Last 20 Lines
View Options

src/main/scala/mesosphere/marathon/state/Group.scala

11package mesosphere.marathon
22package state
33
44import java.util.Objects
55
66import com.wix.accord._
77import com.wix.accord.dsl._
8import mesosphere.marathon.Protos.GroupDefinition
98import mesosphere.marathon.api.v2.Validation._
109import mesosphere.marathon.core.pod.PodDefinition
1110import mesosphere.marathon.plugin.{ Group => IGroup }
1211import mesosphere.marathon.state.Group._
1312import mesosphere.marathon.state.PathId._
14import mesosphere.marathon.stream._
1513
1614class Group(
1715 val id: PathId,
1816 val apps: Map[AppDefinition.AppKey, AppDefinition] = defaultApps,
1917 val pods: Map[PathId, PodDefinition] = defaultPods,
2018 val groupsById: Map[Group.GroupKey, Group] = defaultGroups,
2119 val dependencies: Set[PathId] = defaultDependencies,
2220 val version: Timestamp = defaultVersion,
2321 val transitiveAppsById: Map[AppDefinition.AppKey, AppDefinition],
24 val transitivePodsById: Map[PathId, PodDefinition]) extends MarathonState[GroupDefinition, Group] with IGroup {
25 override def mergeFromProto(msg: GroupDefinition): Group = Group.fromProto(msg)
26 override def mergeFromProto(bytes: Array[Byte]): Group = Group.fromProto(GroupDefinition.parseFrom(bytes))
22 val transitivePodsById: Map[PathId, PodDefinition]) extends IGroup {
2723
28 override def toProto: GroupDefinition = {
29 GroupDefinition.newBuilder
30 .setId(id.toString)
31 .setVersion(version.toString)
32 .addAllDeprecatedApps(apps.map { case (_, app) => app.toProto })
33 .addAllDeprecatedPods(pods.map { case (_, pod) => pod.toProto })
34 .addAllGroups(groupsById.map { case (_, group) => group.toProto })
35 .addAllDependencies(dependencies.map(_.toString))
36 .build()
37 }
38
3924 def app(appId: PathId): Option[AppDefinition] = transitiveAppsById.get(appId)
4025 def pod(podId: PathId): Option[PodDefinition] = transitivePodsById.get(podId)
4126
4227 /**
4328 * Find and return the child group for the given path. If no match is found, then returns None
4429 */
4530 def group(gid: PathId): Option[Group] = {
4631 if (id == gid) Some(this)
▲ Show 20 Lines • Show All 48 Lines • ▼ Show 20 Line(s)
9580 dependencies: Set[PathId] = Group.defaultDependencies,
9681 version: Timestamp = Group.defaultVersion,
9782 transitiveAppsById: Map[AppDefinition.AppKey, AppDefinition],
9883 transitivePodsById: Map[PathId, PodDefinition]): Group =
9984 new Group(id, apps, pods, groupsById, dependencies, version, transitiveAppsById, transitivePodsById)
10085
10186 def empty(id: PathId): Group =
10287 Group(id = id, version = Timestamp(0), transitiveAppsById = Map.empty, transitivePodsById = Map.empty)
103
104 def fromProto(msg: GroupDefinition): Group = {
105 val appsById: Map[AppDefinition.AppKey, AppDefinition] = msg.getDeprecatedAppsList.map { proto =>
106 val app = AppDefinition.fromProto(proto)
107 app.id -> app
108 }(collection.breakOut)
109 val podsById: Map[PathId, PodDefinition] = msg.getDeprecatedPodsList.map { proto =>
110 val pod = PodDefinition.fromProto(proto)
111 pod.id -> pod
112 }(collection.breakOut)
113 val groupsById: Map[PathId, Group] = msg.getGroupsList.map(fromProto).map(group => group.id -> group)(collection.breakOut)
114 Group(
115 id = msg.getId.toPath,
116 apps = appsById,
117 pods = podsById,
118 groupsById = groupsById,
119 dependencies = msg.getDependenciesList.map(PathId.apply)(collection.breakOut),
120 version = Timestamp(msg.getVersion),
121 transitiveAppsById = appsById ++ groupsById.values.flatMap(_.transitiveAppsById),
122 transitivePodsById = podsById ++ groupsById.values.flatMap(_.transitivePodsById))
123 }
12488
12589 def defaultApps: Map[AppDefinition.AppKey, AppDefinition] = Map.empty
12690 val defaultPods = Map.empty[PathId, PodDefinition]
12791 def defaultGroups: Map[Group.GroupKey, Group] = Map.empty
12892 def defaultDependencies: Set[PathId] = Set.empty
12993 def defaultVersion: Timestamp = Timestamp.now()
13094
13195 def valid(base: PathId, enabledFeatures: Set[String]): Validator[Group] =
Show All 31 Lines
View Options

src/main/scala/mesosphere/marathon/state/RootGroup.scala

11package mesosphere.marathon
22package state
33
44import com.wix.accord._
55import com.wix.accord.dsl._
6import mesosphere.marathon.Protos.GroupDefinition
76import mesosphere.marathon.api.v2.Validation._
87import mesosphere.marathon.core.externalvolume.ExternalVolumes
98import mesosphere.marathon.core.pod.PodDefinition
10
119import org.jgrapht.DirectedGraph
1210import org.jgrapht.alg.CycleDetector
1311import org.jgrapht.graph._
1412
1513import scala.annotation.tailrec
1614import scala.collection.JavaConverters._
1715
1816/**
▲ Show 20 Lines • Show All 399 Lines • ▼ Show 20 Line(s)
418416 def apply(
419417 apps: Map[AppDefinition.AppKey, AppDefinition] = Group.defaultApps,
420418 pods: Map[PathId, PodDefinition] = Group.defaultPods,
421419 groupsById: Map[Group.GroupKey, Group] = Group.defaultGroups,
422420 dependencies: Set[PathId] = Group.defaultDependencies,
423421 version: Timestamp = Group.defaultVersion): RootGroup = new RootGroup(apps, pods, groupsById, dependencies, version)
424422
425423 def empty: RootGroup = RootGroup(version = Timestamp(0))
426
427 def fromProto(msg: GroupDefinition): RootGroup = RootGroup.fromGroup(Group.fromProto(msg))
428424
429425 def fromGroup(group: Group): RootGroup = {
430426 require(group.id.isRoot)
431427 RootGroup(group.apps, group.pods, group.groupsById, group.dependencies, group.version)
432428 }
433429
434430 def valid(enabledFeatures: Set[String]): Validator[RootGroup] = {
435431 noCyclicDependencies and
436432 Group.valid(PathId.empty, enabledFeatures) and
437433 ExternalVolumes.validRootGroup()
438434 }
439435
440436 private def noCyclicDependencies: Validator[RootGroup] =
441437 isTrue("Dependency graph has cyclic dependencies.") { _.hasNonCyclicDependencies }
442438}
View Options

src/main/scala/mesosphere/marathon/storage/StorageConf.scala

11package mesosphere.marathon.storage
22
33import mesosphere.marathon.ZookeeperConf
44
55trait StorageConf extends ZookeeperConf {
66 lazy val internalStoreBackend = opt[String](
77 "internal_store_backend",
8 descr = s"The backend storage system to use. One of ${TwitterZk.StoreName}, ${MesosZk.StoreName}, ${InMem.StoreName}, ${CuratorZk.StoreName}",
8 descr = s"The backend storage system to use. One of ${InMem.StoreName}, ${CuratorZk.StoreName}",
99 hidden = true,
10 validate = Set(TwitterZk.StoreName, MesosZk.StoreName, InMem.StoreName, CuratorZk.StoreName).contains,
10 validate = Set(InMem.StoreName, CuratorZk.StoreName).contains,
1111 default = Some(CuratorZk.StoreName)
1212 )
1313
1414 lazy val storeCache = toggle(
1515 "store_cache",
1616 default = Some(true),
1717 noshort = true,
1818 descrYes = "(Default) Enable an in-memory cache for the storage layer.",
Show All 28 Lines
View Options

src/main/scala/mesosphere/marathon/storage/StorageConfig.scala

11package mesosphere.marathon
22package storage
33
44import java.util
55import java.util.Collections
6import java.util.concurrent.TimeUnit
76
87import akka.actor.{ ActorRefFactory, Scheduler }
98import akka.stream.Materializer
10import com.typesafe.config.{ Config, ConfigMemorySize }
9import com.typesafe.config.Config
1110import mesosphere.marathon.core.storage.store.PersistenceStore
1211import mesosphere.marathon.core.storage.store.impl.BasePersistenceStore
1312import mesosphere.marathon.core.storage.store.impl.cache.{ LazyCachingPersistenceStore, LazyVersionCachingPersistentStore, LoadTimeCachingPersistenceStore }
1413import mesosphere.marathon.core.storage.store.impl.memory.{ Identity, InMemoryPersistenceStore, RamId }
1514import mesosphere.marathon.core.storage.store.impl.zk.{ NoRetryPolicy, RichCuratorFramework, ZkId, ZkPersistenceStore, ZkSerialized }
1615import mesosphere.marathon.metrics.Metrics
17import mesosphere.marathon.state.MarathonState
18import mesosphere.marathon.storage.repository.legacy.store._
19import mesosphere.marathon.stream._
2016import mesosphere.marathon.util.{ RetryConfig, toRichConfig }
2117import org.apache.curator.framework.api.ACLProvider
2218import org.apache.curator.framework.imps.GzipCompressionProvider
2319import org.apache.curator.framework.{ AuthInfo, CuratorFrameworkFactory }
24import org.apache.mesos.state.ZooKeeperState
2520import org.apache.zookeeper.ZooDefs
2621import org.apache.zookeeper.data.ACL
2722
2823import scala.concurrent.ExecutionContext
2924import scala.concurrent.duration.{ Duration, _ }
30import scala.reflect.ClassTag
3125
3226sealed trait StorageConfig extends Product with Serializable
33sealed trait LegacyStorageConfig extends StorageConfig {
34 protected[storage] def store: PersistentStore
35 val maxVersions: Int
36 val enableCache: Boolean
37 val availableFeatures: Set[String]
3827
39 def entityStore[T <: MarathonState[_, T]](prefix: String, newState: () => T)(
40 implicit
41 metrics: Metrics, ct: ClassTag[T]): EntityStore[T] = {
42 val marathonStore = new MarathonStore[T](store, metrics, newState, prefix)
43 if (enableCache) new EntityStoreCache[T](marathonStore) else marathonStore
44 }
45}
46
47// only for testing
48private[storage] case class LegacyInMemConfig(maxVersions: Int) extends LegacyStorageConfig {
49 override protected[storage] val store: PersistentStore = new InMemoryStore()
50 override val enableCache: Boolean = false
51 override val availableFeatures: Set[String] = Set.empty
52}
53
54case class TwitterZk(
55 maxVersions: Int,
56 enableCache: Boolean,
57 sessionTimeout: Duration,
58 zkHosts: String,
59 zkPath: String,
60 zkAcl: util.List[ACL],
61 username: Option[String],
62 password: Option[String],
63 retries: Int,
64 enableCompression: Boolean,
65 compressionThreshold: ConfigMemorySize,
66 maxConcurrent: Int,
67 maxOutstanding: Int,
68 availableFeatures: Set[String]) extends LegacyStorageConfig {
69
70 private val sessionTimeoutTw = {
71 com.twitter.util.Duration(sessionTimeout.toMillis, TimeUnit.MILLISECONDS)
72 }
73
74 protected[storage] lazy val store: PersistentStore = {
75 import com.twitter.util.JavaTimer
76 import com.twitter.zk.{ AuthInfo, NativeConnector, ZkClient }
77
78 val authInfo = (username, password) match {
79 case (Some(user), Some(pass)) => Some(AuthInfo.digest(user, pass))
80 case _ => None
81 }
82
83 val connector = NativeConnector(zkHosts, None, sessionTimeoutTw, new JavaTimer(isDaemon = true), authInfo)
84
85 val client = ZkClient(connector)
86 .withAcl(zkAcl.toSeq)
87 .withRetries(retries)
88 val compressionConf = CompressionConf(enableCompression, compressionThreshold.toBytes)
89 new ZKStore(client, client(zkPath), compressionConf, maxConcurrent = maxConcurrent, maxOutstanding = maxOutstanding)
90 }
91}
92
93object TwitterZk {
94 val StoreName = "legacy_zk"
95
96 def apply(config: StorageConf): TwitterZk =
97 TwitterZk(
98 maxVersions = config.maxVersions(),
99 enableCache = config.storeCache(),
100 sessionTimeout = config.zkSessionTimeoutDuration,
101 zkHosts = config.zkHosts,
102 zkPath = config.zooKeeperStatePath,
103 zkAcl = config.zkDefaultCreationACL,
104 username = config.zkUsername,
105 password = config.zkPassword,
106 retries = 3,
107 enableCompression = config.zooKeeperCompressionEnabled(),
108 compressionThreshold = ConfigMemorySize.ofBytes(config.zooKeeperCompressionThreshold()),
109 maxConcurrent = config.zkMaxConcurrency(),
110 maxOutstanding = 1024,
111 availableFeatures = config.availableFeatures)
112
113 def apply(config: Config): TwitterZk = {
114 val username = config.optionalString("username")
115 val password = config.optionalString("password")
116 val acls = (username, password) match {
117 case (Some(_), Some(_)) => ZooDefs.Ids.CREATOR_ALL_ACL
118 case _ => ZooDefs.Ids.OPEN_ACL_UNSAFE
119 }
120 TwitterZk(
121 maxVersions = config.int("max-versions", StorageConfig.DefaultLegacyMaxVersions),
122 enableCache = config.bool("enable-cache", true),
123 sessionTimeout = config.duration("session-timeout", 10.seconds),
124 zkHosts = config.stringList("hosts", Seq("localhost:2181")).mkString(","),
125 zkPath = s"${config.string("path", "marathon")}/state",
126 zkAcl = acls,
127 username = username,
128 password = password,
129 retries = config.int("retries", 3),
130 enableCompression = config.bool("enable-compression", true),
131 compressionThreshold = config.memorySize("compression-threshold", ConfigMemorySize.ofBytes(64 * 1024)),
132 maxConcurrent = config.int("max-concurrent", 32),
133 maxOutstanding = config.int("max-outstanding", 1024),
134 availableFeatures = config.stringList("available-features", Seq.empty).to[Set]
135 )
136 }
137}
138
139case class MesosZk(
140 maxVersions: Int,
141 enableCache: Boolean,
142 zkHosts: String,
143 zkPath: String,
144 timeout: Duration,
145 availableFeatures: Set[String]) extends LegacyStorageConfig {
146 def store: PersistentStore = {
147 val state = new ZooKeeperState(
148 zkHosts,
149 timeout.toMillis,
150 TimeUnit.MILLISECONDS,
151 zkPath
152 )
153 new MesosStateStore(state, timeout)
154 }
155}
156
157object MesosZk {
158 val StoreName = "mesos_zk"
159
160 def apply(config: StorageConf): MesosZk =
161 MesosZk(
162 maxVersions = config.maxVersions(),
163 enableCache = config.storeCache(),
164 zkHosts = config.zkHosts,
165 zkPath = config.zooKeeperStatePath,
166 timeout = config.zkTimeoutDuration,
167 availableFeatures = config.availableFeatures)
168
169 def apply(config: Config): MesosZk =
170 MesosZk(
171 maxVersions = config.int("max-versions", StorageConfig.DefaultLegacyMaxVersions),
172 enableCache = config.bool("enable-cache", true),
173 zkHosts = config.stringList("hosts", Seq("localhost:2181")).mkString(","),
174 zkPath = s"${config.string("path", "marathon")}/state",
175 timeout = config.duration("timeout", 10.seconds),
176 availableFeatures = config.stringList("available-features", Seq.empty).to[Set]
177 )
178}
179
18028sealed trait CacheType
18129case object NoCaching extends CacheType
18230case object EagerCaching extends CacheType
18331case object LazyCaching extends CacheType
18432
18533object CacheType {
18634 def apply(str: String): CacheType = str.toLowerCase match {
18735 case str: String if str.startsWith("eager") => EagerCaching
▲ Show 20 Lines • Show All 173 Lines • ▼ Show 20 Line(s)
361209
362210object StorageConfig {
363211 val DefaultVersionCacheConfig = Option(VersionCacheConfig.Default)
364212
365213 val DefaultLegacyMaxVersions = 25
366214 val DefaultMaxVersions = 5000
367215 def apply(conf: StorageConf): StorageConfig = {
368216 conf.internalStoreBackend() match {
369 case TwitterZk.StoreName => TwitterZk(conf)
370 case MesosZk.StoreName => MesosZk(conf)
371217 case InMem.StoreName => InMem(conf)
372218 case CuratorZk.StoreName => CuratorZk(conf)
373219 }
374220 }
375221
376222 def apply(conf: Config): StorageConfig = {
377223 conf.string("storage-type", "zk") match {
378 case TwitterZk.StoreName => TwitterZk(conf)
379 case MesosZk.StoreName => MesosZk(conf)
380224 case InMem.StoreName => InMem(conf)
381225 case CuratorZk.StoreName => CuratorZk(conf)
382226 }
383227 }
384228}
View Options

src/main/scala/mesosphere/marathon/storage/StorageModule.scala

1package mesosphere.marathon.storage
1package mesosphere.marathon
2package storage
23
34import akka.actor.{ ActorRefFactory, Scheduler }
45import akka.stream.Materializer
5import mesosphere.marathon.PrePostDriverCallback
6import mesosphere.marathon.core.event.EventSubscribers
7import mesosphere.marathon.core.instance.Instance
8import mesosphere.marathon.core.pod.PodDefinition
96import mesosphere.marathon.core.storage.store.impl.cache.LoadTimeCachingPersistenceStore
107import mesosphere.marathon.metrics.Metrics
11import mesosphere.marathon.state.{ AppDefinition, Group, MarathonTaskState, TaskFailure }
128import mesosphere.marathon.storage.migration.Migration
139import mesosphere.marathon.storage.repository._
14import mesosphere.marathon.upgrade.DeploymentPlan
15import mesosphere.util.state.FrameworkId
1610
1711import scala.collection.immutable.Seq
1812import scala.concurrent.ExecutionContext
1913
2014/**
2115 * Provides the repositories for all persistable entities.
2216 */
2317trait StorageModule {
2418 val appRepository: ReadOnlyAppRepository
2519 val podRepository: ReadOnlyPodRepository
2620 val instanceRepository: InstanceRepository
2721 val deploymentRepository: DeploymentRepository
2822 val taskFailureRepository: TaskFailureRepository
2923 val groupRepository: GroupRepository
3024 val frameworkIdRepository: FrameworkIdRepository
3125 val eventSubscribersRepository: EventSubscribersRepository
3226 val migration: Migration
3327 val leadershipInitializers: Seq[PrePostDriverCallback]
3428}
3529
3630object StorageModule {
3731 def apply(conf: StorageConf)(implicit metrics: Metrics, mat: Materializer, ctx: ExecutionContext,
3832 scheduler: Scheduler, actorRefFactory: ActorRefFactory): StorageModule = {
3933 val currentConfig = StorageConfig(conf)
40 val legacyConfig = conf.internalStoreBackend() match {
41 case TwitterZk.StoreName => Some(TwitterZk(conf))
42 case MesosZk.StoreName => Some(MesosZk(conf))
43 case CuratorZk.StoreName => Some(TwitterZk(conf))
44 case InMem.StoreName => None
34 apply(currentConfig)
4535 }
46 apply(currentConfig, legacyConfig)
47 }
4836
4937 def apply(
50 config: StorageConfig,
51 legacyConfig: Option[LegacyStorageConfig])(implicit
38 config: StorageConfig)(implicit
5239 metrics: Metrics,
5340 mat: Materializer, ctx: ExecutionContext,
5441 scheduler: Scheduler, actorRefFactory: ActorRefFactory): StorageModule = {
5542
5643 config match {
57 case l: LegacyStorageConfig =>
58 val appStore = l.entityStore[AppDefinition] _
59 val appRepository = AppRepository.legacyRepository(appStore, l.maxVersions)
60 val podStore = l.entityStore[PodDefinition] _
61 val podRepository = PodRepository.legacyRepository(podStore, l.maxVersions)
62 val taskStore = l.entityStore[MarathonTaskState] _
63 val taskRepository = TaskRepository.legacyRepository(taskStore)
64 val instanceStore = l.entityStore[Instance] _
65 val instanceRepository = InstanceRepository.legacyRepository(instanceStore)
66 val deployStore = l.entityStore[DeploymentPlan] _
67 val deploymentRepository = DeploymentRepository.legacyRepository(deployStore)
68 val taskFailureStore = l.entityStore[TaskFailure] _
69 val taskFailureRepository = TaskFailureRepository.legacyRepository(taskFailureStore)
70 val groupStore = l.entityStore[Group] _
71 val groupRepository = GroupRepository.legacyRepository(groupStore, l.maxVersions, appRepository, podRepository)
72 val frameworkIdStore = l.entityStore[FrameworkId] _
73 val frameworkIdRepository = FrameworkIdRepository.legacyRepository(frameworkIdStore)
74 val eventSubscribersStore = l.entityStore[EventSubscribers] _
75 val eventSubscribersRepository = EventSubscribersRepository.legacyRepository(eventSubscribersStore)
76
77 val migration = new Migration(l.availableFeatures, legacyConfig, None, appRepository, groupRepository,
78 deploymentRepository, taskRepository, instanceRepository, taskFailureRepository,
79 frameworkIdRepository, eventSubscribersRepository)
80
81 val leadershipInitializers = Seq(appStore, taskStore, deployStore, taskFailureStore,
82 groupStore, frameworkIdStore, eventSubscribersStore).collect { case s: PrePostDriverCallback => s }
83
84 StorageModuleImpl(appRepository, podRepository, instanceRepository, deploymentRepository,
85 taskFailureRepository, groupRepository, frameworkIdRepository, eventSubscribersRepository, migration,
86 leadershipInitializers)
8744 case zk: CuratorZk =>
8845 val store = zk.store
8946 val appRepository = AppRepository.zkRepository(store)
9047 val podRepository = PodRepository.zkRepository(store)
9148 val groupRepository = GroupRepository.zkRepository(store, appRepository, podRepository)
9249
9350 val taskRepository = TaskRepository.zkRepository(store)
9451 val instanceRepository = InstanceRepository.zkRepository(store)
9552 val deploymentRepository = DeploymentRepository.zkRepository(store, groupRepository,
9653 appRepository, podRepository, zk.maxVersions)
9754 val taskFailureRepository = TaskFailureRepository.zkRepository(store)
9855 val frameworkIdRepository = FrameworkIdRepository.zkRepository(store)
9956 val eventSubscribersRepository = EventSubscribersRepository.zkRepository(store)
10057
10158 val leadershipInitializers = store match {
10259 case s: LoadTimeCachingPersistenceStore[_, _, _] =>
10360 Seq(s)
10461 case _ =>
10562 Nil
10663 }
10764
108 val migration = new Migration(zk.availableFeatures, legacyConfig, Some(store), appRepository, groupRepository,
65 val migration = new Migration(zk.availableFeatures, store, appRepository, groupRepository,
10966 deploymentRepository, taskRepository, instanceRepository, taskFailureRepository,
11067 frameworkIdRepository, eventSubscribersRepository)
11168 StorageModuleImpl(
11269 appRepository,
11370 podRepository,
11471 instanceRepository,
11572 deploymentRepository,
11673 taskFailureRepository,
Show All 17 Lines
13491
13592 val leadershipInitializers = store match {
13693 case s: LoadTimeCachingPersistenceStore[_, _, _] =>
13794 Seq(s)
13895 case _ =>
13996 Nil
14097 }
14198
142 val migration = new Migration(mem.availableFeatures, legacyConfig, Some(store), appRepository, groupRepository,
99 val migration = new Migration(mem.availableFeatures, store, appRepository, groupRepository,
143100 deploymentRepository, taskRepository, instanceRepository, taskFailureRepository,
144101 frameworkIdRepository, eventSubscribersRepository)
145102 StorageModuleImpl(
146103 appRepository,
147104 podRepository,
148105 instanceRepository,
149106 deploymentRepository,
150107 taskFailureRepository,
Show All 20 Lines
View Options

src/main/scala/mesosphere/marathon/storage/migration/Migration.scala

11package mesosphere.marathon
22package storage.migration
33
44import akka.Done
55import akka.stream.Materializer
66import com.typesafe.scalalogging.StrictLogging
77import mesosphere.marathon.Protos.StorageVersion
88import mesosphere.marathon.core.storage.store.PersistenceStore
99import mesosphere.marathon.metrics.Metrics
10import mesosphere.marathon.storage.LegacyStorageConfig
11import mesosphere.marathon.storage.migration.legacy._
1210import mesosphere.marathon.storage.repository._
13import mesosphere.marathon.storage.repository.legacy.store.{ PersistentStore, PersistentStoreManagement }
1411
1512import scala.async.Async.{ async, await }
1613import scala.concurrent.ExecutionContext.Implicits.global
1714import scala.concurrent.duration._
1815import scala.concurrent.{ Await, Future }
1916import scala.util.control.NonFatal
2017import scala.util.matching.Regex
2118
2219/**
23 * @param legacyConfig Optional configuration for the legacy store. This is used for all migrations
24 * that do not use the new store and the underlying PersistentStore will be closed
25 * when completed
2620 * @param persistenceStore Optional "new" PersistenceStore for new migrations, the repositories
2721 * are assumed to be in the new format.
2822 */
23@SuppressWarnings(Array("UnusedMethodParameter")) // materializer will definitely be used in the future.
2924class Migration(
3025 private[migration] val availableFeatures: Set[String],
31 private[migration] val legacyConfig: Option[LegacyStorageConfig],
32 private[migration] val persistenceStore: Option[PersistenceStore[_, _, _]],
26 private[migration] val persistenceStore: PersistenceStore[_, _, _],
3327 private[migration] val appRepository: AppRepository,
3428 private[migration] val groupRepository: GroupRepository,
3529 private[migration] val deploymentRepository: DeploymentRepository,
3630 private[migration] val taskRepo: TaskRepository,
3731 private[migration] val instanceRepo: InstanceRepository,
3832 private[migration] val taskFailureRepo: TaskFailureRepository,
3933 private[migration] val frameworkIdRepo: FrameworkIdRepository,
4034 private[migration] val eventSubscribersRepo: EventSubscribersRepository)(implicit
4135 mat: Materializer,
4236 metrics: Metrics) extends StrictLogging {
4337
44 import Migration._
4538 import StorageVersions._
4639
4740 type MigrationAction = (StorageVersion, () => Future[Any])
4841
49 private[migration] val minSupportedStorageVersion = StorageVersions(0, 8, 0)
42 private[migration] val minSupportedStorageVersion = StorageVersions(1, 4, 0, StorageVersion.StorageFormat.PERSISTENCE_STORE)
aquamatthiasUnsubmitted
Not Done

Think we should remove the storage format from the version.

untersteinUnsubmitted
Not Done

We should add a big block of documentation within the 1.5 changelog for this minimum requirement of 1.4 running before and that you can not directly upgrade from a version < 1.4

See https://github.com/mesosphere/marathon/issues/4864

jasongilanfarrAuthorUnsubmitted
Not Done

We can't remove it unfortunately. If we read a storage format that has legacy, we can't migrate.

The 1.4 and 1.5 changelog will be very explicit about this.

5043
51 private[migration] lazy val legacyStoreFuture: Future[Option[PersistentStore]] = legacyConfig.map { config =>
52 val store = config.store
53 store match {
54 case s: PersistentStoreManagement with PrePostDriverCallback =>
55 s.preDriverStarts.flatMap(_ => s.initialize()).map(_ => Some(store))
56 case s: PersistentStoreManagement =>
57 s.initialize().map(_ => Some(store))
58 case s: PrePostDriverCallback =>
59 s.preDriverStarts.map(_ => Some(store))
60 case _ =>
61 Future.successful(Some(store))
62 }
63 }.getOrElse(Future.successful(None))
64
6544 /**
6645 * All the migrations, that have to be applied.
6746 * They get applied after the master has been elected.
6847 */
69 def migrations: List[MigrationAction] =
70 List(
71 StorageVersions(0, 7, 0) -> { () =>
72 Future.failed(new IllegalStateException("migration from 0.7.x not supported anymore"))
73 },
74 StorageVersions(0, 11, 0) -> { () =>
75 new MigrationTo0_11(legacyConfig).migrateApps().recover {
76 case NonFatal(e) => throw new MigrationFailedException("while migrating storage to 0.11", e)
77 }
78 },
79 StorageVersions(0, 13, 0) -> { () =>
80 new MigrationTo0_13(legacyConfig).migrate().recover {
81 case NonFatal(e) => throw new MigrationFailedException("while migrating storage to 0.13", e)
82 }
83 },
84 StorageVersions(0, 16, 0) -> { () =>
85 new MigrationTo0_16(legacyConfig).migrate().recover {
86 case NonFatal(e) => throw new MigrationFailedException("while migrating storage to 0.16", e)
87 }
88 },
89 StorageVersions(1, 1, 5) -> { () =>
90 new MigrationTo1_1_5(availableFeatures, legacyConfig).migrate().recover {
91 case NonFatal(e) => throw new MigrationFailedException("while migrating storage to 1.1.5", e)
92 }
93 },
94 StorageVersions(1, 2, 0) -> { () =>
95 new MigrationTo1_2(legacyConfig).migrate().recover {
96 case NonFatal(e) => throw new MigrationFailedException("while migrating storage to 1.2", e)
97 }
98 },
99 StorageVersions(1, 4, 0, StorageVersion.StorageFormat.LEGACY) -> { () =>
100 new MigrationTo_1_4_0(legacyConfig).migrate().recover {
101 case NonFatal(e) => throw new MigrationFailedException("while migrating storage to 1.3.6")
102 }
103 },
104 StorageVersions(1, 4, 0, StorageVersion.StorageFormat.PERSISTENCE_STORE) -> { () =>
105 new MigrationTo1_4_PersistenceStore(this).migrate().recover {
106 case NonFatal(e) => throw new MigrationFailedException("while migrating storage to 1.4.0", e)
107 }
108 }
109 )
48 def migrations: List[MigrationAction] = List.empty
11049
11150 def applyMigrationSteps(from: StorageVersion): Future[Seq[StorageVersion]] = {
11251 migrations.filter(_._1 > from).sortBy(_._1).foldLeft(Future.successful(Seq.empty[StorageVersion])) {
11352 case (resultsFuture, (migrateVersion, change)) => resultsFuture.flatMap { res =>
11453 logger.info(
11554 s"Migration for storage: ${from.str} to current: ${current.str}: " +
11655 s"apply change for version: ${migrateVersion.str} "
11756 )
11857 change.apply().map(_ => res :+ migrateVersion)
11958 }
12059 }
12160 }
12261
12362 @SuppressWarnings(Array("all")) // async/await
12463 def migrate(): Seq[StorageVersion] = {
125 val result = async { // linter:ignore UnnecessaryElseBranch
126 val legacyStore = await(legacyStoreFuture)
127 val currentVersion = await(getCurrentVersion(legacyStore))
64 val result = async {
65 val currentVersion = await(getCurrentVersion())
12866
129 val currentBuildVersion = persistenceStore.fold(StorageVersions.current) { _ =>
130 StorageVersions.current.toBuilder.setFormat(StorageVersion.StorageFormat.PERSISTENCE_STORE).build
131 }
67 val currentBuildVersion = StorageVersions.current
13268
133 val migrations = (currentVersion, persistenceStore) match {
134 case (Some(version), _) if version < minSupportedStorageVersion =>
69 val migrations = currentVersion match {
70 case Some(version) if version < minSupportedStorageVersion =>
13571 val msg = s"Migration from versions < ${minSupportedStorageVersion.str} are not supported. " +
13672 s"Your version: ${version.str}"
13773 throw new MigrationFailedException(msg)
138 case (Some(version), None) if version.getFormat == StorageVersion.StorageFormat.PERSISTENCE_STORE =>
139 val msg = "Migration from this storage format back to the legacy storage format is not supported."
140 throw new MigrationFailedException(msg)
141 case (Some(version), _) if version > currentBuildVersion =>
74 case Some(version) if version > currentBuildVersion =>
14275 val msg = s"Migration from ${version.str} is not supported as it is newer" +
14376 s" than ${StorageVersions.current.str}."
14477 throw new MigrationFailedException(msg)
145 case (Some(version), newStore) if version < currentBuildVersion =>
78 case Some(version) if version < currentBuildVersion =>
14679 val result = await(applyMigrationSteps(version))
14780 await(storeCurrentVersion())
14881 result
149 case (Some(version), _) if version == currentBuildVersion =>
82 case Some(version) if version == currentBuildVersion =>
15083 logger.info("No migration necessary, already at the current version")
15184 Nil
15285 case _ =>
15386 logger.info("No migration necessary, no version stored")
15487 await(storeCurrentVersion())
15588 Nil
15689 }
157 await(closeLegacyStore)
15890 migrations
15991 }.recover {
16092 case ex: MigrationFailedException => throw ex
16193 case NonFatal(ex) => throw new MigrationFailedException(s"Migration Failed: ${ex.getMessage}", ex)
16294 }
16395
16496 val migrations = Await.result(result, Duration.Inf)
16597 logger.info(s"Migration successfully applied for version ${StorageVersions.current.str}")
16698 migrations
16799 }
168100
169 // get the version out of persistence store, if that fails, get the version from the legacy store, if we're
170 // using a legacy store.
171 @SuppressWarnings(Array("all")) // async/await
172 private def getCurrentVersion(legacyStore: Option[PersistentStore]): Future[Option[StorageVersion]] =
173 async { // linter:ignore UnnecessaryElseBranch
174 await {
175 persistenceStore.map(_.storageVersion()).orElse {
176 legacyStore.map(_.load(StorageVersionName).map(_.map(v => StorageVersion.parseFrom(v.bytes.toArray))))
177 }.getOrElse(Future.successful(Some(StorageVersions.current)))
178 }
179 }
101 private def getCurrentVersion(): Future[Option[StorageVersion]] =
102 persistenceStore.storageVersion()
180103
181 @SuppressWarnings(Array("all")) // async/await
182 private def storeCurrentVersion(): Future[Done] = async { // linter:ignore UnnecessaryElseBranch
183 val legacyStore: Option[PersistentStore] = await(legacyStoreFuture)
184 val storageVersionFuture: Future[Done] = persistenceStore match {
185 case Some(store) => store.setStorageVersion(StorageVersions.current)
186 case None =>
187 val bytes = StorageVersions.current.toByteArray
188 legacyStore.map { store =>
189 store.load(StorageVersionName).flatMap {
190 case Some(entity) => store.update(entity.withNewContent(bytes.toIndexedSeq))
191 case None => store.create(StorageVersionName, bytes.toIndexedSeq)
192 }.map(_ => Done)
193 }.getOrElse(Future.successful(Done))
104 private def storeCurrentVersion(): Future[Done] =
105 persistenceStore.setStorageVersion(StorageVersions.current)
194106}
195 await(storageVersionFuture)
196 }
197107
198 @SuppressWarnings(Array("all")) // async/await
199 private def closeLegacyStore: Future[Done] = async { // linter:ignore UnnecessaryElseBranch
200 val legacyStore = await(legacyStoreFuture)
201 val future = legacyStore.map {
202 case s: PersistentStoreManagement with PrePostDriverCallback =>
203 s.postDriverTerminates.flatMap(_ => s.close())
204 case s: PersistentStoreManagement =>
205 s.close()
206 case s: PrePostDriverCallback =>
207 s.postDriverTerminates.map(_ => Done)
208 case _ =>
209 Future.successful(Done)
210 }.getOrElse(Future.successful(Done))
211 await(future)
212 }
213}
214
215108object Migration {
216109 val StorageVersionName = "internal:storage:version"
217110}
218111
219112object StorageVersions {
220113 val VersionRegex: Regex = """^(\d+)\.(\d+)\.(\d+).*""".r
221114
222115 def apply(major: Int, minor: Int, patch: Int,
223 format: StorageVersion.StorageFormat = StorageVersion.StorageFormat.LEGACY): StorageVersion = {
116 format: StorageVersion.StorageFormat = StorageVersion.StorageFormat.PERSISTENCE_STORE): StorageVersion = {
224117 StorageVersion
225118 .newBuilder()
226119 .setMajor(major)
227120 .setMinor(minor)
228121 .setPatch(patch)
229122 .setFormat(format)
230123 .build()
231124 }
232125
233126 def current: StorageVersion = {
234127 BuildInfo.version match {
235128 case VersionRegex(major, minor, patch) =>
236129 StorageVersions(
237130 major.toInt,
238131 minor.toInt,
239132 patch.toInt,
240 StorageVersion.StorageFormat.LEGACY
133 StorageVersion.StorageFormat.PERSISTENCE_STORE
241134 )
242135 }
243136 }
244137
245138 implicit class OrderedStorageVersion(val version: StorageVersion) extends AnyVal with Ordered[StorageVersion] {
246139 override def compare(that: StorageVersion): Int = {
247140 def by(left: Int, right: Int, fn: => Int): Int = if (left.compareTo(right) != 0) left.compareTo(right) else fn
248141 by(version.getFormat.getNumber, that.getFormat.getNumber,
Show All 12 Lines
View Options

src/main/scala/mesosphere/marathon/storage/migration/MigrationTo1_4_PersistenceStore.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_11.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_13.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_16.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo1_1_5.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo1_2.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo_1_4_0.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/repository/Repositories.scala

1package mesosphere.marathon.storage.repository
1package mesosphere.marathon
2package storage.repository
23
34import java.time.OffsetDateTime
4import java.util.UUID
55
66import akka.actor.ActorRefFactory
77import akka.http.scaladsl.marshalling.Marshaller
88import akka.http.scaladsl.unmarshalling.Unmarshaller
99import akka.stream.Materializer
1010import akka.stream.scaladsl.Source
1111import akka.{ Done, NotUsed }
12import mesosphere.marathon.Protos.MarathonTask
1312import mesosphere.marathon.core.event.EventSubscribers
1413import mesosphere.marathon.core.instance.Instance
1514import mesosphere.marathon.core.pod.PodDefinition
1615import mesosphere.marathon.core.storage.repository._
1716import mesosphere.marathon.core.storage.repository.impl.{ PersistenceStoreRepository, PersistenceStoreVersionedRepository }
1817import mesosphere.marathon.core.storage.store.impl.memory.{ Identity, RamId }
1918import mesosphere.marathon.core.storage.store.impl.zk.{ ZkId, ZkSerialized }
2019import mesosphere.marathon.core.storage.store.{ IdResolver, PersistenceStore }
2120import mesosphere.marathon.core.task.Task
2221import mesosphere.marathon.metrics.Metrics
2322import mesosphere.marathon.state._
24import mesosphere.marathon.storage.repository.legacy._
25import mesosphere.marathon.storage.repository.legacy.store.EntityStore
2623import mesosphere.marathon.upgrade.DeploymentPlan
2724import mesosphere.util.state.FrameworkId
28import org.apache.mesos.Protos.{ TaskID, TaskState }
2925
3026import scala.async.Async.{ async, await }
3127import scala.collection.immutable.Seq
3228import scala.concurrent.{ ExecutionContext, Future }
3329
3430trait GroupRepository {
3531 /** Fetch the root, returns an empty root if the root doesn't yet exist */
3632 def root(): Future[RootGroup]
3733 /** List previous versions of the root */
3834 def rootVersions(): Source[OffsetDateTime, NotUsed]
3935 /** Fetch a previous version of the root */
4036 def rootVersion(version: OffsetDateTime): Future[Option[RootGroup]]
4137
4238 /**
4339 * Store the root, new/updated apps and delete apps. fails if it could not
4440 * update the apps or the root, but deletion errors are ignored.
4541 */
4642 def storeRoot(rootGroup: RootGroup, updatedApps: Seq[AppDefinition], deletedApps: Seq[PathId],
4743 updatedPods: Seq[PodDefinition], deletedPods: Seq[PathId]): Future[Done]
4844
4945 def storeRootVersion(rootGroup: RootGroup, updatedApps: Seq[AppDefinition], updatedPods: Seq[PodDefinition]): Future[Done]
5046}
5147
5248object GroupRepository {
53 def legacyRepository(
54 store: (String, () => Group) => EntityStore[Group],
55 maxVersions: Int,
56 appRepository: AppRepository,
57 podRepository: PodRepository)(implicit
58 ctx: ExecutionContext,
59 metrics: Metrics): GroupEntityRepository = {
60 val entityStore = store("group:", () => RootGroup.empty)
61 new GroupEntityRepository(entityStore, maxVersions, appRepository, podRepository)
62 }
63
6449 def zkRepository(
6550 store: PersistenceStore[ZkId, String, ZkSerialized],
6651 appRepository: AppRepository,
6752 podRepository: PodRepository)(implicit
6853 ctx: ExecutionContext,
6954 mat: Materializer): StoredGroupRepositoryImpl[ZkId, String, ZkSerialized] = {
7055 import mesosphere.marathon.storage.store.ZkStoreSerialization._
7156 new StoredGroupRepositoryImpl(store, appRepository, podRepository)
Show All 9 Lines
8166 new StoredGroupRepositoryImpl(store, appRepository, podRepository)
8267 }
8368}
8469
8570trait ReadOnlyAppRepository extends ReadOnlyVersionedRepository[PathId, AppDefinition]
8671trait AppRepository extends VersionedRepository[PathId, AppDefinition] with ReadOnlyAppRepository
8772
8873object AppRepository {
89 def legacyRepository(
90 store: (String, () => AppDefinition) => EntityStore[AppDefinition],
91 maxVersions: Int)(implicit ctx: ExecutionContext, metrics: Metrics): AppEntityRepository = {
92 val entityStore = store("app:", () => AppDefinition.apply(id = AppDefinition.DefaultId))
93 new AppEntityRepository(entityStore, maxVersions)
94 }
95
9674 def zkRepository(
9775 persistenceStore: PersistenceStore[ZkId, String, ZkSerialized])(implicit ctx: ExecutionContext): AppRepositoryImpl[ZkId, String, ZkSerialized] = {
9876 import mesosphere.marathon.storage.store.ZkStoreSerialization._
9977 new AppRepositoryImpl(persistenceStore)
10078 }
10179
10280 def inMemRepository(
10381 persistenceStore: PersistenceStore[RamId, String, Identity])(implicit ctx: ExecutionContext): AppRepositoryImpl[RamId, String, Identity] = {
10482 import mesosphere.marathon.storage.store.InMemoryStoreSerialization._
10583 new AppRepositoryImpl(persistenceStore)
10684 }
10785}
10886
10987trait ReadOnlyPodRepository extends ReadOnlyVersionedRepository[PathId, PodDefinition]
11088trait PodRepository extends VersionedRepository[PathId, PodDefinition] with ReadOnlyPodRepository
11189
11290object PodRepository {
113 def legacyRepository(
114 store: (String, () => PodDefinition) => EntityStore[PodDefinition],
115 maxVersions: Int)(implicit ctx: ExecutionContext, metrics: Metrics): PodEntityRepository = {
116 val entityStore = store("pod:", () => PodDefinition.apply())
117 new PodEntityRepository(entityStore, maxVersions)
118 }
119
12091 def zkRepository(
12192 persistenceStore: PersistenceStore[ZkId, String, ZkSerialized]
12293 )(implicit ctx: ExecutionContext): PodRepositoryImpl[ZkId, String, ZkSerialized] = {
12394 import mesosphere.marathon.storage.store.ZkStoreSerialization._
12495 new PodRepositoryImpl(persistenceStore)
12596 }
12697
12798 def inMemRepository(
12899 persistenceStore: PersistenceStore[RamId, String, Identity]
129100 )(implicit ctx: ExecutionContext): PodRepositoryImpl[RamId, String, Identity] = {
130101 import mesosphere.marathon.storage.store.InMemoryStoreSerialization._
131102 new PodRepositoryImpl(persistenceStore)
132103 }
133104}
134105
135106trait DeploymentRepository extends Repository[String, DeploymentPlan]
136107
137108object DeploymentRepository {
138 def legacyRepository(store: (String, () => DeploymentPlan) => EntityStore[DeploymentPlan])(implicit
139 ctx: ExecutionContext,
140 metrics: Metrics): DeploymentEntityRepository = {
141 val entityStore = store("deployment:", () => DeploymentPlan.empty)
142 new DeploymentEntityRepository(entityStore)
143 }
144109
145110 def zkRepository(
146111 persistenceStore: PersistenceStore[ZkId, String, ZkSerialized],
147112 groupRepository: StoredGroupRepositoryImpl[ZkId, String, ZkSerialized],
148113 appRepository: AppRepositoryImpl[ZkId, String, ZkSerialized],
149114 podRepository: PodRepositoryImpl[ZkId, String, ZkSerialized],
150115 maxVersions: Int)(implicit
151116 ctx: ExecutionContext,
Show All 21 Lines
173138
174139private[storage] trait TaskRepository extends Repository[Task.Id, Task] {
175140 def tasks(appId: PathId): Source[Task.Id, NotUsed] = {
176141 ids().filter(_.runSpecId == appId)
177142 }
178143}
179144
180145object TaskRepository {
181 def legacyRepository(
182 store: (String, () => MarathonTaskState) => EntityStore[MarathonTaskState])(implicit
183 ctx: ExecutionContext,
184 metrics: Metrics): TaskEntityRepository = {
185 val entityStore = store(
186 TaskEntityRepository.storePrefix,
187 () => MarathonTaskState(MarathonTask.newBuilder().setId(UUID.randomUUID().toString).build()))
188 new TaskEntityRepository(entityStore)
189 }
190146
191147 def zkRepository(persistenceStore: PersistenceStore[ZkId, String, ZkSerialized]): TaskRepository = {
192148 import mesosphere.marathon.storage.store.ZkStoreSerialization._
193149 new TaskRepositoryImpl(persistenceStore)
194150 }
195151
196152 def inMemRepository(persistenceStore: PersistenceStore[RamId, String, Identity]): TaskRepository = {
197153 import mesosphere.marathon.storage.store.InMemoryStoreSerialization._
198154 new TaskRepositoryImpl(persistenceStore)
199155 }
200156}
201157
202158trait InstanceRepository extends Repository[Instance.Id, Instance] {
203159 def instances(runSpecId: PathId): Source[Instance.Id, NotUsed] = {
204160 ids().filter(_.runSpecId == runSpecId)
205161 }
206162}
207163
208164object InstanceRepository {
209 def legacyRepository(
210 store: (String, () => Instance) => EntityStore[Instance])(implicit ctx: ExecutionContext, metrics: Metrics): InstanceRepository = {
211 val entityStore = store(
212 "instance:",
213 () => Instance()
214 )
215 new InstanceEntityRepository(entityStore)
216 }
217165
218166 def zkRepository(persistenceStore: PersistenceStore[ZkId, String, ZkSerialized]): InstanceRepository = {
219167 import mesosphere.marathon.storage.store.ZkStoreSerialization._
220168 new InstanceRepositoryImpl(persistenceStore)
221169 }
222170
223171 def inMemRepository(persistenceStore: PersistenceStore[RamId, String, Identity]): InstanceRepository = {
224172 import mesosphere.marathon.storage.store.InMemoryStoreSerialization._
225173 new InstanceRepositoryImpl(persistenceStore)
226174 }
227175}
228176
229177trait TaskFailureRepository extends VersionedRepository[PathId, TaskFailure]
230178
231179object TaskFailureRepository {
232 def legacyRepository(
233 store: (String, () => TaskFailure) => EntityStore[TaskFailure])(implicit
234 ctx: ExecutionContext,
235 metrics: Metrics): TaskFailureEntityRepository = {
236 val entityStore = store("taskFailure:", () => TaskFailure(
237 PathId.empty,
238 TaskID.newBuilder().setValue("").build,
239 TaskState.TASK_STAGING
240 ))
241 new TaskFailureEntityRepository(entityStore, 1)
242 }
243180
244181 def zkRepository(persistenceStore: PersistenceStore[ZkId, String, ZkSerialized]): TaskFailureRepository = {
245182 import mesosphere.marathon.storage.store.ZkStoreSerialization._
246183 new TaskFailureRepositoryImpl(persistenceStore)
247184 }
248185
249186 def inMemRepository(persistenceStore: PersistenceStore[RamId, String, Identity]): TaskFailureRepository = {
250187 import mesosphere.marathon.storage.store.InMemoryStoreSerialization._
251188 new TaskFailureRepositoryImpl(persistenceStore)
252189 }
253190}
254191
255192trait FrameworkIdRepository extends SingletonRepository[FrameworkId]
256193
257194object FrameworkIdRepository {
258 def legacyRepository(store: (String, () => FrameworkId) => EntityStore[FrameworkId]): FrameworkIdEntityRepository = {
259 val entityStore = store("framework:", () => FrameworkId(UUID.randomUUID().toString))
260 new FrameworkIdEntityRepository(entityStore)
261 }
262195
263196 def zkRepository(persistenceStore: PersistenceStore[ZkId, String, ZkSerialized]): FrameworkIdRepository = {
264197 import mesosphere.marathon.storage.store.ZkStoreSerialization._
265198 new FrameworkIdRepositoryImpl(persistenceStore)
266199 }
267200
268201 def inMemRepository(persistenceStore: PersistenceStore[RamId, String, Identity]): FrameworkIdRepository = {
269202 import mesosphere.marathon.storage.store.InMemoryStoreSerialization._
270203 new FrameworkIdRepositoryImpl(persistenceStore)
271204 }
272205}
273206
274207trait EventSubscribersRepository extends SingletonRepository[EventSubscribers]
275208
276209object EventSubscribersRepository {
277 def legacyRepository(store: (String, () => EventSubscribers) => EntityStore[EventSubscribers]): EventSubscribersEntityRepository = {
278 val entityStore = store("events:", () => EventSubscribers(Set.empty[String]))
279 new EventSubscribersEntityRepository(entityStore)
280 }
281210
282211 def zkRepository(persistenceStore: PersistenceStore[ZkId, String, ZkSerialized]): EventSubscribersRepository = {
283212 import mesosphere.marathon.storage.store.ZkStoreSerialization._
284213 new EventSubscribersRepositoryImpl(persistenceStore)
285214 }
286215
287216 def inMemRepository(persistenceStore: PersistenceStore[RamId, String, Identity]): EventSubscribersRepository = {
288217 import mesosphere.marathon.storage.store.InMemoryStoreSerialization._
▲ Show 20 Lines • Show All 128 Lines • Show Last 20 Lines
View Options

src/main/scala/mesosphere/marathon/storage/repository/legacy/LegacyEntityRepository.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/repository/legacy/store/EntityStore.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/repository/legacy/store/EntityStoreCache.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/repository/legacy/store/InMemoryStore.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/repository/legacy/store/JavaFutureToFuture.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/repository/legacy/store/MarathonStore.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/repository/legacy/store/MesosStateStore.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/repository/legacy/store/PersistentStore.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/storage/repository/legacy/store/ZKStore.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/main/scala/mesosphere/marathon/upgrade/DeploymentPlan.scala

11package mesosphere.marathon
22package upgrade
33
44import java.net.URL
55import java.util.UUID
66
77import com.wix.accord._
88import com.wix.accord.dsl._
99import mesosphere.marathon.api.v2.Validation._
1010import mesosphere.marathon.core.instance.Instance
1111import mesosphere.marathon.core.pod.{ MesosContainer, PodDefinition }
1212import mesosphere.marathon.raml.{ ArgvCommand, ShellCommand }
1313import mesosphere.marathon.state._
14import mesosphere.marathon.storage.TwitterZk
15import mesosphere.marathon.storage.repository.legacy.store.{ CompressionConf, ZKData }
1614import mesosphere.marathon.stream._
1715import org.slf4j.LoggerFactory
1816
1917import scala.collection.SortedMap
2018
2119sealed trait DeploymentAction {
2220 def runSpec: RunSpec
2321}
▲ Show 20 Lines • Show All 56 Lines • ▼ Show 20 Line(s)
8078 * understand how we can guarantee that all dependencies for a step are fulfilled
8179 * by prior steps.
8280 */
8381case class DeploymentPlan(
8482 id: String,
8583 original: RootGroup,
8684 target: RootGroup,
8785 steps: Seq[DeploymentStep],
88 version: Timestamp) extends MarathonState[Protos.DeploymentPlanDefinition, DeploymentPlan] {
86 version: Timestamp) {
8987
9088 /**
9189 * Reverts this plan by applying the reverse changes to the given Group.
9290 */
9391 def revert(rootGroup: RootGroup): RootGroup = DeploymentPlanReverter.revert(original, target)(rootGroup)
9492
9593 lazy val isEmpty: Boolean = steps.isEmpty
9694
▲ Show 20 Lines • Show All 70 Lines • ▼ Show 20 Line(s)
167165 steps
168166 .map { _.actions.map(actionString).mkString(" * ", "\n * ", "") }
169167 .zipWithIndex
170168 .map { case (stepsString, index) => s"step ${index + 1}:\n$stepsString" }
171169 .mkString("\n", "\n", "")
172170 } else " NO STEPS"
173171 s"DeploymentPlan id=$id,$version$stepString\n"
174172 }
175
176 override def mergeFromProto(bytes: Array[Byte]): DeploymentPlan =
177 mergeFromProto(Protos.DeploymentPlanDefinition.parseFrom(bytes))
178
179 override def mergeFromProto(msg: Protos.DeploymentPlanDefinition): DeploymentPlan = DeploymentPlan(
180 original = RootGroup.fromProto(msg.getDeprecatedOriginal),
181 target = RootGroup.fromProto(msg.getDeprecatedTarget),
182 version = Timestamp(msg.getTimestamp),
183 id = Some(msg.getId)
184 )
185
186 override def toProto: Protos.DeploymentPlanDefinition =
187 Protos.DeploymentPlanDefinition
188 .newBuilder
189 .setId(id)
190 .setDeprecatedOriginal(original.toProto)
191 .setDeprecatedTarget(target.toProto)
192 .setTimestamp(version.toString)
193 .build()
194173}
195174
196175object DeploymentPlan {
197176 private val log = LoggerFactory.getLogger(getClass)
198177
199178 def empty: DeploymentPlan =
200179 DeploymentPlan(UUID.randomUUID().toString, RootGroup.empty, RootGroup.empty, Nil, Timestamp.now())
201180
202 def fromProto(message: Protos.DeploymentPlanDefinition): DeploymentPlan = empty.mergeFromProto(message)
203
204181 /**
205182 * Perform a "layered" topological sort of all of the run specs.
206183 * The "layered" aspect groups the run specs that have the same length of dependencies for parallel deployment.
207184 */
208185 private[upgrade] def runSpecsGroupedByLongestPath(
209186 rootGroup: RootGroup): SortedMap[Int, Set[RunSpec]] = {
210187
211188 import org.jgrapht.DirectedGraph
▲ Show 20 Lines • Show All 120 Lines • ▼ Show 20 Line(s)
332309 target,
333310 steps.result().filter(_.actions.nonEmpty),
334311 version
335312 )
336313
337314 result
338315 }
339316
340 def deploymentPlanValidator(conf: MarathonConf): Validator[DeploymentPlan] = {
341 val maxSize = conf.zooKeeperMaxNodeSize()
342 val maxSizeError = s"""The way we persist data in ZooKeeper would exceed the maximum ZK node size ($maxSize bytes).
343 |You can adjust this value via --zk_max_node_size, but make sure this value is compatible with
344 |your ZooKeeper ensemble!
345 |See: http://zookeeper.apache.org/doc/r3.3.1/zookeeperAdmin.html#Unsafe+Options""".stripMargin
346
347 val notBeTooBig = isTrue[DeploymentPlan](maxSizeError) { plan =>
348 if (conf.internalStoreBackend() == TwitterZk.StoreName) {
349 val compressionConf = CompressionConf(conf.zooKeeperCompressionEnabled(), conf.zooKeeperCompressionThreshold())
350 val zkDataProto = ZKData(s"deployment-${plan.id}", UUID.fromString(plan.id), plan.toProto.toByteArray.toIndexedSeq)
351 .toProto(compressionConf)
352 zkDataProto.toByteArray.length < maxSize
353 } else {
354 // we could try serializing the proto then gzip compressing it for the new ZK backend, but should we?
355 true
356 }
357 }
358
317 def deploymentPlanValidator(): Validator[DeploymentPlan] = {
359318 validator[DeploymentPlan] { plan =>
360319 plan.createdOrUpdatedApps as "app" is every(valid(AppDefinition.updateIsValid(plan.original)))
361 plan should notBeTooBig
362320 }
363321 }
364322}
View Options

src/test/scala/mesosphere/marathon/core/storage/store/impl/zk/ZkPersistenceStoreTest.scala

11package mesosphere.marathon
22package core.storage.store.impl.zk
33
44import java.nio.ByteOrder
55import java.nio.charset.StandardCharsets
66import java.time.{ Instant, OffsetDateTime, ZoneOffset }
77import java.util.UUID
88
99import akka.http.scaladsl.marshalling.Marshaller
1010import akka.http.scaladsl.unmarshalling.Unmarshaller
1111import akka.util.ByteString
1212import com.codahale.metrics.MetricRegistry
13import com.twitter.zk.ZNode
1413import mesosphere.AkkaUnitTest
1514import mesosphere.marathon.core.storage.store.{ IdResolver, PersistenceStoreTest, TestClass1 }
1615import mesosphere.marathon.integration.setup.ZookeeperServerTest
1716import mesosphere.marathon.metrics.Metrics
18import mesosphere.marathon.storage.migration.{ Migration, StorageVersions }
19import mesosphere.marathon.storage.repository.legacy.store.{ CompressionConf, ZKStore }
20import org.scalatest.concurrent.PatienceConfiguration.Timeout
2117
2218import scala.concurrent.duration._
23import scala.util.Random
2419
2520trait ZkTestClass1Serialization {
2621 implicit object ZkTestClass1Resolver extends IdResolver[String, TestClass1, String, ZkId] {
2722 override def fromStorageId(path: ZkId): String = path.id.replaceAll("_", "/")
2823 override def toStorageId(id: String, version: Option[OffsetDateTime]): ZkId = {
2924 ZkId(category = "test-class", id.replaceAll("/", "_"), version)
3025 }
3126 override val category: String = "test-class"
Show All 36 Lines
6863 def defaultStore: ZkPersistenceStore = {
6964 val root = UUID.randomUUID().toString
7065 val client = zkClient(namespace = Some(root))
7166 implicit val metrics = new Metrics(new MetricRegistry)
7267 new ZkPersistenceStore(client, Duration.Inf)
7368 }
7469
7570 behave like basicPersistenceStore("ZookeeperPersistenceStore", defaultStore)
76
77 it should {
78 "be able to read the storage version from the old store format" in {
79 val root = UUID.randomUUID().toString
80 rootClient.create(s"/$root").futureValue(Timeout(5.seconds))
81 implicit val metrics = new Metrics(new MetricRegistry)
82 val newStore = new ZkPersistenceStore(rootClient.usingNamespace(root), Duration.Inf)
83 val twitterClient = twitterZkClient()
84 val legacyStore = new ZKStore(twitterClient, ZNode(twitterClient, s"/$root"), CompressionConf(true, 64 * 1024),
85 8, 1024)
86
87 val version = StorageVersions(Random.nextInt, Random.nextInt, Random.nextInt)
88 legacyStore.create(Migration.StorageVersionName, version.toByteArray.toIndexedSeq).futureValue
89
90 newStore.storageVersion().futureValue.value should equal(version)
91 }
92 }
9371}
9472
View Options

src/test/scala/mesosphere/marathon/state/MarathonStoreTest.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/marathon/state/RootGroupTest.scala

Show First 20 Lines • Show All 114 Lines • ▼ Show 20 Line(s)
115115 _ => AppDefinition("/some/nested".toPath, cmd = Some("true")),
116116 Timestamp.now())
117117
118118 Then("the group with same path has been replaced by the new app definition")
119119 changed.transitiveGroupsById.keys.map(_.toString) should be(Set("/", "/some"))
120120 changed.transitiveAppsById.keys.map(_.toString) should be(Set("/some/nested"))
121121
122122 Then("the resulting group should be valid when represented in the V2 API model")
123 validate(changed)(RootGroup.valid(Set())) should be (Success)
123 validate(changed)(RootGroup.valid(Set())) should be(Success)
124124 }
125125
126126 it("cannot replace a group with apps by an app definition") {
127127 Given("an existing group /some/nested which does contain an app")
128128 val current =
129129 createRootGroup()
130130 .makeGroup("/some/nested/path".toPath)
131131 .makeGroup("/some/nested/path2".toPath)
Show All 15 Lines
147147 current.transitiveGroupsById.keys.map(_.toString) should be(
148148 Set("/", "/some", "/some/nested", "/some/nested/path", "/some/nested/path2"))
149149 changed.transitiveAppIds.map(_.toString) should be(Set("/some/nested", "/some/nested/path2/app"))
150150
151151 Then("the conflict will be detected by our V2 API model validation")
152152 val result = validate(changed)(RootGroup.valid(Set()))
153153 result.isFailure should be(true)
154154 ValidationHelper.getAllRuleConstrains(result).head
155 .message should be ("Groups and Applications may not have the same identifier.")
155 .message should be("Groups and Applications may not have the same identifier.")
156156 }
157157
158158 it("cannot replace a group with pods by an app definition") {
159159 Given("an existing group /some/nested which does contain an pod")
160160 val current =
161161 createRootGroup()
162162 .makeGroup("/some/nested/path".toPath)
163163 .makeGroup("/some/nested/path2".toPath)
Show All 16 Lines
180180 Set("/", "/some", "/some/nested", "/some/nested/path", "/some/nested/path2"))
181181 changed.transitiveAppIds.map(_.toString) should be(Set("/some/nested"))
182182 changed.transitivePodsById.keySet.map(_.toString) should be(Set("/some/nested/path2/pod"))
183183
184184 Then("the conflict will be detected by our V2 API model validation")
185185 val result = validate(changed)(RootGroup.valid(Set()))
186186 result.isFailure should be(true)
187187 ValidationHelper.getAllRuleConstrains(result).head
188 .message should be ("Groups and Applications may not have the same identifier.")
188 .message should be("Groups and Applications may not have the same identifier.")
189189 }
190190
191191 it("cannot replace a group with pods by an pod definition") {
192192 Given("an existing group /some/nested which does contain an pod")
193193 val current =
194194 createRootGroup()
195195 .makeGroup("/some/nested/path".toPath)
196196 .makeGroup("/some/nested/path2".toPath)
Show All 18 Lines
215215 Set("/", "/some", "/some/nested", "/some/nested/path", "/some/nested/path2"))
216216 changed.transitiveAppIds.map(_.toString) should be(Set.empty[String])
217217 changed.transitivePodsById.keySet.map(_.toString) should be(Set("/some/nested", "/some/nested/path2/pod"))
218218
219219 Then("the conflict will be detected by our V2 API model validation")
220220 val result = validate(changed)(RootGroup.valid(Set()))
221221 result.isFailure should be(true)
222222 ValidationHelper.getAllRuleConstrains(result).head
223 .message should be ("Groups and Pods may not have the same identifier.")
223 .message should be("Groups and Pods may not have the same identifier.")
224224 }
225225
226 it("can marshal and unmarshal from to protos") {
227 Given("a group with subgroups")
228 val now = Timestamp(11)
229 val fullVersion = VersionInfo.forNewConfig(now)
230 val app1 = AppDefinition("/test/group1/app1".toPath, args = Seq("a", "b", "c"), versionInfo = fullVersion)
231 val app2 = AppDefinition("/test/group2/app2".toPath, args = Seq("a", "b"), versionInfo = fullVersion)
232 val current = createRootGroup(
233 groups = Set(
234 createGroup("/test".toPath, groups = Set(
235 createGroup("/test/group1".toPath, Map(app1.id -> app1)),
236 createGroup("/test/group2".toPath, Map(app2.id -> app2))
237 ))))
238
239 When("the group is marshalled and unmarshalled again")
240 val group = Group.fromProto(current.toProto)
241
242 Then("the groups are identical")
243 group should equal(current)
244 }
245
246226 it("can turn a group with group dependencies into a dependency graph") {
247227 Given("a group with subgroups and dependencies")
248228 val redisApp = AppDefinition("/test/database/redis/r1".toPath)
249229 val memcacheApp = AppDefinition("/test/database/memcache/c1".toPath)
250230 val mongoApp = AppDefinition("/test/database/mongo/m1".toPath)
251231 val serviceApp1 = AppDefinition("/test/service/service1/s1".toPath)
252232 val serviceApp2 = AppDefinition("/test/service/service2/s2".toPath)
253233 val frontendApp1 = AppDefinition("/test/frontend/app1/a1".toPath)
Show All 33 Lines
287267 "/test/database/mongo/m1".toPath,
288268 "/test/database/memcache/c1".toPath,
289269 "/test/service/service1/s1".toPath,
290270 "/test/service/service2/s2".toPath,
291271 "/test/frontend/app1/a1".toPath,
292272 "/test/frontend/app2/a2".toPath,
293273 "/test/cache/c1/c1".toPath
294274 )
295 ids should equal (expectedIds)
275 ids should equal(expectedIds)
296276
297277 current.runSpecsWithNoDependencies should have size 2
298278 }
299279
300280 it("can turn a group with app dependencies into a dependency graph") {
301281 Given("a group with subgroups and dependencies")
302282 val redisApp = AppDefinition("/test/database/redis".toPath)
303283 val memcacheApp = AppDefinition("/test/database/memcache".toPath, dependencies = Set("/test/database/mongo".toPath, "/test/database/redis".toPath))
304284 val mongoApp = AppDefinition("/test/database/mongo".toPath, dependencies = Set("/test/database/redis".toPath))
305285 val serviceApp1 = AppDefinition("/test/service/srv1".toPath, dependencies = Set("/test/database/memcache".toPath))
306286 val serviceApp2 = AppDefinition("/test/service/srv2".toPath, dependencies = Set("/test/database/mongo".toPath, "/test/service/srv1".toPath))
307287 val frontendApp1 = AppDefinition("/test/frontend/app1".toPath, dependencies = Set("/test/service/srv2".toPath))
308288 val frontendApp2 = AppDefinition("/test/frontend/app2".toPath, dependencies = Set("/test/service/srv2".toPath, "/test/database/mongo".toPath, "/test/frontend/app1".toPath))
309 val cacheApp = AppDefinition("/test/cache/cache1".toPath) //has no dependencies
289 val cacheApp = AppDefinition("/test/cache/cache1".toPath)
290 //has no dependencies
310291 val current: RootGroup = createRootGroup(
311292 groups = Set(
312293 createGroup("/test".toPath, groups = Set(
313294 createGroup("/test/database".toPath, Map(
314295 redisApp.id -> redisApp,
315296 memcacheApp.id -> memcacheApp,
316297 mongoApp.id -> mongoApp
317298 )),
▲ Show 20 Lines • Show All 150 Lines • ▼ Show 20 Line(s)
468449
469450 When("group is validated")
470451 val invalidResult = validate(invalid)(RootGroup.valid(Set()))
471452
472453 Then("validation is not successful")
473454 invalidResult.isSuccess should be(false)
474455 }
475456
476457 it("Group with app in correct group is valid") {
jdefUnsubmitted
Not Done

why is this validation test removed?

jasongilanfarrAuthorUnsubmitted
Not Done

oh good catch, i meant to only remove the proto one.

477458 Given("Group with nested app of wrong path")
478459 val app = AppDefinition(PathId("/nested/foo"), cmd = Some("test"))
479460 val valid = createRootGroup(groups = Set(
480461 createGroup(PathId("nested"), apps = Map(app.id -> app))
481462 ))
482463
483464 When("group is validated")
484465 val validResult = validate(valid)(RootGroup.valid(Set()))
485466
486467 Then("validation is successful")
487468 validResult.isSuccess should be(true)
488 }
489
490 it("can be written to proto and read back again") {
491 Given("A group with subgroups, apps and pods")
492 val app = AppDefinition(PathId("/app"))
493 val pod = PodDefinition(PathId("/pod"))
494 val subGroup = Group(PathId("/group"), transitiveAppsById = Map.empty, transitivePodsById = Map.empty)
495 val group = Group(
496 id = PathId.empty,
497 apps = Map(app.id -> app),
498 pods = Map(pod.id -> pod),
499 groupsById = Map(subGroup.id -> subGroup),
500 transitiveAppsById = Map(app.id -> app),
501 transitivePodsById = Map(pod.id -> pod)
502 )
503 val rootGroup = RootGroup(
504 apps = Map(app.id -> app),
505 pods = Map(pod.id -> pod),
506 groupsById = Map(subGroup.id -> subGroup)
507 )
508
509 When("Round trip toProto -> fromProto")
510 val protoGroup = Group.fromProto(group.toProto)
511 val protoRootGroup = RootGroup.fromProto(rootGroup.toProto)
512
513 Then("The round trip is successful")
514 protoGroup should be(group)
515 protoRootGroup should be(rootGroup)
516469 }
517470 }
518471}
View Options

src/test/scala/mesosphere/marathon/storage/migration/MigrationTest.scala

11package mesosphere.marathon
22package storage.migration
33
44import akka.Done
55import akka.stream.scaladsl.Source
66import com.codahale.metrics.MetricRegistry
77import mesosphere.AkkaUnitTest
88import mesosphere.marathon.Protos.StorageVersion
99import mesosphere.marathon.core.storage.store.PersistenceStore
10import mesosphere.marathon.core.storage.store.impl.memory.InMemoryPersistenceStore
1011import mesosphere.marathon.metrics.Metrics
11import mesosphere.marathon.storage.LegacyStorageConfig
1212import mesosphere.marathon.storage.migration.StorageVersions._
13import mesosphere.marathon.storage.repository.legacy.store.{ InMemoryEntity, PersistentEntity, PersistentStore, PersistentStoreManagement }
1413import mesosphere.marathon.storage.repository._
1514import mesosphere.marathon.test.Mockito
1615import org.scalatest.GivenWhenThen
1716
1817import scala.concurrent.Future
1918
2019class MigrationTest extends AkkaUnitTest with Mockito with GivenWhenThen {
2120 implicit private def metrics = new Metrics(new MetricRegistry)
2221
2322 // scalastyle:off
2423 private[this] def migration(
25 legacyConfig: Option[LegacyStorageConfig] = None,
26 persistenceStore: Option[PersistenceStore[_, _, _]] = None,
24 persistenceStore: PersistenceStore[_, _, _] = new InMemoryPersistenceStore(),
2725 appRepository: AppRepository = mock[AppRepository],
2826 groupRepository: GroupRepository = mock[GroupRepository],
2927 deploymentRepository: DeploymentRepository = mock[DeploymentRepository],
3028 taskRepository: TaskRepository = mock[TaskRepository],
3129 instanceRepository: InstanceRepository = mock[InstanceRepository],
3230 taskFailureRepository: TaskFailureRepository = mock[TaskFailureRepository],
3331 frameworkIdRepository: FrameworkIdRepository = mock[FrameworkIdRepository],
3432 eventSubscribersRepository: EventSubscribersRepository = mock[EventSubscribersRepository]): Migration = {
35 new Migration(Set.empty, legacyConfig, persistenceStore, appRepository, groupRepository, deploymentRepository,
33 new Migration(Set.empty, persistenceStore, appRepository, groupRepository, deploymentRepository,
3634 taskRepository, instanceRepository, taskFailureRepository, frameworkIdRepository, eventSubscribersRepository)
3735 }
3836 // scalastyle:on
3937
40 val currentVersion: StorageVersion = if (StorageVersions.current < StorageVersions(1, 3, 0)) {
41 StorageVersions(1, 3, 0)
42 } else {
43 StorageVersions.current
44 }
38 val currentVersion: StorageVersion = StorageVersions.current
4539
4640 "Migration" should {
47 "be filterable by version" in {
41 // we don't have any migrations yet
42 "be filterable by version" is pendingUntilFixed {
4843 val migrate = migration()
4944 val all = migrate.migrations.filter(_._1 > StorageVersions(0, 0, 0)).sortBy(_._1)
5045 all should have size migrate.migrations.size.toLong
5146
5247 val none = migrate.migrations.filter(_._1 > StorageVersions(Int.MaxValue, 0, 0, StorageVersion.StorageFormat.PERSISTENCE_STORE))
5348 none should be('empty)
5449
5550 val some = migrate.migrations.filter(_._1 < StorageVersions(0, 10, 0))
5651 some should have size 1
5752 }
5853
5954 "migrate on an empty database will set the storage version" in {
6055 val mockedStore = mock[PersistenceStore[_, _, _]]
61 val migrate = migration(persistenceStore = Option(mockedStore))
56 val migrate = migration(persistenceStore = mockedStore)
6257
6358 mockedStore.storageVersion() returns Future.successful(None)
6459 mockedStore.setStorageVersion(any) returns Future.successful(Done)
6560
6661 migrate.migrate()
6762
6863 verify(mockedStore).storageVersion()
6964 verify(mockedStore).setStorageVersion(StorageVersions.current)
7065 noMoreInteractions(mockedStore)
7166 }
7267
73 "migrate on an empty legacy database will set the storage version" in {
74 val legacyConfig = mock[LegacyStorageConfig]
75 val mockedPersistentStore = mock[PersistentStore]
76 mockedPersistentStore.load(Migration.StorageVersionName) returns Future.successful(None)
77 mockedPersistentStore.create(eq(Migration.StorageVersionName), eq(StorageVersions.current.toByteArray.toIndexedSeq)) returns
78 Future.successful(mock[PersistentEntity])
79
80 legacyConfig.store returns mockedPersistentStore
81 val migrate = migration(legacyConfig = Some(legacyConfig), persistenceStore = None)
82
83 migrate.migrate()
84
85 verify(mockedPersistentStore, times(2)).load(Migration.StorageVersionName)
86 verify(mockedPersistentStore).create(Migration.StorageVersionName, StorageVersions.current.toByteArray.toIndexedSeq)
87 noMoreInteractions(mockedPersistentStore)
88 }
89
9068 "migrate on a database with the same version will do nothing" in {
9169 val mockedStore = mock[PersistenceStore[_, _, _]]
92 val migrate = migration(persistenceStore = Option(mockedStore))
70 val migrate = migration(persistenceStore = mockedStore)
9371
9472 val currentPersistenceVersion =
9573 StorageVersions.current.toBuilder.setFormat(StorageVersion.StorageFormat.PERSISTENCE_STORE).build()
9674 mockedStore.storageVersion() returns Future.successful(Some(currentPersistenceVersion))
9775 migrate.migrate()
9876
9977 verify(mockedStore).storageVersion()
10078 noMoreInteractions(mockedStore)
10179 }
10280
103 "migrate on a legacy database with the same version will do nothing" in {
104 val legacyConfig = mock[LegacyStorageConfig]
105 val mockedPersistentStore = mock[PersistentStore]
106 val currentVersionEntity = InMemoryEntity(Migration.StorageVersionName, 0, StorageVersions(1, 5, 0).toByteArray.toIndexedSeq)
107 mockedPersistentStore.load(Migration.StorageVersionName) returns Future.successful(Some(currentVersionEntity))
108
109 legacyConfig.store returns mockedPersistentStore
110
111 val migrate = migration(legacyConfig = Some(legacyConfig), persistenceStore = None)
112
113 migrate.migrate()
114 verify(mockedPersistentStore).load(Migration.StorageVersionName)
115 noMoreInteractions(mockedPersistentStore)
116 }
117
11881 "migrate throws an error for early unsupported versions" in {
11982 val mockedStore = mock[PersistenceStore[_, _, _]]
120 val migrate = migration(persistenceStore = Option(mockedStore))
83 val migrate = migration(persistenceStore = mockedStore)
12184 val minVersion = migrate.minSupportedStorageVersion
12285
12386 Given("An unsupported storage version")
12487 val unsupportedVersion = StorageVersions(0, 2, 0)
12588 mockedStore.storageVersion() returns Future.successful(Some(unsupportedVersion))
12689
12790 When("migrate is called for that version")
12891 val ex = intercept[RuntimeException] {
12992 migrate.migrate()
13093 }
13194
13295 Then("Migration exits with a readable error message")
13396 ex.getMessage should equal (s"Migration from versions < ${minVersion.str} are not supported. Your version: ${unsupportedVersion.str}")
13497 }
13598
13699 "migrate throws an error for versions > current" in {
137100 val mockedStore = mock[PersistenceStore[_, _, _]]
138 val migrate = migration(persistenceStore = Option(mockedStore))
101 val migrate = migration(persistenceStore = mockedStore)
139102 val minVersion = migrate.minSupportedStorageVersion
140103
141104 Given("An unsupported storage version")
142105 val unsupportedVersion = StorageVersions(Int.MaxValue, Int.MaxValue, Int.MaxValue, StorageVersion.StorageFormat.PERSISTENCE_STORE)
143106 mockedStore.storageVersion() returns Future.successful(Some(unsupportedVersion))
144107
145108 When("migrate is called for that version")
146109 val ex = intercept[RuntimeException] {
147110 migrate.migrate()
148111 }
149112
150113 Then("Migration exits with a readable error message")
151114 ex.getMessage should equal (s"Migration from ${unsupportedVersion.str} is not supported as it is newer than ${StorageVersions.current.str}.")
152115 }
153116
154 "migrate throws an error if using legacy store with a PersistenceStore version" in {
155 val legacyConfig = mock[LegacyStorageConfig]
156 val mockedPersistentStore = mock[PersistentStore]
157 legacyConfig.store returns mockedPersistentStore
158
159 Given("An unsupported storage version")
160 val unsupportedVersion = StorageVersions.current.toBuilder.setFormat(StorageVersion.StorageFormat.PERSISTENCE_STORE).build()
161 val entity = InMemoryEntity(Migration.StorageVersionName, 0, unsupportedVersion.toByteArray.toIndexedSeq)
162 mockedPersistentStore.load(Migration.StorageVersionName) returns Future.successful(Some(entity))
163
164 val migrate = migration(Some(legacyConfig), None)
165
166 When("migrate is called for that version")
167 val ex = intercept[RuntimeException] {
168 migrate.migrate()
169 }
170
171 Then("Migration exits with a readable error message")
172 ex.getMessage should equal ("Migration from this storage format back to the legacy storage format" +
173 " is not supported.")
174 }
175
176 "initializes and closes the persistent store when performing a legacy migration" in {
177 val legacyConfig = mock[LegacyStorageConfig]
178 trait Store extends PersistentStore with PersistentStoreManagement
179 val mockedPersistentStore = mock[Store]
180 val currentVersionEntity = InMemoryEntity(Migration.StorageVersionName, 0, StorageVersions(1, 5, 0).toByteArray.toIndexedSeq)
181 mockedPersistentStore.initialize() returns Future.successful(())
182 mockedPersistentStore.close() returns Future.successful(Done)
183 mockedPersistentStore.load(Migration.StorageVersionName) returns Future.successful(Some(currentVersionEntity))
184
185 legacyConfig.store returns mockedPersistentStore
186 val migrate = migration(legacyConfig = Some(legacyConfig))
187
188 migrate.migrate()
189 verify(mockedPersistentStore).initialize()
190 verify(mockedPersistentStore).close()
191 verify(mockedPersistentStore).load(Migration.StorageVersionName)
192 noMoreInteractions(mockedPersistentStore)
193 }
194
195 "migrations are executed sequentially" in {
117 // we don't have any migrations for 1.5 yet.
118 "migrations are executed sequentially" is pendingUntilFixed {
196119 val mockedStore = mock[PersistenceStore[_, _, _]]
197120 mockedStore.storageVersion() returns Future.successful(Some(StorageVersions(0, 8, 0)))
198121 mockedStore.versions(any)(any) returns Source.empty
199122 mockedStore.ids()(any) returns Source.empty
200123 mockedStore.get(any)(any, any) returns Future.successful(None)
201124 mockedStore.get(any, any)(any, any) returns Future.successful(None)
202125 mockedStore.store(any, any)(any, any) returns Future.successful(Done)
203126 mockedStore.store(any, any, any)(any, any) returns Future.successful(Done)
204127 mockedStore.setStorageVersion(any) returns Future.successful(Done)
205128
206 val migrate = migration(persistenceStore = Some(mockedStore))
129 val migrate = migration(persistenceStore = mockedStore)
207130 val result = migrate.migrate()
208131 result should be ('nonEmpty)
209132 result should be(migrate.migrations.drop(1).map(_._1))
210133 }
211134 }
212135}
View Options

src/test/scala/mesosphere/marathon/storage/migration/MigrationTo1_1_5_Test.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/marathon/storage/migration/MigrationTo1_4_PersistenceStoreTest.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_11Test.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_13Test.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo0_16Test.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo1_2Test.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo_1_40Test.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/marathon/storage/repository/GroupRepositoryTest.scala

1package mesosphere.marathon.storage.repository
1package mesosphere.marathon
2package storage.repository
23
34import java.time.OffsetDateTime
45import java.util.UUID
56
67import akka.Done
78import akka.stream.scaladsl.Sink
89import com.codahale.metrics.MetricRegistry
9import com.twitter.zk.ZNode
1010import mesosphere.AkkaUnitTest
1111import mesosphere.marathon.core.storage.store.impl.cache.{ LazyCachingPersistenceStore, LoadTimeCachingPersistenceStore }
1212import mesosphere.marathon.core.storage.store.impl.memory.InMemoryPersistenceStore
1313import mesosphere.marathon.core.storage.store.impl.zk.ZkPersistenceStore
1414import mesosphere.marathon.integration.setup.ZookeeperServerTest
1515import mesosphere.marathon.metrics.Metrics
16import mesosphere.marathon.state.{ AppDefinition, Group, PathId, Timestamp }
17import mesosphere.marathon.storage.repository.legacy.GroupEntityRepository
18import mesosphere.marathon.storage.repository.legacy.store.{ CompressionConf, EntityStore, InMemoryStore, MarathonStore, ZKStore }
16import mesosphere.marathon.state.{ AppDefinition, PathId, Timestamp }
1917import mesosphere.marathon.test.Mockito
20import org.scalatest.concurrent.PatienceConfiguration.Timeout
2118
2219import scala.collection.immutable.Seq
2320import scala.concurrent.Future
24import scala.concurrent.duration.{ Duration, _ }
21import scala.concurrent.duration.Duration
2522
2623class GroupRepositoryTest extends AkkaUnitTest with Mockito with ZookeeperServerTest {
2724 import PathId._
2825
2926 def basicGroupRepository[K, C, S](name: String, createRepo: (AppRepository, PodRepository, Int) => GroupRepository): Unit = {
3027 name should {
3128 "return an empty root if no root exists" in {
3229 val repo = createRepo(mock[AppRepository], mock[PodRepository], 1)
▲ Show 20 Lines • Show All 48 Lines • ▼ Show 20 Line(s)
8178 appRepo.store(any) returns Future.failed(exception)
8279
8380 repo.storeRoot(newRoot, apps, Nil, Nil, Nil).failed.futureValue should equal(exception)
8481 repo.root().futureValue should equal(root)
8582
8683 repo match {
8784 case s: StoredGroupRepositoryImpl[_, _, _] =>
8885 s.underlyingRoot().futureValue should equal(root)
89 case s: GroupEntityRepository =>
90 s.store.fetch("root").futureValue.value should equal(root)
9186 }
9287
9388 verify(appRepo).store(apps.head)
9489 verify(appRepo).store(apps.tail.head)
9590 noMoreInteractions(appRepo)
9691 }
9792 "store the group if deleting apps fails" in {
9893 val appRepo = mock[AppRepository]
▲ Show 20 Lines • Show All 79 Lines • ▼ Show 20 Line(s)
178173
179174 def createLoadCachingRepos(appRepository: AppRepository, podRepository: PodRepository, maxVersions: Int): GroupRepository = { // linter:ignore:UnusedParameter
180175 implicit val metrics = new Metrics(new MetricRegistry)
181176 val store = new LoadTimeCachingPersistenceStore(new InMemoryPersistenceStore())
182177 store.preDriverStarts.futureValue
183178 GroupRepository.inMemRepository(store, appRepository, podRepository)
184179 }
185180
186 def createLegacyInMemRepos(appRepository: AppRepository, podRepository: PodRepository, maxVersions: Int): GroupRepository = {
187 implicit val metrics = new Metrics(new MetricRegistry)
188 val persistentStore = new InMemoryStore()
189 def entityStore(name: String, newState: () => Group): EntityStore[Group] = {
190 new MarathonStore(persistentStore, metrics, newState, name)
191 }
192 GroupRepository.legacyRepository(entityStore, maxVersions, appRepository, podRepository)
193 }
194
195 def createLegacyZkRepos(appRepository: AppRepository, podRepository: PodRepository, maxVersions: Int): GroupRepository = {
196 implicit val metrics = new Metrics(new MetricRegistry)
197 val client = twitterZkClient()
198 val persistentStore = new ZKStore(client, ZNode(client, s"/${UUID.randomUUID().toString}"),
199 CompressionConf(true, 64 * 1024), 8, 1024)
200 persistentStore.initialize().futureValue(Timeout(5.seconds))
201 def entityStore(name: String, newState: () => Group): EntityStore[Group] = {
202 new MarathonStore(persistentStore, metrics, newState, name)
203 }
204 GroupRepository.legacyRepository(entityStore, maxVersions, appRepository, podRepository)
205 }
206
207181 behave like basicGroupRepository("InMemory", createInMemRepos)
208182 behave like basicGroupRepository("Zk", createZkRepos)
209183 behave like basicGroupRepository("LazyCaching", createLazyCachingRepos)
210184 behave like basicGroupRepository("LoadCaching", createLoadCachingRepos)
211 behave like basicGroupRepository("LegacyInMem", createLegacyInMemRepos)
212 behave like basicGroupRepository("LegacyZk", createLegacyZkRepos)
213185}
214186
View Options

src/test/scala/mesosphere/marathon/storage/repository/RepositoryTest.scala

11package mesosphere.marathon
22package storage.repository
33
44import java.util.UUID
55
66import akka.Done
77import com.codahale.metrics.MetricRegistry
8import com.twitter.zk.ZNode
98import mesosphere.AkkaUnitTest
109import mesosphere.marathon.core.storage.repository.{ Repository, VersionedRepository }
1110import mesosphere.marathon.core.storage.store.impl.cache.{ LazyCachingPersistenceStore, LazyVersionCachingPersistentStore, LoadTimeCachingPersistenceStore }
1211import mesosphere.marathon.core.storage.store.impl.memory.InMemoryPersistenceStore
1312import mesosphere.marathon.core.storage.store.impl.zk.ZkPersistenceStore
1413import mesosphere.marathon.integration.setup.ZookeeperServerTest
1514import mesosphere.marathon.metrics.Metrics
1615import mesosphere.marathon.state.{ AppDefinition, PathId, Timestamp, VersionInfo }
17import mesosphere.marathon.storage.repository.legacy.store.{ CompressionConf, EntityStore, InMemoryStore, MarathonStore, PersistentStore, ZKStore }
1816import mesosphere.marathon.stream.Sink
1917import org.scalatest.GivenWhenThen
20import org.scalatest.concurrent.PatienceConfiguration.Timeout
2118
2219import scala.concurrent.duration._
2320
2421class RepositoryTest extends AkkaUnitTest with ZookeeperServerTest with GivenWhenThen {
2522 import PathId._
2623
2724 def randomAppId = UUID.randomUUID().toString.toRootPath
2825 def randomApp = AppDefinition(randomAppId, versionInfo = VersionInfo.OnlyVersion(Timestamp.now()))
2926
30 def basic(name: String, createRepo: (Int) => Repository[PathId, AppDefinition]): Unit = {
27 def basic(name: String, createRepo: () => Repository[PathId, AppDefinition]): Unit = {
3128 s"$name:unversioned" should {
3229 "get of a non-existent value should return nothing" in {
33 val repo = createRepo(0)
30 val repo = createRepo()
3431 repo.get(randomAppId).futureValue should be('empty)
3532 }
3633 "delete should be idempotent" in {
37 val repo = createRepo(0)
34 val repo = createRepo()
3835 val id = randomAppId
3936 repo.delete(id).futureValue should be(Done)
4037 repo.delete(id).futureValue should be(Done)
4138 }
4239 "ids should return nothing" in {
43 val repo = createRepo(0)
40 val repo = createRepo()
4441 repo.ids().runWith(Sink.seq).futureValue should be('empty)
4542 }
4643 "retrieve the previously stored value for two keys" in {
47 val repo = createRepo(0)
44 val repo = createRepo()
4845 val app1 = randomApp
4946 val app2 = randomApp
5047
5148 repo.store(app1).futureValue
5249 repo.store(app2).futureValue
5350
5451 repo.get(app1.id).futureValue.value should equal(app1)
5552 repo.get(app2.id).futureValue.value should equal(app2)
5653 }
5754 "store with the same id should update the object" in {
58 val repo = createRepo(0)
55 val repo = createRepo()
5956 val start = randomApp
6057 val end = start.copy(cmd = Some("abcd"))
6158
6259 repo.store(start).futureValue
6360 repo.store(end).futureValue
6461
6562 repo.get(end.id).futureValue.value should equal(end)
6663 repo.get(start.id).futureValue.value should equal(end)
6764 repo.all().runWith(Sink.seq).futureValue should equal(Seq(end))
6865 }
6966 "stored objects should list in the ids and all" in {
70 val repo = createRepo(0)
67 val repo = createRepo()
7168 val app1 = randomApp
7269 val app2 = randomApp
7370
7471 Given("Two objects")
7572 repo.store(app1).futureValue
7673 repo.store(app2).futureValue
7774
7875 Then("They should list in the ids and all")
7976 repo.ids().runWith(Sink.seq).futureValue should contain theSameElementsAs Seq(app1.id, app2.id)
8077 repo.all().runWith(Sink.seq).futureValue should contain theSameElementsAs Seq(app1, app2)
8178
8279 When("one of them is removed")
8380 repo.delete(app2.id).futureValue
8481
8582 Then("it should no longer be in the ids")
8683 repo.ids().runWith(Sink.seq).futureValue should contain theSameElementsAs Seq(app1.id)
8784 repo.all().runWith(Sink.seq).futureValue should contain theSameElementsAs Seq(app1)
8885 }
8986 }
9087 }
9188
92 def versioned(name: String, createRepo: (Int) => VersionedRepository[PathId, AppDefinition]): Unit = {
89 def versioned(name: String, createRepo: () => VersionedRepository[PathId, AppDefinition]): Unit = {
9390 s"$name:versioned" should {
9491 "list no versions when empty" in {
95 val repo = createRepo(2)
92 val repo = createRepo()
9693 repo.versions(randomAppId).runWith(Sink.seq).futureValue should be('empty)
9794 }
9895 "list and retrieve the current and all previous versions up to the cap" in {
99 val repo = createRepo(3)
96 val repo = createRepo()
10097 val app = randomApp.copy(versionInfo = VersionInfo.OnlyVersion(Timestamp(1)))
10198 val lastVersion = app.copy(versionInfo = VersionInfo.OnlyVersion(Timestamp(4)))
10299 // two previous versions and current (so app is gone)
103100 val versions = Seq(
104101 app,
105102 app.copy(versionInfo = VersionInfo.OnlyVersion(Timestamp(2))),
106103 app.copy(versionInfo = VersionInfo.OnlyVersion(Timestamp(3))),
107104 lastVersion)
Show All 25 Lines
133130
134131 When("deleting all")
135132 repo.delete(app.id).futureValue
136133
137134 Then("No versions remain")
138135 repo.versions(app.id).runWith(Sink.seq).futureValue should be('empty)
139136 }
140137 "be able to store a specific version" in {
141 val repo = createRepo(2)
138 val repo = createRepo()
142139 val app = randomApp
143140 repo.storeVersion(app).futureValue
144141
145142 repo.versions(app.id).runWith(Sink.seq).futureValue should
146143 contain theSameElementsAs Seq(app.version.toOffsetDateTime)
147144 repo.get(app.id).futureValue should be ('empty)
148145 repo.getVersion(app.id, app.version.toOffsetDateTime).futureValue.value should equal(app)
149146 }
150147 }
151148 }
152149
153 def createLegacyRepo(maxVersions: Int, store: PersistentStore): AppRepository = {
150 def createInMemRepo(): AppRepository = {
154151 implicit val metrics = new Metrics(new MetricRegistry)
155 def entityStore(name: String, newState: () => AppDefinition): EntityStore[AppDefinition] = {
156 new MarathonStore(store, metrics, newState, name)
157 }
158 AppRepository.legacyRepository(entityStore, maxVersions)
159 }
160
161 def zkStore(): PersistentStore = {
162 implicit val metrics = new Metrics(new MetricRegistry)
163 val client = twitterZkClient()
164 val persistentStore = new ZKStore(client, ZNode(client, s"/${UUID.randomUUID().toString}"),
165 CompressionConf(true, 64 * 1024), 8, 1024)
166 persistentStore.initialize().futureValue(Timeout(5.seconds))
167 persistentStore
168 }
169
170 def createInMemRepo(maxVersions: Int): AppRepository = { // linter:ignore:UnusedParameter
171 implicit val metrics = new Metrics(new MetricRegistry)
172152 AppRepository.inMemRepository(new InMemoryPersistenceStore())
173153 }
174154
175 def createLoadTimeCachingRepo(maxVersions: Int): AppRepository = { // linter:ignore:UnusedParameter
155 def createLoadTimeCachingRepo(): AppRepository = {
176156 implicit val metrics = new Metrics(new MetricRegistry)
177157 val cached = new LoadTimeCachingPersistenceStore(new InMemoryPersistenceStore())
178158 cached.preDriverStarts.futureValue
179159 AppRepository.inMemRepository(cached)
180160 }
181161
182 def createZKRepo(maxVersions: Int): AppRepository = { // linter:ignore:UnusedParameter
162 def createZKRepo(): AppRepository = {
183163 implicit val metrics = new Metrics(new MetricRegistry)
184164 val root = UUID.randomUUID().toString
185165 val rootClient = zkClient(namespace = Some(root))
186166 val store = new ZkPersistenceStore(rootClient, Duration.Inf)
187167 AppRepository.zkRepository(store)
188168 }
189169
190 def createLazyCachingRepo(maxVersions: Int): AppRepository = { // linter:ignore:UnusedParameter
170 def createLazyCachingRepo(): AppRepository = {
191171 implicit val metrics = new Metrics(new MetricRegistry)
192172 AppRepository.inMemRepository(LazyCachingPersistenceStore(new InMemoryPersistenceStore()))
193173 }
194174
195 def createLazyVersionCachingRepo(maxVersions: Int): AppRepository = { // linter:ignore:UnusedParameter
175 def createLazyVersionCachingRepo(): AppRepository = {
196176 implicit val metrics = new Metrics(new MetricRegistry)
197177 AppRepository.inMemRepository(LazyVersionCachingPersistentStore(new InMemoryPersistenceStore()))
198178 }
199179
200 behave like basic("InMemEntity", createLegacyRepo(_, new InMemoryStore()))
201 behave like basic("ZkEntity", createLegacyRepo(_, zkStore()))
202180 behave like basic("InMemoryPersistence", createInMemRepo)
203181 behave like basic("ZkPersistence", createZKRepo)
204182 behave like basic("LoadTimeCachingPersistence", createLoadTimeCachingRepo)
205183 behave like basic("LazyCachingPersistence", createLazyCachingRepo)
206184
207 behave like versioned("InMemEntity", createLegacyRepo(_, new InMemoryStore()))
208 behave like versioned("ZkEntity", createLegacyRepo(_, zkStore()))
209185 behave like versioned("InMemoryPersistence", createInMemRepo)
210186 behave like versioned("ZkPersistence", createZKRepo)
211187 behave like versioned("LoadTimeCachingPersistence", createLoadTimeCachingRepo)
212188 behave like versioned("LazyCachingPersistence", createLazyCachingRepo)
213189 behave like versioned("LazyVersionCachingPersistence", createLazyVersionCachingRepo)
214190}
View Options

src/test/scala/mesosphere/marathon/storage/repository/SingletonRepositoryTest.scala

1package mesosphere.marathon.storage.repository
1package mesosphere.marathon
2package storage.repository
23
34import java.util.UUID
45
56import akka.Done
67import com.codahale.metrics.MetricRegistry
7import com.twitter.zk.ZNode
88import mesosphere.AkkaUnitTest
99import mesosphere.marathon.core.storage.repository.SingletonRepository
1010import mesosphere.marathon.core.storage.store.impl.cache.{ LazyCachingPersistenceStore, LoadTimeCachingPersistenceStore }
1111import mesosphere.marathon.core.storage.store.impl.memory.InMemoryPersistenceStore
1212import mesosphere.marathon.core.storage.store.impl.zk.ZkPersistenceStore
1313import mesosphere.marathon.integration.setup.ZookeeperServerTest
1414import mesosphere.marathon.metrics.Metrics
15import mesosphere.marathon.storage.repository.legacy.store.{ CompressionConf, EntityStore, InMemoryStore, MarathonStore, PersistentStore, ZKStore }
1615import mesosphere.util.state.FrameworkId
17import org.scalatest.concurrent.PatienceConfiguration.Timeout
1816
1917import scala.concurrent.duration._
2018
2119class SingletonRepositoryTest extends AkkaUnitTest with ZookeeperServerTest {
2220 def basic(name: String, createRepo: => SingletonRepository[FrameworkId]): Unit = {
2321 name should {
2422 "return none if nothing has been stored" in {
2523 val repo = createRepo
Show All 14 Lines
4038 val id = FrameworkId(UUID.randomUUID().toString)
4139 repo.store(id).futureValue
4240 repo.delete().futureValue should be(Done)
4341 repo.get().futureValue should be ('empty)
4442 }
4543 }
4644 }
4745
48 def createLegacyRepo(store: PersistentStore): FrameworkIdRepository = {
49 implicit val metrics = new Metrics(new MetricRegistry)
50 def entityStore(name: String, newState: () => FrameworkId): EntityStore[FrameworkId] = {
51 new MarathonStore(store, metrics, newState, name)
52 }
53 FrameworkIdRepository.legacyRepository(entityStore)
54 }
55
56 def zkStore(): PersistentStore = {
57 implicit val metrics = new Metrics(new MetricRegistry)
58 val client = twitterZkClient()
59 val persistentStore = new ZKStore(client, ZNode(client, s"/${UUID.randomUUID().toString}"),
60 CompressionConf(true, 64 * 1024), 8, 1024)
61 persistentStore.initialize().futureValue(Timeout(5.seconds))
62 persistentStore
63 }
64
6546 def createInMemRepo(): FrameworkIdRepository = {
6647 implicit val metrics = new Metrics(new MetricRegistry)
6748 FrameworkIdRepository.inMemRepository(new InMemoryPersistenceStore())
6849 }
6950
7051 def createLoadTimeCachingRepo(): FrameworkIdRepository = {
7152 implicit val metrics = new Metrics(new MetricRegistry)
7253 val cached = new LoadTimeCachingPersistenceStore(new InMemoryPersistenceStore())
7354 cached.preDriverStarts.futureValue
7455 FrameworkIdRepository.inMemRepository(cached)
7556 }
7657
7758 def createZKRepo(): FrameworkIdRepository = {
7859 implicit val metrics = new Metrics(new MetricRegistry)
7960 FrameworkIdRepository.zkRepository(new ZkPersistenceStore(zkClient(), 10.seconds))
8061 }
8162
8263 def createLazyCachingRepo(): FrameworkIdRepository = {
8364 implicit val metrics = new Metrics(new MetricRegistry)
8465 FrameworkIdRepository.inMemRepository(LazyCachingPersistenceStore(new InMemoryPersistenceStore()))
8566 }
8667
87 behave like basic("InMemEntity", createLegacyRepo(new InMemoryStore()))
88 behave like basic("ZkEntity", createLegacyRepo(zkStore()))
8968 behave like basic("InMemoryPersistence", createInMemRepo())
9069 behave like basic("ZkPersistence", createZKRepo())
9170 behave like basic("LoadTimeCachingPersistence", createLoadTimeCachingRepo())
9271 behave like basic("LazyCachingPersistence", createLazyCachingRepo())
9372}
View Options

src/test/scala/mesosphere/marathon/storage/repository/legacy/store/EntityStoreCacheTest.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/marathon/storage/repository/legacy/store/ZKStoreTest.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/marathon/tasks/InstanceTrackerImplTest.scala

11package mesosphere.marathon
22package tasks
33
44import com.codahale.metrics.MetricRegistry
55import mesosphere.AkkaFunTest
66import mesosphere.marathon.core.base.ConstantClock
77import mesosphere.marathon.core.instance.update.InstanceUpdateOperation
8import mesosphere.marathon.core.instance.{Instance, TestInstanceBuilder}
8import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
99import mesosphere.marathon.core.leadership.AlwaysElectedLeadershipModule
1010import mesosphere.marathon.core.storage.store.impl.memory.InMemoryPersistenceStore
1111import mesosphere.marathon.core.task.Task
12import mesosphere.marathon.core.task.tracker.{InstanceTracker, TaskStateOpProcessor}
12import mesosphere.marathon.core.task.tracker.{ InstanceTracker, TaskStateOpProcessor }
1313import mesosphere.marathon.metrics.Metrics
1414import mesosphere.marathon.state.PathId
1515import mesosphere.marathon.state.PathId.StringPathId
1616import mesosphere.marathon.storage.repository.InstanceRepository
1717import mesosphere.marathon.stream.Sink
18import mesosphere.marathon.test.{MarathonShutdownHookSupport, MarathonTestHelper}
18import mesosphere.marathon.test.{ MarathonShutdownHookSupport, MarathonTestHelper }
1919import mesosphere.mesos.protos.Implicits._
2020import mesosphere.mesos.protos.TextAttribute
2121import org.apache.mesos.Protos
22import org.apache.mesos.Protos.{TaskState, TaskStatus}
22import org.apache.mesos.Protos.{ TaskState, TaskStatus }
2323import org.mockito.Mockito.spy
2424
2525import scala.collection.immutable.Seq
2626
2727class InstanceTrackerImplTest extends AkkaFunTest with MarathonShutdownHookSupport {
2828
2929 val TEST_APP_NAME = PathId("/foo")
3030 var instanceTracker: InstanceTracker = _
▲ Show 20 Lines • Show All 459 Lines • Show Last 20 Lines
View Options

src/test/scala/mesosphere/marathon/test/MarathonTestHelper.scala

Show All 9 Lines
1010import mesosphere.marathon.Protos.Constraint.Operator
1111import mesosphere.marathon.api.JsonTestHelper
1212import mesosphere.marathon.api.serialization.LabelsSerializer
1313import mesosphere.marathon.core.base.Clock
1414import mesosphere.marathon.core.condition.Condition
1515import mesosphere.marathon.core.instance.Instance
1616import mesosphere.marathon.core.instance.Instance.InstanceState
1717import mesosphere.marathon.core.instance.update.InstanceChangeHandler
18import mesosphere.marathon.core.launcher.impl.{ReservationLabels, TaskLabels}
18import mesosphere.marathon.core.launcher.impl.{ ReservationLabels, TaskLabels }
1919import mesosphere.marathon.core.leadership.LeadershipModule
2020import mesosphere.marathon.core.storage.store.impl.memory.InMemoryPersistenceStore
2121import mesosphere.marathon.core.task.Task
22import mesosphere.marathon.core.task.tracker.{InstanceTracker, InstanceTrackerModule}
22import mesosphere.marathon.core.task.tracker.{ InstanceTracker, InstanceTrackerModule }
2323import mesosphere.marathon.metrics.Metrics
2424import mesosphere.marathon.raml.Resources
25import mesosphere.marathon.state.Container.{Docker, PortMapping}
25import mesosphere.marathon.state.Container.{ Docker, PortMapping }
2626import mesosphere.marathon.state.PathId._
2727import mesosphere.marathon.state._
2828import mesosphere.marathon.storage.repository.InstanceRepository
2929import mesosphere.marathon.stream._
30import mesosphere.mesos.protos.{FrameworkID, OfferID, Range, RangesResource, Resource, ScalarResource, SlaveID}
30import mesosphere.mesos.protos.{ FrameworkID, OfferID, Range, RangesResource, Resource, ScalarResource, SlaveID }
3131import mesosphere.util.state.FrameworkId
32import org.apache.mesos.Protos.Resource.{DiskInfo, ReservationInfo}
32import org.apache.mesos.Protos.Resource.{ DiskInfo, ReservationInfo }
3333import org.apache.mesos.Protos._
34import org.apache.mesos.{Protos => Mesos}
34import org.apache.mesos.{ Protos => Mesos }
3535import play.api.libs.json.Json
3636
3737import scala.concurrent.ExecutionContext
3838import scala.util.Random
3939
4040object MarathonTestHelper {
4141
4242 import mesosphere.mesos.protos.Implicits._
▲ Show 20 Lines • Show All 445 Lines • Show Last 20 Lines
View Options

src/test/scala/mesosphere/marathon/upgrade/DeploymentManagerTest.scala

1package mesosphere.marathon.upgrade
1package mesosphere.marathon
2package upgrade
23
34import java.util.concurrent.LinkedBlockingDeque
45
56import akka.Done
67import akka.actor.{ ActorRef, Props }
78import akka.event.EventStream
89import akka.stream.scaladsl.Source
910import akka.testkit.TestActor.{ AutoPilot, NoAutoPilot }
1011import akka.testkit.{ ImplicitSender, TestActor, TestActorRef, TestProbe }
1112import akka.util.Timeout
1213import com.codahale.metrics.MetricRegistry
1314import mesosphere.marathon.MarathonSchedulerActor.{ CommandFailed, DeploymentStarted, LoadedDeploymentsOnLeaderElection }
1415import mesosphere.marathon.core.health.HealthCheckManager
1516import mesosphere.marathon.core.launchqueue.LaunchQueue
1617import mesosphere.marathon.core.leadership.AlwaysElectedLeadershipModule
1718import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
19import mesosphere.marathon.core.storage.store.impl.memory.InMemoryPersistenceStore
1820import mesosphere.marathon.core.task.termination.KillService
1921import mesosphere.marathon.core.task.tracker.InstanceTracker
2022import mesosphere.marathon.io.storage.StorageProvider
2123import mesosphere.marathon.metrics.Metrics
24import mesosphere.marathon.state.AppDefinition
2225import mesosphere.marathon.state.PathId._
23import mesosphere.marathon.state.{ AppDefinition, PathId }
24import mesosphere.marathon.storage.repository.legacy.AppEntityRepository
25import mesosphere.marathon.storage.repository.legacy.store.{ InMemoryStore, MarathonStore }
2626import mesosphere.marathon.storage.repository.{ AppRepository, DeploymentRepository }
2727import mesosphere.marathon.test.{ GroupCreation, MarathonActorSupport, MarathonTestHelper, Mockito }
2828import mesosphere.marathon.upgrade.DeploymentActor.Cancel
2929import mesosphere.marathon.upgrade.DeploymentManager._
30import mesosphere.marathon.{ MarathonConf, SchedulerActions }
3130import org.apache.mesos.SchedulerDriver
3231import org.rogach.scallop.ScallopConf
3332import org.scalatest.concurrent.Eventually
3433import org.scalatest.time.{ Seconds, Span }
3534import org.scalatest.{ BeforeAndAfter, FunSuiteLike, Matchers }
3635import org.slf4j.LoggerFactory
3736
3837import scala.concurrent.duration._
▲ Show 20 Lines • Show All 136 Lines • ▼ Show 20 Line(s)
175174 probe.setAutoPilot(new AutoPilot {
176175 override def run(sender: ActorRef, msg: Any): AutoPilot = msg match {
177176 case Cancel(_) =>
178177 system.stop(probe.ref)
179178 NoAutoPilot
180179 }
181180 })
182181
183 val ex = new Exception
182 val ex = new Exception("")
184183
185184 val res = manager.underlyingActor.stopActor(probe.ref, ex)
186185
187186 Await.result(res, 5.seconds) should be(true)
188187 }
189188
190189 test("Cancel deployment") {
191190 val f = new Fixture
▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Line(s)
233232
234233 val driver: SchedulerDriver = mock[SchedulerDriver]
235234 val deploymentRepo = mock[DeploymentRepository]
236235 val eventBus: EventStream = mock[EventStream]
237236 val launchQueue: LaunchQueue = mock[LaunchQueue]
238237 val config: MarathonConf = new ScallopConf(Seq("--master", "foo")) with MarathonConf {
239238 verify()
240239 }
241 val metrics: Metrics = new Metrics(new MetricRegistry)
240 implicit val metrics: Metrics = new Metrics(new MetricRegistry)
241 implicit val ctx: ExecutionContext = ExecutionContext.global
242242 val taskTracker: InstanceTracker = MarathonTestHelper.createTaskTracker(
243243 AlwaysElectedLeadershipModule.forActorSystem(system)
244244 )
245245 val taskKillService: KillService = mock[KillService]
246246 val scheduler: SchedulerActions = mock[SchedulerActions]
247 val appRepo: AppRepository = new AppEntityRepository(
248 new MarathonStore[AppDefinition](new InMemoryStore, metrics, () => AppDefinition(id = PathId("/test")), prefix = "app:"),
249 0
250 )(ExecutionContext.global, metrics)
247 val appRepo: AppRepository = AppRepository.inMemRepository(new InMemoryPersistenceStore())
251248 val storage: StorageProvider = mock[StorageProvider]
252249 val hcManager: HealthCheckManager = mock[HealthCheckManager]
253250 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
254251
255252 // A method that returns dummy props. Used to control the deployments progress. Otherwise the tests become racy
256253 // and depending on when DeploymentActor sends DeploymentFinished message.
257254 val deploymentActorProps: (Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any) => Props = (_, _, _, _, _, _, _, _, _, _, _) => TestActor.props(new LinkedBlockingDeque())
258255
Show All 18 Lines
View Options

src/test/scala/mesosphere/marathon/upgrade/DeploymentPlanTest.scala

1package mesosphere.marathon.upgrade
1package mesosphere.marathon
2package upgrade
23
34import com.wix.accord._
4import mesosphere.marathon._
5import mesosphere.marathon.api.v2.ValidationHelper
65import mesosphere.marathon.core.instance.TestInstanceBuilder
7import mesosphere.marathon.state.VersionInfo._
86import mesosphere.marathon.state.PathId._
97import mesosphere.marathon.state.VersionInfo._
108import mesosphere.marathon.state._
11import mesosphere.marathon.storage.TwitterZk
129import mesosphere.marathon.test.{ GroupCreation, MarathonSpec, MarathonTestHelper, Mockito }
1310import org.apache.mesos.{ Protos => mesos }
1411import org.scalatest.{ GivenWhenThen, Matchers }
1512
1613import scala.collection.immutable.Seq
1714
1815class DeploymentPlanTest extends MarathonSpec with Matchers with GivenWhenThen with Mockito with GroupCreation {
1916
▲ Show 20 Lines • Show All 403 Lines • ▼ Show 20 Line(s)
423420 val app2 = f.validResident.copy(upgradeStrategy = AppDefinition.DefaultUpgradeStrategy)
424421 val rootGroup2 = f.rootGroup.updateApps(PathId("/test"), _ => Map(app2.id -> app2), f.rootGroup.version)
425422 val plan2 = DeploymentPlan(f.rootGroup, rootGroup2)
426423
427424 Then("The deployment is not valid")
428425 validate(plan2)(f.validator).isSuccess should be(false)
429426 }
430427
431 test("Deployment plan validation fails if the deployment plan is too big") {
432 Given("All options are supplied and we have a valid group change, but the deployment plan size limit is small")
433 val f = new Fixture()
434 val validator = DeploymentPlan.deploymentPlanValidator(MarathonTestHelper.defaultConfig(
435 internalStorageBackend = Some(TwitterZk.StoreName),
436 maxZkNodeSize = Some(1)))
437
438 When("We create a scale deployment")
439 val app = f.validResident.copy(instances = 123)
440 val rootGroup = f.rootGroup.updateApps(PathId.empty, _ => Map(app.id -> app), f.rootGroup.version)
441 val plan = DeploymentPlan(f.rootGroup, rootGroup)
442
443 Then("The deployment is valid")
444 val result = validate(plan)(validator)
445 val violations = ValidationHelper.getAllRuleConstrains(result)
446
447 result.isFailure should be(true)
448 ValidationHelper.getAllRuleConstrains(result).head.message should be ("""The way we persist data in ZooKeeper would exceed the maximum ZK node size (1 bytes).
449 |You can adjust this value via --zk_max_node_size, but make sure this value is compatible with
450 |your ZooKeeper ensemble!
451 |See: http://zookeeper.apache.org/doc/r3.3.1/zookeeperAdmin.html#Unsafe+Options""".stripMargin)
452 }
453
454428 class Fixture {
455429 def persistentVolume(path: String) = PersistentVolume(path, PersistentVolumeInfo(123), mesos.Volume.Mode.RW)
456430 val zero = UpgradeStrategy(0, 0)
457431
458432 def residentApp(id: String, volumes: Seq[PersistentVolume]): AppDefinition = {
459433 AppDefinition(
460434 id = PathId(id),
461435 container = Some(Container.Mesos(volumes)),
462436 residency = Some(Residency(123, Protos.ResidencyDefinition.TaskLostBehavior.RELAUNCH_AFTER_TIMEOUT))
463437 )
464438 }
465439 val vol1 = persistentVolume("foo")
466440 val vol2 = persistentVolume("bla")
467441 val vol3 = persistentVolume("test")
468442 val validResident = residentApp("/app1", Seq(vol1, vol2)).copy(upgradeStrategy = zero)
469443 val rootGroup = createRootGroup(groups = Set(createGroup(PathId("/test"), apps = Map(validResident.id -> validResident))))
470444 val marathonConf = MarathonTestHelper.defaultConfig()
471 val validator = DeploymentPlan.deploymentPlanValidator(marathonConf)
445 val validator = DeploymentPlan.deploymentPlanValidator()
472446 }
473447}
View Options

src/test/scala/mesosphere/marathon/upgrade/TaskStartActorTest.scala

1package mesosphere.marathon.upgrade
1package mesosphere.marathon
2package upgrade
23
3import akka.testkit.{TestActorRef, TestProbe}
4import akka.testkit.{ TestActorRef, TestProbe }
45import com.codahale.metrics.MetricRegistry
56import mesosphere.marathon.core.condition.Condition
6import mesosphere.marathon.core.condition.Condition.{Failed, Running}
7import mesosphere.marathon.core.event.{DeploymentStatus, _}
7import mesosphere.marathon.core.condition.Condition.{ Failed, Running }
8import mesosphere.marathon.core.event.{ DeploymentStatus, _ }
89import mesosphere.marathon.core.health.MesosCommandHealthCheck
910import mesosphere.marathon.core.instance.update.InstanceUpdateOperation
10import mesosphere.marathon.core.instance.{Instance, TestInstanceBuilder}
11import mesosphere.marathon.core.instance.{ Instance, TestInstanceBuilder }
1112import mesosphere.marathon.core.launcher.impl.LaunchQueueTestHelper
1213import mesosphere.marathon.core.launchqueue.LaunchQueue
1314import mesosphere.marathon.core.leadership.AlwaysElectedLeadershipModule
1415import mesosphere.marathon.core.readiness.ReadinessCheckExecutor
15import mesosphere.marathon.core.task.tracker.{InstanceCreationHandler, InstanceTracker}
16import mesosphere.marathon.core.task.tracker.{ InstanceCreationHandler, InstanceTracker }
1617import mesosphere.marathon.metrics.Metrics
1718import mesosphere.marathon.state.PathId._
18import mesosphere.marathon.state.{AppDefinition, Command, Timestamp}
19import mesosphere.marathon.test.{MarathonActorSupport, MarathonTestHelper, Mockito}
20import mesosphere.marathon.{SchedulerActions, TaskUpgradeCanceledException}
21import mesosphere.{IntegrationTag, Unstable}
22import org.mockito.Mockito.{spy, when}
19import mesosphere.marathon.state.{ AppDefinition, Command, Timestamp }
20import mesosphere.marathon.test.{ MarathonActorSupport, MarathonTestHelper, Mockito }
21import mesosphere.{ IntegrationTag, Unstable }
22import org.mockito.Mockito.{ spy, when }
2323import org.scalatest.concurrent.ScalaFutures
24import org.scalatest.{BeforeAndAfter, FunSuiteLike, Matchers}
24import org.scalatest.{ BeforeAndAfter, FunSuiteLike, Matchers }
2525
2626import scala.concurrent.duration._
27import scala.concurrent.{Await, Promise}
27import scala.concurrent.{ Await, Promise }
2828
2929class TaskStartActorTest
3030 extends MarathonActorSupport
3131 with FunSuiteLike
3232 with Matchers
3333 with Mockito
3434 with ScalaFutures
3535 with BeforeAndAfter {
▲ Show 20 Lines • Show All 215 Lines • ▼ Show 20 Line(s)
251251 }
252252
253253 class Fixture {
254254
255255 val scheduler: SchedulerActions = mock[SchedulerActions]
256256 val launchQueue: LaunchQueue = mock[LaunchQueue]
257257 val metrics: Metrics = new Metrics(new MetricRegistry)
258258 val leadershipModule = AlwaysElectedLeadershipModule.forActorSystem(system)
259 val taskTrackerModule = MarathonTestHelper.createTaskTrackerModule(leadershipModule)
259 val taskTrackerModule = MarathonTestHelper.createTaskTrackerModule(
260 leadershipModule, metrics = metrics)
260261 val taskTracker: InstanceTracker = spy(taskTrackerModule.instanceTracker)
261262 val taskCreationHandler: InstanceCreationHandler = taskTrackerModule.instanceCreationHandler
262263 val deploymentManager = TestProbe()
263264 val status: DeploymentStatus = mock[DeploymentStatus]
264265 val readinessCheckExecutor: ReadinessCheckExecutor = mock[ReadinessCheckExecutor]
265266
266267 def instanceChange(app: AppDefinition, id: Instance.Id, condition: Condition): InstanceChanged = {
267268 val instance: Instance = mock[Instance]
Show All 13 Lines
View Options

src/test/scala/mesosphere/util/state/PersistentStoreTest.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/util/state/memory/InMemoryStoreTest.scala

  • This file was deleted.
This file was completely deleted. Show File Contents
View Options

src/test/scala/mesosphere/util/state/mesos/MesosStateStoreTest.scala

  • This file was deleted.
This file was completely deleted. Show File Contents