Add a system controller which implements the functionality of SystemResource.
ClosedAll Users
Actions

src/main/scala/mesosphere/marathon/api/akkahttp/SystemController.scala
121	`ping` does not. Patch D783 introduces a `authenticatedAndAuthorized` directive, which would make sense here. doing `authenticate` in `route` would mean to declare an implicit identity on every sub-route.

Closed by commit rMARATHONc8e4d007c147: Add a system controller which implements the functionality of SystemResource. (authored by aquamatthias, committed by Mesosphere CI Robot <mesosphere-ci@users.noreply.github.com>). · Explain WhyMay 15 2017, 1:48 PM

This revision was automatically updated to reflect the committed changes.

Revision Contents

		Path
M		src/main/scala/mesosphere/marathon/api/akkahttp/AkkaHttpMarathonService.scala (8 lines)
M		src/main/scala/mesosphere/marathon/api/akkahttp/AkkaHttpModule.scala (5 lines)
M		src/main/scala/mesosphere/marathon/api/akkahttp/EntityMarshallers.scala (24 lines)
A	M	src/main/scala/mesosphere/marathon/api/akkahttp/SystemController.scala (133 lines)

Diff	ID	Base	Description	Created	Lint	Unit
Base			Base
Diff 1	3159	685dac6		May 11 2017, 4:49 PM	★	★
Diff 2	3174	a2e4cfd	rebase to master	May 12 2017, 11:34 AM	★	★
Diff 3	3195	a17f069	rMARATHONc8e4d007c147e5d084577ed95eff0d429ee7440a	May 15 2017, 1:31 PM	★	★

Commit	Tree	Parents	Author	Summary	Date
768eee2939e3	d375a5434d54	a2e4cfd48931	Matthias Veit	Add a system controller which implements the functionality of SystemResource. (Show More…)	May 11 2017, 4:41 PM

Diff 3195

View Options

src/main/scala/mesosphere/marathon/api/akkahttp/AkkaHttpMarathonService.scala

Show All 14 Lines
15	15	import scala.async.Async._
16	16	import akka.http.scaladsl.model.StatusCodes._
17	17	import akka.http.scaladsl.server._
18	18	import mesosphere.marathon.api.akkahttp.Rejections.Message
19	19	import play.api.libs.json.Json
20	20
21	21	class AkkaHttpMarathonService(
22	22	config: MarathonConf with HttpConf,
	23	systemController: SystemController,
23	24	v2Controller: V2Controller
24	25	)(
25	26	implicit
26	27	actorSystem: ActorSystem) extends AbstractIdleService with MarathonHttpService with StrictLogging {
27	28	import actorSystem.dispatcher
28	29	implicit val materializer = ActorMaterializer()
29	30	private var handler: Option[Future[Http.ServerBinding]] = None
30	31	import Directives._
Show All 14 Lines
45	46	// Map RejectionResponse is the akka http suggested way of enforcing other content types.
46	47	case res @ HttpResponse(_, _, HttpEntity.Strict(ContentTypes.`text/plain(UTF-8)`, data), _) =>
47	48	val message = Json.stringify(Json.toJson(Message(data.utf8String)))
48	49	res.copy(entity = HttpEntity(ContentTypes.`application/json`, message))
49	50	case response: HttpResponse => response
50	51	}
51	52
52	53	val route: Route = {
	54	systemController.route ~
53	55	pathPrefix("v2") {
54	56	v2Controller.route
55	57	}
56	58	}
57	59
58	60	override def startUp(): Unit = synchronized {
59	61	if (handler.isEmpty) {
60	62	logger.info(s"Listening via Akka HTTP on ${config.httpPort()}")
Show All 28 Lines

View Options

src/main/scala/mesosphere/marathon/api/akkahttp/AkkaHttpModule.scala

1	1	package mesosphere.marathon
2	2	package api.akkahttp
3	3
4	4	import akka.actor.ActorSystem
5	5	import akka.event.EventStream
6	6	import com.google.inject.AbstractModule
7	7	import com.google.inject.{ Provides, Scopes, Singleton }
	8	import com.typesafe.config.Config
8	9	import mesosphere.chaos.http.HttpConf
9	10	import mesosphere.marathon.api.MarathonHttpService
10	11	import mesosphere.marathon.core.appinfo._
11	12	import mesosphere.marathon.core.base.Clock
12	13	import mesosphere.marathon.core.election.ElectionService
13	14	import mesosphere.marathon.core.group.GroupManager
14	15	import mesosphere.marathon.core.plugin.PluginManager
15	16	import mesosphere.marathon.plugin.auth._
16	17	import v2.AppsController
17	18
18	19	class AkkaHttpModule(conf: MarathonConf with HttpConf) extends AbstractModule {
19	20	override def configure(): Unit = {
20	21	bind(classOf[MarathonHttpService]).to(classOf[AkkaHttpMarathonService]).in(Scopes.SINGLETON)
21	22	}
22	23
23	24	@Provides
24	25	@Singleton
25	26	@SuppressWarnings(Array("MaxParameters"))
26	27	def provideAkkaHttpMarathonService(
27	28	clock: Clock,
	29	config: Config,
28	30	eventBus: EventStream,
29	31	appInfoService: AppInfoService,
30	32	groupManager: GroupManager,
31	33	pluginManager: PluginManager,
32	34	marathonSchedulerService: MarathonSchedulerService,
33	35	appTasksRes: mesosphere.marathon.api.v2.AppTasksResource)(implicit
34	36	actorSystem: ActorSystem,
35	37	authenticator: Authenticator,
36	38	authorizer: Authorizer,
37	39	electionService: ElectionService): AkkaHttpMarathonService = {
38	40
39	41	import actorSystem.dispatcher
40	42	val appsController = new AppsController(
41	43	clock = clock,
42	44	eventBus = eventBus,
43	45	appTasksRes = appTasksRes,
44	46	service = marathonSchedulerService,
45	47	appInfoService = appInfoService,
46	48	config = conf,
47	49	groupManager = groupManager,
48	50	pluginManager = pluginManager)
49	51
	52	val systemController = new SystemController(config)
	53
50	54	val v2Controller = new V2Controller(appsController)
51	55	new AkkaHttpMarathonService(
52	56	conf,
	57	systemController,
53	58	v2Controller)
54	59	}
55	60	}

View Options

src/main/scala/mesosphere/marathon/api/akkahttp/EntityMarshallers.scala

1	1	package mesosphere.marathon
2	2	package api.akkahttp
3	3
4	4	import akka.http.scaladsl.marshalling.{ Marshaller, ToEntityMarshaller }
5	5	import akka.http.scaladsl.model.MediaTypes.`application/json`
6	6	import akka.http.scaladsl.model.StatusCodes
7	7	import akka.http.scaladsl.server.{ Rejection, RejectionError, Route }
8	8	import akka.http.scaladsl.unmarshalling.{ FromEntityUnmarshaller, Unmarshaller }
9	9	import akka.util.ByteString
10	10	import com.wix.accord.{ Failure, Success, Validator }
	11	import kamon.metric.SubscriptionsDispatcher.TickMetricSnapshot
11	12	import mesosphere.marathon.api.v2.Validation
12	13	import mesosphere.marathon.core.appinfo.AppInfo
	14	import mesosphere.marathon.raml.{ LoggerChange, Metrics }
13	15	import mesosphere.marathon.state.AppDefinition
14	16	import play.api.libs.json._
15	17
16	18	object EntityMarshallers {
17	19	import Directives.complete
18	20	import mesosphere.marathon.api.v2.json.Formats._
	21	import mesosphere.marathon.raml.MetricsConversion._
19	22
20	23	private val jsonStringUnmarshaller =
21	24	Unmarshaller.byteStringUnmarshaller
22	25	.forContentTypes(`application/json`)
23	26	.mapWithCharset {
24	27	case (ByteString.empty, _) => throw Unmarshaller.NoContentException
25	28	case (data, charset) => data.decodeString(charset.nioCharset.name)
26	29	}
Show All 32 Lines
59	62	*/
60	63	private def playJsonMarshaller[A](
61	64	implicit
62	65	writes: Writes[A],
63	66	printer: JsValue => String = Json.prettyPrint
64	67	): ToEntityMarshaller[A] =
65	68	jsonStringMarshaller.compose(printer).compose(writes.writes)
66	69
	70	/**
	71	* `Internal` => `Raml` => HTTP entity
	72	* @param ramlWrites converter that translates between the internal and the raml type.
	73	* @param writes the play json formatter, that converts the raml type into a json object
	74	* @param printer the printer, that converts a json object into a string
	75	* @tparam Internal the internal type
	76	* @tparam Raml the raml type
	77	* @return a ToEntityMarshaller for the Internal type
	78	*/
	79	private def internalToRamlJsonMarshaller[Internal, Raml](
	80	implicit
	81	ramlWrites: raml.Writes[Internal, Raml],
	82	writes: Writes[Raml],
	83	printer: JsValue => String = Json.prettyPrint
	84	): ToEntityMarshaller[Internal] =
	85	jsonStringMarshaller.compose(printer).compose(internal => writes.writes(ramlWrites.write(internal)))
	86
67	87	import mesosphere.marathon.raml.AppConversion.appRamlReader
68	88	implicit def appUnmarshaller(
69	89	implicit
70	90	normalization: Normalization[raml.App], validator: Validator[AppDefinition]) = {
71	91	validEntityRaml(playJsonUnmarshaller[raml.App])
72	92	}
73	93
74	94	implicit val jsValueMarshaller = playJsonMarshaller[JsValue]
75	95	implicit val wixResultMarshaller = playJsonMarshaller[com.wix.accord.Failure](Validation.failureWrites)
76	96	implicit val messageMarshaller = playJsonMarshaller[Rejections.Message]
77	97	implicit val appInfoMarshaller = playJsonMarshaller[AppInfo]
	98	implicit val metricsMarshaller = internalToRamlJsonMarshaller[TickMetricSnapshot, Metrics]
	99	implicit val loggerChangeMarshaller = playJsonMarshaller[LoggerChange]
	100	implicit val loggerChangeUnmarshaller = playJsonUnmarshaller[LoggerChange]
	101	implicit val stringMapMarshaller = playJsonMarshaller[Map[String, String]]
78	102
79	103	private def validEntityRaml[A, B](um: FromEntityUnmarshaller[A])(
80	104	implicit
81	105	normalization: Normalization[A], reader: raml.Reads[A, B], validator: Validator[B]): FromEntityUnmarshaller[B] = {
82	106	um.map { ent =>
83	107	try {
84	108	// Note: Normalization also validates which can throw an exception
85	109	val normalized = reader.read(normalization.normalized(ent))
Show All 19 Lines

View Options

src/main/scala/mesosphere/marathon/api/akkahttp/SystemController.scala

This file was added.

1		package mesosphere.marathon
2		package api.akkahttp
3
4		import akka.actor.ActorSystem
5		import akka.http.scaladsl.server.Route
6		import ch.qos.logback.classic.{ Level, Logger, LoggerContext }
7		import com.typesafe.config.{ Config, ConfigRenderOptions }
8		import com.typesafe.scalalogging.StrictLogging
9		import mesosphere.marathon.core.election.ElectionService
10		import mesosphere.marathon.metrics.Metrics
11		import mesosphere.marathon.plugin.auth.AuthorizedResource.{ SystemConfig, SystemMetrics }
12		import mesosphere.marathon.plugin.auth.{ Authenticator, Authorizer, UpdateResource, ViewResource }
13		import mesosphere.marathon.raml.LoggerChange
14		import org.slf4j.LoggerFactory
15		import stream.Implicits._
16
17		import scala.concurrent.duration._
18		import scala.concurrent.ExecutionContext
19
20		/**
21		* The SystemController handles system level functionality like configuration, metrics and logging.
22		*/
23		class SystemController(cfg: Config)(
24		implicit
25		val actorSystem: ActorSystem,
26		val executionContext: ExecutionContext,
27		val authenticator: Authenticator,
28		val authorizer: Authorizer,
29		val electionService: ElectionService
30		) extends Controller with StrictLogging {
31
32		import Directives._
33		import EntityMarshallers._
34
35		/**
36		* GET /ping
37		* @return a simple pong as text/plain
38		*/
39		def ping: Route = complete("pong")
40
41		/**
42		* GET /metrics
43		* @return a snapshot of all system level metrics.
44		*/
45		def metrics: Route = {
46		authenticated.apply { implicit identity =>
47		authorized(ViewResource, SystemMetrics).apply {
48		complete(Metrics.snapshot())
49		}
50		}
51		}
52
53		/**
54		* GET /config
55		* @return the complete application config (typesafe config)
56		*/
57		def config: Route = {
58		authenticated.apply { implicit identity =>
59		authorized(ViewResource, SystemMetrics).apply {
60		complete(cfg.root().render(ConfigRenderOptions.defaults().setJson(true)))
61		}
62		}
63		}
64
65		/**
66		* GET /logging
67		* @return a map of all loggers with related log level
68		*/
69		def showLoggers: Route = {
70		authenticated.apply { implicit identity =>
71		authorized(ViewResource, SystemConfig).apply {
72		LoggerFactory.getILoggerFactory match {
73		case lc: LoggerContext =>
74		complete(lc.getLoggerList.map { logger =>
75		logger.getName -> Option(logger.getLevel).map(_.levelStr).getOrElse(logger.getEffectiveLevel.levelStr + " (inherited)")
76		}.toMap[String, String])
77		}
78		}
79		}
80		}
81
82		/**
83		* POST /logging
84		* @return the log change.
85		*/
86		def changeLoggers: Route = {
87		authenticated.apply { implicit identity =>
88		authorized(UpdateResource, SystemConfig).apply {
89		entity(as[LoggerChange]) { change =>
90		LoggerFactory.getILoggerFactory.getLogger(change.logger) match {
91		case log: Logger =>
92		val level = Level.valueOf(change.level.value.toUpperCase)
93
94		// current level can be null, which means: use the parent level
95		// the current level should be preserved, no matter what the effective level is
96		val currentLevel = log.getLevel
97		val currentEffectiveLevel = log.getEffectiveLevel
98		logger.info(s"Set logger ${log.getName} to $level current: $currentEffectiveLevel")
99		log.setLevel(level)
100
101		// if a duration is given, we schedule a timer to reset to the current level
102		change.durationSeconds.foreach(duration => actorSystem.scheduler.scheduleOnce(duration.seconds, new Runnable {
103		override def run(): Unit = {
104		logger.info(s"Duration expired. Reset Logger ${log.getName} back to $currentEffectiveLevel")
105		log.setLevel(currentLevel)
106		}
107		}))
108		complete(change)
109		}
110		}
111		}
112		}
113		}
114
115		override val route: Route = {
116		// To maintain the functionality of the original API, this endpoint is leader aware.
117		// It would make sense to allow the functionality of this controller on every instance.
118		asLeader(electionService) {
119		path("ping") {
120		get { ping }
121		} ~
	timcharperUnsubmitted Not Done looks like all of these routes are authenticated; does it make sense to extract out the authenticated directive call to this outer route?
	aquamatthiasAuthorUnsubmitted Not Done `ping` does not. Patch D783 introduces a `authenticatedAndAuthorized` directive, which would make sense here. doing `authenticate` in `route` would mean to declare an implicit identity on every sub-route.
122		path("metrics") {
123		get { metrics }
124		} ~
125		path("config") {
126		get { config }
127		} ~
128		path("logging") {
129		get { showLoggers } ~ post { changeLoggers }
130		}
131		}
132		}
133		}

Add a system controller which implements the functionality of SystemResource.ClosedAll UsersActions

Details

Diff Detail

Revision Contents

Diff 3195

src/main/scala/mesosphere/marathon/api/akkahttp/AkkaHttpMarathonService.scala

src/main/scala/mesosphere/marathon/api/akkahttp/AkkaHttpModule.scala

src/main/scala/mesosphere/marathon/api/akkahttp/EntityMarshallers.scala

src/main/scala/mesosphere/marathon/api/akkahttp/SystemController.scala

Add a system controller which implements the functionality of SystemResource.
ClosedAll Users
Actions