src/main/scala/mesosphere/util/state/RuntimeConfiguration.scala
8 ↗	(On Diff #2524)	Can we make this in raml please? I was playing with using src/main/raml at one point so we could make a clear delineation that some stuff is internal types.

src/main/scala/mesosphere/marathon/storage/store/ZkStoreSerialization.scala
190	minor, but there are a bunch of constants for UTF-8, e.g. StandardCharsets.UTF8

src/main/scala/mesosphere/marathon/api/v2/LeaderResource.scala
48–54	this would probably be more clear (vs nested `withValid` and the validators outside the block, several lines away): assumeValid { validate(block)(optional(UriIO.valid)) validate(restore)(optional(UriIO.valid)) result(...) ... }

docs/docs/rest-api/public/api/v2/leader.raml
27	I think we should do better here explaining how `delete` could trigger a re-election. We don't need to duplicate word for word what we have written somewhere else, but this is still too brief in my opinion.

docs/docs/rest-api/public/api/v2/types/pragma.raml
36	We need some kind of compile-time separation between public and internal types. (a) I wonder if internal types should actually live in a separate package? Grouping them together in the same package as public API types is confusing and will lead to mistakes down the line. (b) What rules can we implement in the RAML generator to enforce such separation? For example, does it make sense to implement a check that public types cannot refer to internal types? Check that API endpoints don't refer to any internal types? At the very least please file a JIRA to track this concern

Diff	ID	Base	Description	Created	Lint	Unit
Base			Base
Diff 1	2524	2b81cb4		Apr 5 2017, 9:19 AM	★	★
Diff 2	2630	020fb9b	- Describe RuntimeConfiguration in raml and use the generated type.	Apr 11 2017, 10:19 AM	★	★
Diff 3	2649	020fb9b	- Add API documentation for the leader endpoint.	Apr 12 2017, 3:19 PM	★	★
Diff 4	2750	08cd1ec	- Describe RuntimeConfiguration in raml and use the generated type.	Apr 19 2017, 6:49 PM	★	★
Diff 5	2774	c0ecf80	rebase	Apr 20 2017, 8:35 AM	★	★
Diff 6	2777	6813225	rebase	Apr 20 2017, 3:46 PM	★	★
Diff 7	2809	cf55b88	Improve documentation on leader abdication.	Apr 21 2017, 8:41 AM	★	★
Diff 8	2867	db87e10	rebase	Apr 24 2017, 3:03 PM	★	★
Diff 9	2875	db87e10	rMARATHON31674fbe95931d1edfa2c4b55d3659cae0ce39df	Apr 24 2017, 7:16 PM	★	★

Commit	Tree	Parents	Author	Summary	Date
7fdd35577fba	6f3f703f783d	c7e66d1f127a	Matthias Veit	Improve documentation on leader abdication.	Apr 21 2017, 8:40 AM
c7e66d1f127a	0cf0aefc9dd3	1f18a9cf9f11	Matthias Veit	Incorporate feedback from review.	Apr 19 2017, 6:42 PM
1f18a9cf9f11	8274a21e3b87	6b25031cd8a3	Matthias Veit	Add API documentation for the leader endpoint.	Apr 12 2017, 3:17 PM
6b25031cd8a3	35b53d6bce28	aaa29debbb9e	Matthias Veit	Describe RuntimeConfiguration in raml and use the generated type.	Apr 11 2017, 10:06 AM
aaa29debbb9e	1a6ca266e521	db87e10d68fd	Matthias Veit	Make Backup & Restore operations available via /v2/leader API (Show More…)	Apr 5 2017, 9:18 AM

Diff 2875

docs/docs/rest-api/public/api/api.raml

Show All 9 Lines
10	10	error: v2/types/error.raml
11	11	queue: v2/types/queue.raml
12	12	group: v2/types/group.raml
13	13	strings: v2/types/stringTypes.raml
14	14	pod: v2/types/pod.raml
15	15	podStatus: v2/types/podStatus.raml
16	16	metrics: v2/types/metrics.raml
17	17	logging: v2/types/logging.raml
	18	runtimeConfig: v2/types/runtimeConfiguration.raml
18	19
19	20
20	21	traits: !include traits.raml
21	22
22	23
23	24	#documentation:
24	25	# - title: Using the Marathon REST API
25	26	# content: Make me real!
Show All 21 Lines

View Options

docs/docs/rest-api/public/api/v2/leader.raml

Show All 15 Lines
16	16		application/json:
17	17		example: \|
18	18		{
19	19		"message":"There is no leader"
20	20		}
21	21
22	22
23	23		delete:
24			description: Causes the current leader to abdicate, triggering a new election.
	24		description: \|
	25		Causes the current leader to abdicate, triggering a new election.
	26
	27		Abdicating a Marathon instance causes this instance to exit.
		jdefUnsubmitted Not Done I think we should do better here explaining how `delete` could trigger a re-election. We don't need to duplicate word for word what we have written somewhere else, but this is still too brief in my opinion.
	28		The underlying system is responsible for restarting this instance.
	29
	30		Every abdication triggers a new leader election.
	31		The next elected leader will read the state from the persistent store and continue the work from the previous leader.
	32
	33		It is possible to use this endpoint to trigger a backup or restore operation for the persistent data store.
	34		The requested operation will be perfomed by the next leading Marathon master.
	35		Currently two providers are allowed:
	36		- File provider: file:///path/to/file
	37		- S3 provider (experimental): s3://bucket-name/key-in-bucket?access_key=xxx&secret_key=xxx&region=eu-central-1
	38
	39		queryParameters:
	40		backup?:
	41		required: false
	42		description: URI to backup zk state.
	43		example: file:///var/backup/marathon/backup
	44		restore?:
	45		required: false
	46		description: URI to restore zk state.
	47		example: file:///var/backup/marathon/backup
25	48		responses:
26	49		200:
27	50		description: The abdication message from the current leader.
28	51		body:
29	52		application/json:
30	53		example: \|
31	54		{
32	55		"message": "Leadership abdicated"
Show All 9 Lines

View Options

docs/docs/rest-api/public/api/v2/types/pragma.raml

Show All 27 Lines
28	28		description: \|
29	29		The field is deprecated.
30	30		omitEmpty:
31	31		type: nil
32	32		description: \|
33	33		Applies to field declarations. Do not emit anything for this field if it is empty.
34	34		Should not be used in combination with required fields, or else fields that provide
35	35		default values. Useful with collection or non-built-in field types.
	36		internal:
		jdefUnsubmitted Not Done We need some kind of compile-time separation between public and internal types. (a) I wonder if internal types should actually live in a separate package? Grouping them together in the same package as public API types is confusing and will lead to mistakes down the line. (b) What rules can we implement in the RAML generator to enforce such separation? For example, does it make sense to implement a check that public types cannot refer to internal types? Check that API endpoints don't refer to any internal types? At the very least please file a JIRA to track this concern
	37		type: string
	38		description: \|
	39		Mark a type as internal to indicate, that it is not used in the external API.
	40		No functional change - only to indicate the use of one type.
	41		Use only on type level.
	42
	43

View Options

docs/docs/rest-api/public/api/v2/types/runtimeConfiguration.raml

This file was added.

	1		#%RAML 1.0 Library
	2
	3		uses:
	4		pragma: pragma.raml
	5
	6		types:
	7		RuntimeConfiguration:
	8		type: object
	9		(pragma.internal): Used to store config in zk. Not available in the API.
	10		description: Configuration that is collected during runtime and stored in zk.
	11		properties:
	12		backup?:
	13		type: string
	14		description: URI to backup zk state.
	15		example: file:///var/backup/marathon/backup
	16		restore?:
	17		type: string
	18		description: URI to restore zk state.
	19		example: file:///var/backup/marathon/backup

View Options

src/main/scala/mesosphere/marathon/api/v2/LeaderResource.scala

1	1		package mesosphere.marathon
2	2		package api.v2
3	3
4	4		import javax.servlet.http.HttpServletRequest
5	5		import javax.ws.rs.core.{ Context, Response }
6			import javax.ws.rs.{ DELETE, GET, Path, Produces }
	6		import javax.ws.rs._
7	7
8	8		import com.google.inject.Inject
9	9		import mesosphere.chaos.http.HttpConf
10	10		import mesosphere.marathon.MarathonConf
11	11		import mesosphere.marathon.api.{ AuthResource, MarathonMediaType, RestResource }
12	12		import mesosphere.marathon.core.election.ElectionService
13	13		import mesosphere.marathon.plugin.auth._
	14		import mesosphere.marathon.storage.repository.RuntimeConfigurationRepository
	15		import mesosphere.marathon.raml.RuntimeConfiguration
	16		import Validation._
	17		import mesosphere.marathon.stream.UriIO
14	18
15	19		@Path("v2/leader")
16	20		class LeaderResource @Inject() (
17	21		electionService: ElectionService,
18	22		val config: MarathonConf with HttpConf,
	23		val runtimeConfigRepo: RuntimeConfigurationRepository,
19	24		val authenticator: Authenticator,
20	25		val authorizer: Authorizer)
21	26		extends RestResource with AuthResource {
22	27
23	28		@GET
24	29		@Produces(Array(MarathonMediaType.PREFERRED_APPLICATION_JSON))
25	30		def index(@Context req: HttpServletRequest): Response = authenticated(req) { implicit identity =>
26	31		withAuthorization(ViewResource, AuthorizedResource.Leader) {
27	32		electionService.leaderHostPort match {
28	33		case None => notFound("There is no leader")
29	34		case Some(leader) =>
30	35		ok(jsonObjString("leader" -> leader))
31	36		}
32	37		}
33	38		}
34	39
35	40		@DELETE
36	41		@Produces(Array(MarathonMediaType.PREFERRED_APPLICATION_JSON))
37			def delete(@Context req: HttpServletRequest): Response = authenticated(req) { implicit identity =>
	42		def delete(
	43		@QueryParam("backup") backupNullable: String,
	44		@QueryParam("restore") restoreNullable: String,
	45		@Context req: HttpServletRequest): Response = authenticated(req) { implicit identity =>
38	46		withAuthorization(UpdateResource, AuthorizedResource.Leader) {
39	47		if (electionService.isLeader) {
	48		assumeValid {
	49		val backup = validateOrThrow(Option(backupNullable))(optional(UriIO.valid))
	50		val restore = validateOrThrow(Option(restoreNullable))(optional(UriIO.valid))
	51		result(runtimeConfigRepo.store(RuntimeConfiguration(backup, restore)))
40	52		electionService.abdicateLeadership()
41	53		ok(jsonObjString("message" -> "Leadership abdicated"))
	54		}
		jdefUnsubmitted Done this would probably be more clear (vs nested `withValid` and the validators outside the block, several lines away): assumeValid { validate(block)(optional(UriIO.valid)) validate(restore)(optional(UriIO.valid)) result(...) ... }
42	55		} else {
43	56		notFound("There is no leader")
44	57		}
45	58		}
46	59		}
47	60		}

View Options

src/main/scala/mesosphere/marathon/core/CoreGuiceModule.scala

Show First 20 Lines • Show All 137 Lines • ▼ Show 20 Line(s)
138	138	coreModule.storageModule.taskFailureRepository
139	139
140	140	@Provides
141	141	@Singleton
142	142	def groupRepository(coreModule: CoreModule): GroupRepository =
143	143	coreModule.storageModule.groupRepository
144	144
145	145	@Provides @Singleton
146		def framworkIdRepository(coreModule: CoreModule): FrameworkIdRepository =
	146	def frameworkIdRepository(coreModule: CoreModule): FrameworkIdRepository =
147	147	coreModule.storageModule.frameworkIdRepository
	148
	149	@Provides @Singleton
	150	def RuntimeConfigurationRepository(coreModule: CoreModule): RuntimeConfigurationRepository =
	151	coreModule.storageModule.runtimeConfigurationRepository
148	152
149	153	@Provides @Singleton
150	154	def groupManager(coreModule: CoreModule): GroupManager = coreModule.groupManagerModule.groupManager
151	155
152	156	@Provides @Singleton
153	157	def podSystem(coreModule: CoreModule): PodManager = coreModule.podModule.podManager
154	158
155	159	@Provides @Singleton
▲ Show 20 Lines • Show All 88 Lines • Show Last 20 Lines

View Options

src/main/scala/mesosphere/marathon/core/storage/backup/Backup.scala

Show First 20 Lines • Show All 70 Lines • ▼ Show 20 Line(s)
71	71	*
72	72	* Snapshot backups can be created at all time.
73	73	*
74	74	* There are several command line parameters to define the exact behaviour and location.
75	75	* Please use --help to see all command line parameters
76	76	*/
77	77	object Backup extends BackupRestoreAction {
78	78	def main(args: Array[String]): Unit = {
79		action(new BackupConfig(args.toVector), _.backup())
	79	val config = new BackupConfig(args.toVector)
	80	action(config, _.backup(config.backupLocation()))
80	81	}
81	82	}
82	83
83	84	/**
84	85	* Command line utility to restore a Marathon state from an external storage location.
85	86	*
86	87	* Please note: restoring a backup will overwrite all existing data in the store.
87	88	* All changes that were applied between the creation of this snapshot to the current state will be lost!
88	89	*
89	90	* There are several command line parameters to define the exact behaviour and location.
90	91	* Please use --help to see all command line parameters
91	92	*/
92	93	object Restore extends BackupRestoreAction {
93	94	def main(args: Array[String]): Unit = {
94		action(new BackupConfig(args.toVector), _.restore())
	95	val config = new BackupConfig(args.toVector)
	96	action(config, _.restore(config.backupLocation()))
95	97	}
96	98	}

View Options

src/main/scala/mesosphere/marathon/core/storage/backup/BackupConf.scala

Show All 13 Lines
14	14	*/
15	15	trait BackupConf extends ScallopConf {
16	16
17	17	lazy val backupLocation = opt[String](
18	18	"backup_location",
19	19	descr = "The location uri of the backup to create.",
20	20	validate = loc => Try(new URI(loc)).map(UriIO.isValid).getOrElse(false),
21	21	noshort = true
22		)
	22	).map(new URI(_))
23	23	}

View Options

src/main/scala/mesosphere/marathon/core/storage/backup/PersistentStoreBackup.scala

Show All 13 Lines
14	14
15	15	/**
16	16	* Backup & Restore functionality for configured persistent store and backup location.
17	17	*/
18	18	trait PersistentStoreBackup {
19	19
20	20	/**
21	21	* Backup the state of the configured persistent store.
	22	* @param to the location to write this backup to.
22	23	* @return A future which succeeds, if the backup is written completely.
23	24	*/
24		def backup(): Future[Done]
	25	def backup(to: URI): Future[Done]
25	26
26	27	/**
27	28	* Restore the state from a given backup.
	29	* @param from the location to read this backup from.
28	30	* @return a future which succeeds, if the backup is restored completely.
29	31	*/
30		def restore(): Future[Done]
	32	def restore(from: URI): Future[Done]
31	33
32	34	}
33	35
34	36	object PersistentStoreBackup extends StrictLogging {
35
36		def apply(location: Option[String], store: PersistenceStore[_, _, _])(implicit materializer: Materializer, actorSystem: ActorSystem, ec: ExecutionContext): PersistentStoreBackup = {
37		location.map(new URI(_)).map(new PersistentStoreBackupImpl(store, _)).getOrElse(NoBackup)
38		}
39
40		object NoBackup extends PersistentStoreBackup {
41		override def backup() = {
42		logger.warn("No backup location defined. Please use --backup_location.")
43		Future.successful(Done)
44		}
45		override def restore() = {
46		logger.warn("No backup location defined. Please use --backup_location.")
47		Future.successful(Done)
48		}
	37	def apply(store: PersistenceStore[_, _, _])(implicit materializer: Materializer, actorSystem: ActorSystem, ec: ExecutionContext): PersistentStoreBackup = {
	38	new PersistentStoreBackupImpl(store)
49	39	}
50	40	}

View Options

src/main/scala/mesosphere/marathon/core/storage/backup/impl/PersistentStoreBackupImpl.scala

1	1	package mesosphere.marathon
2	2	package core.storage.backup.impl
3	3
4	4	import java.net.URI
5	5
6	6	import akka.Done
7	7	import akka.actor.ActorSystem
8	8	import akka.stream.Materializer
9	9	import akka.stream.scaladsl.Flow
10	10	import com.typesafe.scalalogging.StrictLogging
11	11	import mesosphere.marathon.core.storage.backup.{ BackupItem, PersistentStoreBackup }
12	12	import mesosphere.marathon.core.storage.store.PersistenceStore
13	13	import mesosphere.marathon.stream.UriIO
14	14
15	15	import scala.concurrent.{ ExecutionContext, Future }
16	16
17		class PersistentStoreBackupImpl(store: PersistenceStore[_, _, _], location: URI)(implicit materializer: Materializer, actorSystem: ActorSystem, ec: ExecutionContext)
	17	class PersistentStoreBackupImpl(store: PersistenceStore[_, _, _])(implicit materializer: Materializer, actorSystem: ActorSystem, ec: ExecutionContext)
18	18	extends PersistentStoreBackup with StrictLogging {
19	19
20		override def backup(): Future[Done] = {
21		logger.info(s"Create backup at $location")
	20	override def backup(to: URI): Future[Done] = {
	21	logger.info(s"Create backup at $to")
22	22	store.backup()
23	23	.via(logFlow("Backup"))
24	24	.via(TarBackupFlow.tar)
25		.runWith(UriIO.writer(location))
	25	.runWith(UriIO.writer(to))
26	26	}
27	27
28		override def restore(): Future[Done] = {
29		logger.info(s"Restore backup from $location")
30		UriIO.reader(location)
	28	override def restore(from: URI): Future[Done] = {
	29	logger.info(s"Restore backup from $from")
	30	UriIO.reader(from)
31	31	.via(TarBackupFlow.untar)
32	32	.via(logFlow("Restore"))
33	33	.runWith(store.restore())
34	34	}
35	35
36	36	private[this] def logFlow(message: String) = Flow.fromFunction[BackupItem, BackupItem] { item =>
37	37	logger.info(s"$message item category:${item.category} key:${item.key} version:${item.version}")
38	38	item
39	39	}
40	40	}
41	41

View Options

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

1	1	package mesosphere.marathon
2	2	package storage
3	3
	4	import java.net.URI
4	5	import java.util
5	6	import java.util.Collections
6	7
7	8	import akka.actor.{ ActorSystem, Scheduler }
8	9	import akka.stream.Materializer
9	10	import com.typesafe.config.Config
10	11	import mesosphere.marathon.core.base.LifecycleState
11	12	import mesosphere.marathon.core.storage.store.PersistenceStore
12	13	import mesosphere.marathon.core.storage.store.impl.BasePersistenceStore
13	14	import mesosphere.marathon.core.storage.store.impl.cache.{ LazyCachingPersistenceStore, LazyVersionCachingPersistentStore, LoadTimeCachingPersistenceStore }
14	15	import mesosphere.marathon.core.storage.store.impl.memory.{ Identity, InMemoryPersistenceStore, RamId }
15	16	import mesosphere.marathon.core.storage.store.impl.zk.{ NoRetryPolicy, RichCuratorFramework, ZkId, ZkPersistenceStore, ZkSerialized }
16	17	import mesosphere.marathon.util.{ RetryConfig, toRichConfig }
17	18	import org.apache.curator.framework.api.ACLProvider
18	19	import org.apache.curator.framework.imps.GzipCompressionProvider
19	20	import org.apache.curator.framework.{ AuthInfo, CuratorFrameworkFactory }
20	21	import org.apache.zookeeper.ZooDefs
21	22	import org.apache.zookeeper.data.ACL
22	23
23	24	import scala.concurrent.ExecutionContext
24	25	import scala.concurrent.duration._
25	26
26		sealed trait StorageConfig extends Product with Serializable
	27	sealed trait StorageConfig extends Product with Serializable {
	28	def backupLocation: Option[URI]
	29	}
27	30
28	31	sealed trait CacheType
29	32	case object NoCaching extends CacheType
30	33	case object EagerCaching extends CacheType
31	34	case object LazyCaching extends CacheType
32	35
33	36	object CacheType {
34	37	def apply(str: String): CacheType = str.toLowerCase match {
▲ Show 20 Lines • Show All 62 Lines • ▼ Show 20 Line(s)
97	100	password: Option[String],
98	101	enableCompression: Boolean,
99	102	retryConfig: RetryConfig,
100	103	maxConcurrent: Int,
101	104	maxOutstanding: Int,
102	105	maxVersions: Int,
103	106	versionCacheConfig: Option[VersionCacheConfig],
104	107	availableFeatures: Set[String],
105		backupLocation: Option[String],
106	108	lifecycleState: LifecycleState,
107		defaultNetworkName: Option[String]
	109	defaultNetworkName: Option[String],
	110	backupLocation: Option[URI]
108	111	) extends PersistenceStorageConfig[ZkId, String, ZkSerialized] {
109	112
110	113	lazy val client: RichCuratorFramework = {
111	114	val builder = CuratorFrameworkFactory.builder()
112	115	builder.connectString(zkHosts)
113	116	sessionTimeout.foreach(t => builder.sessionTimeoutMs(t.toMillis.toInt))
114	117	connectionTimeout.foreach(t => builder.connectionTimeoutMs(t.toMillis.toInt))
115	118	if (enableCompression) builder.compressionProvider(new GzipCompressionProvider)
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Line(s)
187	190	enableCompression = config.bool("enable-compression", true),
188	191	retryConfig = RetryConfig(config),
189	192	maxConcurrent = config.int("max-concurrent-requests", 32),
190	193	maxOutstanding = config.int("max-concurrent-outstanding", Int.MaxValue),
191	194	maxVersions = config.int("max-versions", StorageConfig.DefaultMaxVersions),
192	195	versionCacheConfig =
193	196	if (config.bool("version-cache-enabled", true)) StorageConfig.DefaultVersionCacheConfig else None,
194	197	availableFeatures = config.stringList("available-features", Seq.empty).to[Set],
195		backupLocation = config.optionalString("backup-location"),
196	198	lifecycleState = lifecycleState,
197		defaultNetworkName = config.optionalString("default-network-name")
	199	defaultNetworkName = config.optionalString("default-network-name"),
	200	backupLocation = config.optionalString("backup-location").map(new URI(_))
198	201	)
199	202	}
200	203	}
201	204
202	205	case class InMem(
203	206	maxVersions: Int,
204	207	availableFeatures: Set[String],
205		backupLocation: Option[String],
206		defaultNetworkName: Option[String]
	208	defaultNetworkName: Option[String],
	209	backupLocation: Option[URI]
207	210	) extends PersistenceStorageConfig[RamId, String, Identity] {
208	211	override val cacheType: CacheType = NoCaching
209	212	override val versionCacheConfig: Option[VersionCacheConfig] = None
210	213
211	214	protected def leafStore(implicit mat: Materializer, ctx: ExecutionContext,
212	215	scheduler: Scheduler, actorSystem: ActorSystem): BasePersistenceStore[RamId, String, Identity] =
213	216	new InMemoryPersistenceStore()
214	217	}
215	218
216	219	object InMem {
217	220	val StoreName = "mem"
218	221
219	222	def apply(conf: StorageConf): InMem =
220		InMem(conf.maxVersions(), conf.availableFeatures, conf.backupLocation.get, conf.defaultNetworkName.get)
	223	InMem(conf.maxVersions(), conf.availableFeatures, conf.defaultNetworkName.get, conf.backupLocation.get)
221	224
222	225	def apply(conf: Config): InMem =
223	226	InMem(
224	227	conf.int("max-versions", StorageConfig.DefaultMaxVersions),
225	228	availableFeatures = conf.stringList("available-features", Seq.empty).to[Set],
226		backupLocation = conf.optionalString("backup-location"),
227		defaultNetworkName = conf.optionalString("default-network-name")
	229	defaultNetworkName = conf.optionalString("default-network-name"),
	230	backupLocation = conf.optionalString("backup-location").map(new URI(_))
228	231	)
229	232	}
230	233
231	234	object StorageConfig {
232	235	val DefaultVersionCacheConfig = Option(VersionCacheConfig.Default)
233	236
234	237	val DefaultLegacyMaxVersions = 25
235	238	val DefaultMaxVersions = 5000
Show All 14 Lines

View Options

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

Show All 16 Lines
17	17	*/
18	18	trait StorageModule {
19	19	val instanceRepository: InstanceRepository
20	20	val deploymentRepository: DeploymentRepository
21	21	val taskFailureRepository: TaskFailureRepository
22	22	val groupRepository: GroupRepository
23	23	val frameworkIdRepository: FrameworkIdRepository
24	24	val eventSubscribersRepository: EventSubscribersRepository
	25	val runtimeConfigurationRepository: RuntimeConfigurationRepository
25	26	val migration: Migration
26	27	val leadershipInitializers: Seq[PrePostDriverCallback]
27	28	val persistentStoreBackup: PersistentStoreBackup
28	29	}
29	30
30	31	object StorageModule {
31	32	def apply(conf: StorageConf with NetworkConf, lifecycleState: LifecycleState)(implicit mat: Materializer, ctx: ExecutionContext,
32	33	scheduler: Scheduler, actorSystem: ActorSystem): StorageModule = {
Show All 14 Lines
47	48
48	49	val taskRepository = TaskRepository.zkRepository(store)
49	50	val instanceRepository = InstanceRepository.zkRepository(store)
50	51	val deploymentRepository = DeploymentRepository.zkRepository(store, groupRepository,
51	52	appRepository, podRepository, zk.maxVersions)
52	53	val taskFailureRepository = TaskFailureRepository.zkRepository(store)
53	54	val frameworkIdRepository = FrameworkIdRepository.zkRepository(store)
54	55	val eventSubscribersRepository = EventSubscribersRepository.zkRepository(store)
	56	val runtimeConfigurationRepository = RuntimeConfigurationRepository.zkRepository(store)
55	57
56	58	val leadershipInitializers = store match {
57	59	case s: LoadTimeCachingPersistenceStore[_, _, _] =>
58	60	Seq(s)
59	61	case _ =>
60	62	Nil
61	63	}
62	64
63		val backup = PersistentStoreBackup(zk.backupLocation, store)
	65	val backup = PersistentStoreBackup(store)
64	66	val migration = new Migration(zk.availableFeatures, zk.defaultNetworkName, mesosBridgeName, store,
65	67	appRepository, groupRepository, deploymentRepository, taskRepository, instanceRepository,
66	68	taskFailureRepository, frameworkIdRepository, eventSubscribersRepository,
67		ServiceDefinitionRepository.zkRepository(store), backup)
	69	ServiceDefinitionRepository.zkRepository(store), runtimeConfigurationRepository, backup, config)
68	70
69	71	StorageModuleImpl(
70	72	instanceRepository,
71	73	deploymentRepository,
72	74	taskFailureRepository,
73	75	groupRepository,
74	76	frameworkIdRepository,
75	77	eventSubscribersRepository,
	78	runtimeConfigurationRepository,
76	79	migration,
77	80	leadershipInitializers,
78	81	backup
79	82	)
80	83	case mem: InMem =>
81	84	val store = mem.store
82	85	val appRepository = AppRepository.inMemRepository(store)
83	86	val podRepository = PodRepository.inMemRepository(store)
84	87	val taskRepository = TaskRepository.inMemRepository(store)
85	88	val instanceRepository = InstanceRepository.inMemRepository(store)
86	89	val groupRepository = GroupRepository.inMemRepository(store, appRepository, podRepository)
87	90	val deploymentRepository = DeploymentRepository.inMemRepository(store, groupRepository,
88	91	appRepository, podRepository, mem.maxVersions)
89	92	val taskFailureRepository = TaskFailureRepository.inMemRepository(store)
90	93	val frameworkIdRepository = FrameworkIdRepository.inMemRepository(store)
91	94	val eventSubscribersRepository = EventSubscribersRepository.inMemRepository(store)
	95	val runtimeConfigurationRepository = RuntimeConfigurationRepository.inMemRepository(store)
92	96
93	97	val leadershipInitializers = store match {
94	98	case s: LoadTimeCachingPersistenceStore[_, _, _] =>
95	99	Seq(s)
96	100	case _ =>
97	101	Nil
98	102	}
99	103
100		val backup = PersistentStoreBackup(mem.backupLocation, store)
	104	val backup = PersistentStoreBackup(store)
101	105	val migration = new Migration(mem.availableFeatures, mem.defaultNetworkName, mesosBridgeName,
102	106	store, appRepository, groupRepository, deploymentRepository, taskRepository, instanceRepository,
103	107	taskFailureRepository, frameworkIdRepository, eventSubscribersRepository,
104		ServiceDefinitionRepository.inMemRepository(store), backup)
	108	ServiceDefinitionRepository.inMemRepository(store), runtimeConfigurationRepository, backup, config)
105	109
106	110	StorageModuleImpl(
107	111	instanceRepository,
108	112	deploymentRepository,
109	113	taskFailureRepository,
110	114	groupRepository,
111	115	frameworkIdRepository,
112	116	eventSubscribersRepository,
	117	runtimeConfigurationRepository,
113	118	migration,
114	119	leadershipInitializers,
115	120	backup
116	121	)
117	122	}
118	123	}
119	124	}
120	125
121	126	private[storage] case class StorageModuleImpl(
122	127	instanceRepository: InstanceRepository,
123	128	deploymentRepository: DeploymentRepository,
124	129	taskFailureRepository: TaskFailureRepository,
125	130	groupRepository: GroupRepository,
126	131	frameworkIdRepository: FrameworkIdRepository,
127	132	eventSubscribersRepository: EventSubscribersRepository,
	133	runtimeConfigurationRepository: RuntimeConfigurationRepository,
128	134	migration: Migration,
129	135	leadershipInitializers: Seq[PrePostDriverCallback],
130	136	persistentStoreBackup: PersistentStoreBackup
131	137	) extends StorageModule

View Options

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

1	1	package mesosphere.marathon
2	2	package storage.migration
3	3
	4	import java.net.URI
	5
4	6	import akka.Done
5	7	import akka.stream.Materializer
6	8	import com.typesafe.scalalogging.StrictLogging
7	9	import mesosphere.marathon.Protos.StorageVersion
8	10	import mesosphere.marathon.core.async.ExecutionContexts.global
9	11	import mesosphere.marathon.core.storage.backup.PersistentStoreBackup
10	12	import mesosphere.marathon.core.storage.store.PersistenceStore
11	13	import mesosphere.marathon.storage.migration.legacy.MigrationTo_1_4_2
	14	import mesosphere.marathon.storage.StorageConfig
12	15	import mesosphere.marathon.storage.repository._
13	16
14	17	import scala.async.Async.{ async, await }
15	18	import scala.concurrent.duration._
16	19	import scala.concurrent.{ Await, Future }
17	20	import scala.util.control.NonFatal
18	21	import scala.util.matching.Regex
	22	import mesosphere.marathon.raml.RuntimeConfiguration
19	23
20	24	/**
21	25	* @param persistenceStore Optional "new" PersistenceStore for new migrations, the repositories
22	26	* are assumed to be in the new format.
23	27	*/
24	28	@SuppressWarnings(Array("UnusedMethodParameter")) // materializer will definitely be used in the future.
25	29	class Migration(
26	30	private[migration] val availableFeatures: Set[String],
27	31	private[migration] val defaultNetworkName: Option[String],
28	32	private[migration] val mesosBridgeName: String,
29	33	private[migration] val persistenceStore: PersistenceStore[_, _, _],
30	34	private[migration] val appRepository: AppRepository,
31	35	private[migration] val groupRepository: GroupRepository,
32	36	private[migration] val deploymentRepository: DeploymentRepository,
33	37	private[migration] val taskRepo: TaskRepository,
34	38	private[migration] val instanceRepo: InstanceRepository,
35	39	private[migration] val taskFailureRepo: TaskFailureRepository,
36	40	private[migration] val frameworkIdRepo: FrameworkIdRepository,
37	41	private[migration] val eventSubscribersRepo: EventSubscribersRepository,
38	42	private[migration] val serviceDefinitionRepo: ServiceDefinitionRepository,
39		private[migration] val backup: PersistentStoreBackup)(implicit mat: Materializer) extends StrictLogging {
	43	private[migration] val runtimeConfigurationRepository: RuntimeConfigurationRepository,
	44	private[migration] val backup: PersistentStoreBackup,
	45	private[migration] val config: StorageConfig
	46	)(implicit mat: Materializer) extends StrictLogging {
40	47
41	48	import StorageVersions._
42	49
43	50	type MigrationAction = (StorageVersion, () => Future[Any])
44	51
45	52	private[migration] val minSupportedStorageVersion = StorageVersions(1, 4, 0, StorageVersion.StorageFormat.PERSISTENCE_STORE)
46	53
47	54	/**
Show All 22 Lines
70	77	}.map(_ => res :+ migrateVersion)
71	78	}
72	79	}
73	80	}
74	81
75	82	@SuppressWarnings(Array("all")) // async/await
76	83	def migrate(): Seq[StorageVersion] = {
77	84	val result = async {
78		val currentVersion = await(getCurrentVersion)
	85	val config = await(runtimeConfigurationRepository.get()).getOrElse(RuntimeConfiguration())
	86	// before backup/restore called, reset the runtime configuration
	87	await(runtimeConfigurationRepository.store(RuntimeConfiguration(None, None)))
	88	// step 1: backup current zk state
	89	await(config.backup.map(uri => backup.backup(new URI(uri))).getOrElse(Future.successful(Done)))
	90	// step 2: restore state from given backup
	91	await(config.restore.map(uri => backup.restore(new URI(uri))).getOrElse(Future.successful(Done)))
	92	// last step run the migration, to ensure we can operate on the zk state
	93	await(migrateStorage(backupCreated = config.backup.isDefined \|\| config.restore.isDefined))
	94	}
	95	val migrations = Await.result(result, Duration.Inf)
	96	logger.info(s"Migration successfully applied for version ${StorageVersions.current.str}")
	97	migrations
	98	}
79	99
	100	@SuppressWarnings(Array("all")) // async/await
	101	def migrateStorage(backupCreated: Boolean = false): Future[Seq[StorageVersion]] = {
	102	async {
	103	val currentVersion = await(getCurrentVersion)
80	104	val currentBuildVersion = StorageVersions.current
81	105
82	106	val migrations = currentVersion match {
83	107	case Some(version) if version < minSupportedStorageVersion =>
84	108	val msg = s"Migration from versions < ${minSupportedStorageVersion.str} are not supported. " +
85	109	s"Your version: ${version.str}"
86	110	throw new MigrationFailedException(msg)
87	111	case Some(version) if version > currentBuildVersion =>
88	112	val msg = s"Migration from ${version.str} is not supported as it is newer" +
89	113	s" than ${StorageVersions.current.str}."
90	114	throw new MigrationFailedException(msg)
91	115	case Some(version) if version < currentBuildVersion =>
	116	if (!backupCreated && config.backupLocation.isDefined) {
92	117	logger.info("Backup current state")
93		await(backup.backup())
	118	await(backup.backup(config.backupLocation.get))
94	119	logger.info("Backup finished. Apply migration.")
	120	}
95	121	val result = await(applyMigrationSteps(version))
96	122	await(storeCurrentVersion())
97	123	result
98	124	case Some(version) if version == currentBuildVersion =>
99	125	logger.info("No migration necessary, already at the current version")
100	126	Nil
101	127	case _ =>
102	128	logger.info("No migration necessary, no version stored")
103	129	await(storeCurrentVersion())
104	130	Nil
105	131	}
106	132	migrations
107	133	}.recover {
108	134	case ex: MigrationFailedException => throw ex
109	135	case NonFatal(ex) => throw new MigrationFailedException(s"Migration Failed: ${ex.getMessage} ${ex.getStackTrace.mkString}", ex)
110	136	}
111
112		val migrations = Await.result(result, Duration.Inf)
113		logger.info(s"Migration successfully applied for version ${StorageVersions.current.str}")
114		migrations
115	137	}
116	138
117	139	private def getCurrentVersion: Future[Option[StorageVersion]] =
118	140	persistenceStore.storageVersion()
119	141
120	142	private def storeCurrentVersion(): Future[Done] =
121	143	persistenceStore.setStorageVersion(StorageVersions.current)
122	144	}
▲ Show 20 Lines • Show All 54 Lines • Show Last 20 Lines

View Options

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

Show All 15 Lines
16	16	import mesosphere.marathon.core.storage.repository.impl.{ PersistenceStoreRepository, PersistenceStoreVersionedRepository }
17	17	import mesosphere.marathon.core.storage.store.impl.memory.{ Identity, RamId }
18	18	import mesosphere.marathon.core.storage.store.impl.zk.{ ZkId, ZkSerialized }
19	19	import mesosphere.marathon.core.storage.store.{ IdResolver, PersistenceStore }
20	20	import mesosphere.marathon.core.task.Task
21	21	import mesosphere.marathon.state._
22	22	import mesosphere.marathon.core.deployment.DeploymentPlan
23	23	import mesosphere.util.state.FrameworkId
	24	import mesosphere.marathon.raml.RuntimeConfiguration
24	25
25	26	import scala.async.Async.{ async, await }
26	27	import scala.collection.immutable.Seq
27	28	import scala.concurrent.{ ExecutionContext, Future }
28	29
29	30	trait GroupRepository {
30	31	/** Fetch the root, returns an empty root if the root doesn't yet exist */
31	32	def root(): Future[RootGroup]
▲ Show 20 Lines • Show All 172 Lines • ▼ Show 20 Line(s)
204	205	}
205	206
206	207	def inMemRepository(persistenceStore: PersistenceStore[RamId, String, Identity]): FrameworkIdRepository = {
207	208	import mesosphere.marathon.storage.store.InMemoryStoreSerialization._
208	209	new FrameworkIdRepositoryImpl(persistenceStore)
209	210	}
210	211	}
211	212
	213	trait RuntimeConfigurationRepository extends SingletonRepository[RuntimeConfiguration]
	214
	215	object RuntimeConfigurationRepository {
	216	def zkRepository(persistenceStore: PersistenceStore[ZkId, String, ZkSerialized]): RuntimeConfigurationRepository = {
	217	import mesosphere.marathon.storage.store.ZkStoreSerialization._
	218	new RuntimeConfigurationRepositoryImpl(persistenceStore)
	219	}
	220
	221	def inMemRepository(persistenceStore: PersistenceStore[RamId, String, Identity]): RuntimeConfigurationRepository = {
	222	import mesosphere.marathon.storage.store.InMemoryStoreSerialization._
	223	new RuntimeConfigurationRepositoryImpl(persistenceStore)
	224	}
	225	}
	226
212	227	trait EventSubscribersRepository extends SingletonRepository[EventSubscribers]
213	228
214	229	object EventSubscribersRepository {
215	230
216	231	def zkRepository(persistenceStore: PersistenceStore[ZkId, String, ZkSerialized]): EventSubscribersRepository = {
217	232	import mesosphere.marathon.storage.store.ZkStoreSerialization._
218	233	new EventSubscribersRepositoryImpl(persistenceStore)
219	234	}
▲ Show 20 Lines • Show All 108 Lines • ▼ Show 20 Line(s)
328	343	ir: IdResolver[String, FrameworkId, C, K],
329	344	marshaller: Marshaller[FrameworkId, S],
330	345	unmarshaller: Unmarshaller[S, FrameworkId]
331	346	) extends FrameworkIdRepository {
332	347	private val ID = "id"
333	348	private val repo = new PersistenceStoreRepository[String, FrameworkId, K, C, S](persistenceStore, _ => ID)
334	349	override def get(): Future[Option[FrameworkId]] = repo.get(ID)
335	350	override def store(v: FrameworkId): Future[Done] = repo.store(v)
	351	override def delete(): Future[Done] = repo.delete(ID)
	352	}
	353
	354	class RuntimeConfigurationRepositoryImpl[K, C, S](persistenceStore: PersistenceStore[K, C, S])(
	355	implicit
	356	ir: IdResolver[String, RuntimeConfiguration, C, K],
	357	marshaller: Marshaller[RuntimeConfiguration, S],
	358	unmarshaller: Unmarshaller[S, RuntimeConfiguration]
	359	) extends RuntimeConfigurationRepository {
	360	private val ID = "id"
	361	private val repo = new PersistenceStoreRepository[String, RuntimeConfiguration, K, C, S](persistenceStore, _ => ID)
	362	override def get(): Future[Option[RuntimeConfiguration]] = repo.get(ID)
	363	override def store(v: RuntimeConfiguration): Future[Done] = repo.store(v)
336	364	override def delete(): Future[Done] = repo.delete(ID)
337	365	}
338	366
339	367	class EventSubscribersRepositoryImpl[K, C, S](persistenceStore: PersistenceStore[K, C, S])(
340	368	implicit
341	369	ir: IdResolver[String, EventSubscribers, C, K],
342	370	marshaller: Marshaller[EventSubscribers, S],
343	371	unmarshaller: Unmarshaller[S, EventSubscribers]
344	372	) extends EventSubscribersRepository {
345	373	private val ID = "id"
346	374	private val repo = new PersistenceStoreRepository[String, EventSubscribers, K, C, S](persistenceStore, _ => ID)
347	375	override def get(): Future[Option[EventSubscribers]] = repo.get(ID)
348	376	override def store(v: EventSubscribers): Future[Done] = repo.store(v)
349	377	override def delete(): Future[Done] = repo.delete(ID)
350	378	}

View Options

src/main/scala/mesosphere/marathon/storage/store/InMemoryStoreSerialization.scala

Show All 9 Lines
10	10	import mesosphere.marathon.core.instance.Instance.Id
11	11	import mesosphere.marathon.core.pod.PodDefinition
12	12	import mesosphere.marathon.core.storage.store.IdResolver
13	13	import mesosphere.marathon.core.storage.store.impl.memory.{ Identity, RamId }
14	14	import mesosphere.marathon.core.task.Task
15	15	import mesosphere.marathon.state.{ AppDefinition, PathId, TaskFailure }
16	16	import mesosphere.marathon.storage.repository.{ StoredGroup, StoredPlan }
17	17	import mesosphere.util.state.FrameworkId
	18	import mesosphere.marathon.raml.RuntimeConfiguration
18	19
19	20	trait InMemoryStoreSerialization {
20	21	implicit def marshaller[V]: Marshaller[V, Identity] = Marshaller.opaque { a: V => Identity(a) }
21	22
22	23	@SuppressWarnings(Array("AsInstanceOf"))
23	24	implicit def unmarshaller[V]: Unmarshaller[Identity, V] =
24	25	Unmarshaller.strict { a: Identity => a.value.asInstanceOf[V] }
25	26
▲ Show 20 Lines • Show All 54 Lines • ▼ Show 20 Line(s)
80	81
81	82	implicit val frameworkIdResolver = new IdResolver[String, FrameworkId, String, RamId] {
82	83	override def toStorageId(id: String, version: Option[OffsetDateTime]): RamId =
83	84	RamId(category, id, version)
84	85	override val category: String = "framework-id"
85	86	override def fromStorageId(key: RamId): String = key.id
86	87	override val hasVersions = false
87	88	override def version(v: FrameworkId): OffsetDateTime = OffsetDateTime.MIN
	89	}
	90
	91	implicit val runtimeConfigurationResolver = new IdResolver[String, RuntimeConfiguration, String, RamId] {
	92	override def toStorageId(id: String, version: Option[OffsetDateTime]): RamId =
	93	RamId(category, id, version)
	94	override val category: String = "runtime-configuration"
	95	override def fromStorageId(key: RamId): String = key.id
	96	override val hasVersions = false
	97	override def version(v: RuntimeConfiguration): OffsetDateTime = OffsetDateTime.MIN
88	98	}
89	99
90	100	implicit val eventSubscribersResolver = new IdResolver[String, EventSubscribers, String, RamId] {
91	101	override def toStorageId(id: String, version: Option[OffsetDateTime]): RamId =
92	102	RamId(category, id, version)
93	103	override val category: String = "event-subscribers"
94	104	override def fromStorageId(key: RamId): String = key.id
95	105	override val hasVersions = true
96	106	override def version(v: EventSubscribers): OffsetDateTime = OffsetDateTime.MIN
97	107	}
98	108	}
99	109
100	110	object InMemoryStoreSerialization extends InMemoryStoreSerialization

View Options

src/main/scala/mesosphere/marathon/storage/store/ZkStoreSerialization.scala

1	1		package mesosphere.marathon
2	2		package storage.store
3	3
	4		import java.nio.charset.StandardCharsets
4	5		import java.time.OffsetDateTime
5	6
6	7		import akka.http.scaladsl.marshalling.Marshaller
7	8		import akka.http.scaladsl.unmarshalling.Unmarshaller
8	9		import akka.util.ByteString
9	10		import mesosphere.marathon.Protos.{ DeploymentPlanDefinition, MarathonTask, ServiceDefinition }
10	11		import mesosphere.marathon.core.event.EventSubscribers
11	12		import mesosphere.marathon.core.instance.Instance
12	13		import mesosphere.marathon.core.instance.Instance.Id
13	14		import mesosphere.marathon.core.pod.PodDefinition
14	15		import mesosphere.marathon.core.storage.store.IdResolver
15	16		import mesosphere.marathon.core.storage.store.impl.zk.{ ZkId, ZkSerialized }
16	17		import mesosphere.marathon.core.task.Task
17	18		import mesosphere.marathon.core.task.tracker.impl.TaskSerializer
18	19		import mesosphere.marathon.raml.{ Pod, Raml }
19	20		import mesosphere.marathon.state.{ AppDefinition, PathId, TaskFailure }
20	21		import mesosphere.marathon.storage.repository.{ StoredGroup, StoredGroupRepositoryImpl, StoredPlan }
21	22		import mesosphere.util.state.FrameworkId
	23		import mesosphere.marathon.raml.RuntimeConfiguration
22	24		import play.api.libs.json.Json
23	25
24	26		trait ZkStoreSerialization {
25	27		/** General id resolver for a key of Path.Id */
26	28		class ZkPathIdResolver[T](
27	29		val category: String,
28	30		val hasVersions: Boolean,
29	31		getVersion: (T) => OffsetDateTime)
Show All 17 Lines
47	49		AppDefinition.fromProto(proto)
48	50		}
49	51
50	52		implicit val podDefResolver: IdResolver[PathId, PodDefinition, String, ZkId] =
51	53		new ZkPathIdResolver[PodDefinition]("pods", true, _.version.toOffsetDateTime)
52	54
53	55		implicit val podDefMarshaller: Marshaller[PodDefinition, ZkSerialized] =
54	56		Marshaller.opaque { podDef =>
55			ZkSerialized(ByteString(Json.stringify(Json.toJson(Raml.toRaml(podDef))), "UTF-8"))
	57		ZkSerialized(ByteString(Json.stringify(Json.toJson(Raml.toRaml(podDef))), StandardCharsets.UTF_8.name()))
56	58		}
57	59
58	60		implicit val podDefUnmarshaller: Unmarshaller[ZkSerialized, PodDefinition] =
59	61		Unmarshaller.strict {
60	62		case ZkSerialized(byteString) =>
61	63		Raml.fromRaml(Json.parse(byteString.utf8String).as[Pod])
62	64		}
63	65
64	66		implicit val instanceResolver: IdResolver[Instance.Id, Instance, String, ZkId] =
65	67		new IdResolver[Instance.Id, Instance, String, ZkId] {
66	68		override def toStorageId(id: Id, version: Option[OffsetDateTime]): ZkId =
67	69		ZkId(category, id.idString, version)
68	70		override val category: String = "instance"
69	71		override def fromStorageId(key: ZkId): Id = Instance.Id(key.id)
70	72		override val hasVersions: Boolean = false
71	73		override def version(v: Instance): OffsetDateTime = OffsetDateTime.MIN
72	74		}
73	75
74	76		implicit val instanceMarshaller: Marshaller[Instance, ZkSerialized] =
75	77		Marshaller.opaque { instance =>
76			ZkSerialized(ByteString(Json.stringify(Json.toJson(instance)), "UTF-8"))
	78		ZkSerialized(ByteString(Json.stringify(Json.toJson(instance)), StandardCharsets.UTF_8.name()))
77	79		}
78	80
79	81		implicit val instanceUnmarshaller: Unmarshaller[ZkSerialized, Instance] =
80	82		Unmarshaller.strict {
81	83		case ZkSerialized(byteString) =>
82	84		Json.parse(byteString.utf8String).as[Instance]
83	85		}
84	86
▲ Show 20 Lines • Show All 83 Lines • ▼ Show 20 Line(s)
168	170
169	171		implicit val frameworkIdMarshaller: Marshaller[FrameworkId, ZkSerialized] =
170	172		Marshaller.opaque(id => ZkSerialized(ByteString(id.toProtoByteArray)))
171	173
172	174		implicit val frameworkIdUnmarshaller: Unmarshaller[ZkSerialized, FrameworkId] =
173	175		Unmarshaller.strict {
174	176		case ZkSerialized(byteString) =>
175	177		FrameworkId.fromProtoBytes(byteString.toArray)
	178		}
	179
	180		implicit val runtimeConfigurationResolver = new IdResolver[String, RuntimeConfiguration, String, ZkId] {
	181		override def toStorageId(id: String, version: Option[OffsetDateTime]): ZkId =
	182		ZkId(category, id, version)
	183		override val category: String = "runtime-configuration"
	184		override def fromStorageId(key: ZkId): String = key.id
	185		override val hasVersions = false
	186		override def version(v: RuntimeConfiguration): OffsetDateTime = OffsetDateTime.MIN
	187		}
	188
	189		implicit val runtimeConfigurationMarshaller: Marshaller[RuntimeConfiguration, ZkSerialized] =
	190		Marshaller.opaque(config => ZkSerialized(ByteString(Json.stringify(Json.toJson(config)), StandardCharsets.UTF_8.name())))
		jasongilanfarrUnsubmitted Done minor, but there are a bunch of constants for UTF-8, e.g. StandardCharsets.UTF8
	191
	192		implicit val runtimeConfigurationUnmarshaller: Unmarshaller[ZkSerialized, RuntimeConfiguration] =
	193		Unmarshaller.strict {
	194		case ZkSerialized(byteString) => Json.parse(byteString.utf8String).as[RuntimeConfiguration]
176	195		}
177	196
178	197		implicit val eventSubscribersResolver = new IdResolver[String, EventSubscribers, String, ZkId] {
179	198		override def toStorageId(id: String, version: Option[OffsetDateTime]): ZkId =
180	199		ZkId(id, category, version)
181	200		override val category: String = "event-subscribers"
182	201		override def fromStorageId(key: ZkId): String = key.id
183	202		override val hasVersions = false
Show All 14 Lines

View Options

src/main/scala/mesosphere/marathon/stream/UriIO.scala

Show All 9 Lines
10	10	import akka.stream.alpakka.s3.acl.CannedAcl
11	11	import akka.stream.alpakka.s3.auth.AWSCredentials
12	12	import akka.stream.alpakka.s3.impl.MetaHeaders
13	13	import akka.stream.alpakka.s3.scaladsl.S3Client
14	14	import akka.stream.scaladsl.{ FileIO, Source, Sink => ScalaSink }
15	15	import akka.util.ByteString
16	16	import akka.{ Done, NotUsed }
17	17	import com.typesafe.scalalogging.StrictLogging
	18	import com.wix.accord.Validator
	19	import com.wix.accord.dsl._
	20	import mesosphere.marathon.api.v2.Validation.{ isTrue, uriIsValid }
18	21
19	22	import scala.concurrent.{ ExecutionContext, Future }
20	23
21	24	/**
22	25	* UriIO provides sources from and sinks to an URI.
23	26	* It supports different providers, so multiple schemes are supported.
24	27	*
25	28	* Provided schemes:
▲ Show 20 Lines • Show All 63 Lines • ▼ Show 20 Line(s)
89	92	def isValid(uri: URI): Boolean = {
90	93	def nonEmpty(nullable: String): Boolean = nullable != null && nullable.nonEmpty
91	94	uri.getScheme match {
92	95	case "file" if nonEmpty(uri.getPath) && uri.getPath.length > 1 => true
93	96	case "s3" if nonEmpty(uri.getHost) && nonEmpty(uri.getPath) => true
94	97	case _ => false
95	98	}
96	99	}
	100
	101	def valid: Validator[String] = uriIsValid and isTrue[String]{ uri: String => s"Invalid URI or unsupported scheme: $uri" }(uri => isValid(new URI(uri)))
97	102
98	103	private[this] def s3Client(uri: URI)(implicit actorSystem: ActorSystem, materializer: Materializer): S3Client = {
99	104	val params = parseParams(uri)
100	105	val region = params.getOrElse("region", "us-east-1")
101	106	val accessKey = params.getOrElse("access_key", "")
102	107	val accessSecret = params.getOrElse("secret_key", "")
103	108	logger.info(s"S3 settings region: $region accessKey:$accessKey accessSecret:$accessSecret")
104	109	val credentials = AWSCredentials(accessKey, accessSecret)
Show All 14 Lines

View Options

src/test/scala/mesosphere/marathon/api/v2/LeaderResourceTest.scala

1	1	package mesosphere.marathon
2	2	package api.v2
3	3
4	4	import mesosphere.UnitTest
5	5	import mesosphere.marathon.api.TestAuthFixture
6	6	import mesosphere.marathon.core.election.ElectionService
	7	import mesosphere.marathon.storage.repository.RuntimeConfigurationRepository
7	8
8	9	class LeaderResourceTest extends UnitTest {
9	10
10	11	"LeaderResource" should {
11	12	"access without authentication is denied" in {
12	13	Given("An unauthenticated request")
13	14	val f = new Fixture
14	15	val resource = f.leaderResource()
15	16	f.auth.authenticated = false
16	17
17	18	When("we try to get the leader info")
18	19	val index = resource.index(f.auth.request)
19	20	Then("we receive a NotAuthenticated response")
20	21	index.getStatus should be(f.auth.NotAuthenticatedStatus)
21	22
22	23	When("we try to delete the current leader")
23		val delete = resource.delete(f.auth.request)
	24	val delete = resource.delete(null, null, f.auth.request)
24	25	Then("we receive a NotAuthenticated response")
25	26	delete.getStatus should be(f.auth.NotAuthenticatedStatus)
26	27	}
27	28
28	29	"access without authorization is denied" in {
29	30	Given("An unauthenticated request")
30	31	val f = new Fixture
31	32	val resource = f.leaderResource()
32	33	f.auth.authenticated = true
33	34	f.auth.authorized = false
34	35
35	36	When("we try to get the leader info")
36	37	val index = resource.index(f.auth.request)
37	38	Then("we receive a Unauthorized response")
38	39	index.getStatus should be(f.auth.UnauthorizedStatus)
39	40
40	41	When("we try to delete the current leader")
41		val delete = resource.delete(f.auth.request)
	42	val delete = resource.delete(null, null, f.auth.request)
42	43	Then("we receive a Unauthorized response")
43	44	delete.getStatus should be(f.auth.UnauthorizedStatus)
44	45	}
45	46	}
46	47	class Fixture {
47	48	val schedulerService = mock[MarathonSchedulerService]
48	49	val electionService = mock[ElectionService]
	50	val runtimeRepo = mock[RuntimeConfigurationRepository]
49	51	val auth = new TestAuthFixture
50	52	val config = AllConf.withTestConfig("--event_subscriber", "http_callback")
51		def leaderResource() = new LeaderResource(electionService, config, auth.auth, auth.auth)
	53	def leaderResource() = new LeaderResource(electionService, config, runtimeRepo, auth.auth, auth.auth)
52	54	}
53	55	}
54	56

View Options

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

1	1	package mesosphere.marathon
2	2	package storage.migration
3	3
4	4	import akka.Done
5	5	import akka.stream.scaladsl.Source
6	6	import mesosphere.AkkaUnitTest
7	7	import mesosphere.marathon.Protos.StorageVersion
8	8	import mesosphere.marathon.core.storage.backup.PersistentStoreBackup
9	9	import mesosphere.marathon.core.storage.store.PersistenceStore
10	10	import mesosphere.marathon.core.storage.store.impl.memory.InMemoryPersistenceStore
	11	import mesosphere.marathon.storage.{ InMem, StorageConfig }
11	12	import mesosphere.marathon.storage.migration.StorageVersions._
12	13	import mesosphere.marathon.storage.repository._
13	14	import mesosphere.marathon.test.Mockito
14	15	import org.scalatest.GivenWhenThen
15	16
16	17	import scala.concurrent.Future
17	18
18	19	class MigrationTest extends AkkaUnitTest with Mockito with GivenWhenThen {
19	20
20	21	private[this] def migration(
21	22	persistenceStore: PersistenceStore[_, _, _] = new InMemoryPersistenceStore(),
22	23	appRepository: AppRepository = mock[AppRepository],
23	24	groupRepository: GroupRepository = mock[GroupRepository],
24	25	deploymentRepository: DeploymentRepository = mock[DeploymentRepository],
25	26	taskRepository: TaskRepository = mock[TaskRepository],
26	27	instanceRepository: InstanceRepository = mock[InstanceRepository],
27	28	taskFailureRepository: TaskFailureRepository = mock[TaskFailureRepository],
28	29	frameworkIdRepository: FrameworkIdRepository = mock[FrameworkIdRepository],
	30	configurationRepository: RuntimeConfigurationRepository = mock[RuntimeConfigurationRepository],
29	31	backup: PersistentStoreBackup = mock[PersistentStoreBackup],
30	32	eventSubscribersRepository: EventSubscribersRepository = mock[EventSubscribersRepository],
31		serviceDefinitionRepository: ServiceDefinitionRepository = mock[ServiceDefinitionRepository]): Migration = {
	33	serviceDefinitionRepository: ServiceDefinitionRepository = mock[ServiceDefinitionRepository],
	34	config: StorageConfig = InMem(1, Set.empty, None, None)): Migration = {
	35
	36	// assume no runtime config is stored in repository
	37	configurationRepository.get() returns Future.successful(None)
	38	configurationRepository.store(any) returns Future.successful(Done)
	39
32	40	new Migration(Set.empty, None, "bridge-name", persistenceStore, appRepository, groupRepository, deploymentRepository,
33	41	taskRepository, instanceRepository, taskFailureRepository, frameworkIdRepository,
34		eventSubscribersRepository, serviceDefinitionRepository, backup)
	42	eventSubscribersRepository, serviceDefinitionRepository, configurationRepository, backup, config)
35	43	}
36	44
37	45	val currentVersion: StorageVersion = StorageVersions.current
38	46
39	47	"Migration" should {
40	48	// we don't have any migrations yet
41	49	"be filterable by version" is pendingUntilFixed {
42	50	val migrate = migration()
▲ Show 20 Lines • Show All 93 Lines • Show Last 20 Lines

jenkins
• jasongilanfarr
meichstedt
timcharper

		Path
M		docs/docs/rest-api/public/api/api.raml (1 line)
M		docs/docs/rest-api/public/api/v2/leader.raml (25 lines)
M		docs/docs/rest-api/public/api/v2/types/pragma.raml (8 lines)
A	M	docs/docs/rest-api/public/api/v2/types/runtimeConfiguration.raml (19 lines)
M		src/main/scala/mesosphere/marathon/api/v2/LeaderResource.scala (21 lines)
M		src/main/scala/mesosphere/marathon/core/CoreGuiceModule.scala (6 lines)
M		src/main/scala/mesosphere/marathon/core/storage/backup/Backup.scala (6 lines)
M		src/main/scala/mesosphere/marathon/core/storage/backup/BackupConf.scala (2 lines)
M		src/main/scala/mesosphere/marathon/core/storage/backup/PersistentStoreBackup.scala (22 lines)
M		src/main/scala/mesosphere/marathon/core/storage/backup/impl/PersistentStoreBackupImpl.scala (14 lines)
M		src/main/scala/mesosphere/marathon/storage/StorageConfig.scala (23 lines)
M		src/main/scala/mesosphere/marathon/storage/StorageModule.scala (14 lines)
M		src/main/scala/mesosphere/marathon/storage/migration/Migration.scala (40 lines)
M		src/main/scala/mesosphere/marathon/storage/repository/Repositories.scala (28 lines)
M		src/main/scala/mesosphere/marathon/storage/store/InMemoryStoreSerialization.scala (10 lines)
M		src/main/scala/mesosphere/marathon/storage/store/ZkStoreSerialization.scala (23 lines)
M		src/main/scala/mesosphere/marathon/stream/UriIO.scala (5 lines)
M		src/test/scala/mesosphere/marathon/api/v2/LeaderResourceTest.scala (8 lines)
M		src/test/scala/mesosphere/marathon/storage/migration/MigrationTest.scala (12 lines)

Make Backup & Restore operations available via /v2/leader API
ClosedAll Users
Actions

Details

Diff Detail

Revision Contents

Diff 2875

docs/docs/rest-api/public/api/api.raml

docs/docs/rest-api/public/api/v2/leader.raml

docs/docs/rest-api/public/api/v2/types/pragma.raml

docs/docs/rest-api/public/api/v2/types/runtimeConfiguration.raml

src/main/scala/mesosphere/marathon/api/v2/LeaderResource.scala

src/main/scala/mesosphere/marathon/core/CoreGuiceModule.scala

src/main/scala/mesosphere/marathon/core/storage/backup/Backup.scala

src/main/scala/mesosphere/marathon/core/storage/backup/BackupConf.scala

src/main/scala/mesosphere/marathon/core/storage/backup/PersistentStoreBackup.scala

src/main/scala/mesosphere/marathon/core/storage/backup/impl/PersistentStoreBackupImpl.scala

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

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

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

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

src/main/scala/mesosphere/marathon/storage/store/InMemoryStoreSerialization.scala

src/main/scala/mesosphere/marathon/storage/store/ZkStoreSerialization.scala

src/main/scala/mesosphere/marathon/stream/UriIO.scala

src/test/scala/mesosphere/marathon/api/v2/LeaderResourceTest.scala

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

Make Backup & Restore operations available via /v2/leader APIClosedAll UsersActions

Details

Diff Detail

Revision Contents

Diff 2875

docs/docs/rest-api/public/api/api.raml

docs/docs/rest-api/public/api/v2/leader.raml

docs/docs/rest-api/public/api/v2/types/pragma.raml

docs/docs/rest-api/public/api/v2/types/runtimeConfiguration.raml

src/main/scala/mesosphere/marathon/api/v2/LeaderResource.scala

src/main/scala/mesosphere/marathon/core/CoreGuiceModule.scala

src/main/scala/mesosphere/marathon/core/storage/backup/Backup.scala

src/main/scala/mesosphere/marathon/core/storage/backup/BackupConf.scala

src/main/scala/mesosphere/marathon/core/storage/backup/PersistentStoreBackup.scala

src/main/scala/mesosphere/marathon/core/storage/backup/impl/PersistentStoreBackupImpl.scala

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

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

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

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

src/main/scala/mesosphere/marathon/storage/store/InMemoryStoreSerialization.scala

src/main/scala/mesosphere/marathon/storage/store/ZkStoreSerialization.scala

src/main/scala/mesosphere/marathon/stream/UriIO.scala

src/test/scala/mesosphere/marathon/api/v2/LeaderResourceTest.scala

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

Make Backup & Restore operations available via /v2/leader API
ClosedAll Users
Actions