Switch tests to use SSE event stream instead of webhooks
ClosedAll Users
Actions

			Path	Packages
M			mesos-simulation/src/test/scala/mesosphere/mesos/scale/SingleAppScalingTest.scala (10 lines)
M			project/Dependencies.scala (6 lines)
M			src/test/scala/mesosphere/marathon/integration/LeaderIntegrationTest.scala (2 lines)
M			src/test/scala/mesosphere/marathon/integration/MesosAppIntegrationTest.scala (4 lines)
M			src/test/scala/mesosphere/marathon/integration/NetworkPartitionIntegrationTest.scala (1 line)
M			src/test/scala/mesosphere/marathon/integration/RestartIntegrationTest.scala (1 line)
M			src/test/scala/mesosphere/marathon/integration/facades/MarathonFacade.scala (71 lines)
M			src/test/scala/mesosphere/marathon/integration/setup/MarathonTest.scala (201 lines)

Diff	ID	Base	Description	Created	Lint	Unit
Base			Base
Diff 1	2391	1a06163		Mar 24 2017, 11:04 PM	★	★
Diff 2	2392	1a06163	lol	Mar 25 2017, 3:37 AM	★	★
Diff 3	2393	60ff746	Prevent race conditions leading to flaky tests due to dropped SSE events	Mar 27 2017, 2:02 AM	★	★
Diff 4	2398	60ff746	try to fix NPE	Mar 27 2017, 3:49 PM	★	★
Diff 5	2399	60ff746	incorp feedback	Mar 27 2017, 3:58 PM	★	★
Diff 6	2405	60ff746	jason's requested changes	Mar 27 2017, 8:17 PM	★	★
Diff 7	2407	60ff746	Fix usage of withMarathon	Mar 27 2017, 11:09 PM	★	★
Diff 8	2408	60ff746	remove magic interval	Mar 27 2017, 11:11 PM	★	★
Diff 9	2409	60ff746	NetworkPartitionIntegrationTest passes again	Mar 27 2017, 11:38 PM	★	★
Diff 10	2410	779d949	fix compile error	Mar 28 2017, 12:18 AM	★	★
Diff 11	2414	39456b0	rebase	Mar 28 2017, 4:01 AM	★	★
Diff 12	2419	39456b0	lint errors	Mar 28 2017, 5:04 PM	★	★
Diff 13	2420	39456b0	♬ ~ ᕕ( ᐛ )ᕗ	Mar 28 2017, 5:29 PM	★	★
Diff 14	2421	39456b0	fix another build failure	Mar 28 2017, 6:09 PM	★	★
Diff 15	2446	39456b0	increase some logging; reduce number of times we cycle through leader election…	Mar 30 2017, 2:57 AM	★	★
Diff 16	2447	39456b0	update jira issues	Mar 30 2017, 2:58 AM	★	★
Diff 17	2455	165e94b	fix mesos-simulation test	Mar 30 2017, 10:27 PM	★	★
Diff 18	2459	165e94b	phase 1: start marathon. phase 2: ???, phase 3: start sse stream	Mar 31 2017, 1:55 AM	★	★
Diff 19	2460	165e94b	rMARATHON1c48c8e2d28cb5e8028aad010ba32b1883956a9f	Mar 31 2017, 3:59 AM	★	★

Commit	Tree	Parents	Author	Summary	Date
0b91f5753ba9	1180ce89935b	165e94ba65a2	Tim Harper	Switch tests to use SSE event stream instead of webhooks (Show More…)	Mar 24 2017, 11:00 PM

Diff 2460

View Options

mesos-simulation/src/test/scala/mesosphere/mesos/scale/SingleAppScalingTest.scala

1	1	package mesosphere.mesos.scale
2	2
3	3	import mesosphere.AkkaIntegrationTest
4	4	import mesosphere.marathon.IntegrationTest
5	5	import mesosphere.marathon.integration.facades.MarathonFacade._
6		import mesosphere.marathon.integration.facades.{ITDeploymentResult, MarathonFacade}
	6	import mesosphere.marathon.integration.facades.{ ITDeploymentResult, MarathonFacade }
7	7	import mesosphere.marathon.integration.setup._
8		import mesosphere.marathon.raml.{App, AppUpdate}
	8	import mesosphere.marathon.raml.{ App, AppUpdate }
9	9	import mesosphere.marathon.state.PathId
10	10	import org.scalatest.concurrent.Eventually
11	11	import org.slf4j.LoggerFactory
12	12	import play.api.libs.json._
13	13
14	14	import scala.collection.immutable.Seq
15	15	import scala.concurrent.duration._
	16	import scala.concurrent.Future
16	17
17	18	object SingleAppScalingTest {
18	19	val metricsFile = "scaleUp20s-metrics"
19	20	val appInfosFile = "scaleUp20s-appInfos"
20	21	}
21	22
22	23	@IntegrationTest
23	24	class SingleAppScalingTest extends AkkaIntegrationTest with ZookeeperServerTest with SimulatedMesosTest with MarathonTest with Eventually {
24	25
25	26	import PathId._
26	27
27	28	val maxTasksPerOffer = Option(System.getenv("MARATHON_MAX_TASKS_PER_OFFER")).getOrElse("1")
28	29
29	30	lazy val marathonServer = LocalMarathon(autoStart = false, suiteName = suiteName, "localhost:5050", zkUrl = s"zk://${zkServer.connectUri}/marathon", conf = Map(
30	31	"max_tasks_per_offer" -> maxTasksPerOffer,
31	32	"task_launch_timeout" -> "20000",
32	33	"task_launch_confirm_timeout" -> "1000"),
33	34	mainClass = "mesosphere.mesos.simulation.SimulateMesosMain")
34	35
	36	override lazy val leadingMarathon = Future.successful(marathonServer)
	37
35	38	override lazy val marathonUrl: String = s"http://localhost:${marathonServer.httpPort}"
36	39	override lazy val testBasePath: PathId = PathId.empty
37	40	override lazy val marathon: MarathonFacade = new MarathonFacade(marathonUrl, testBasePath)
38	41
39	42	private[this] val log = LoggerFactory.getLogger(getClass)
40	43
41	44	override def beforeAll(): Unit = {
42	45	super.beforeAll()
43	46	marathonServer.start().futureValue
	47	val sseStream = startEventSubscriber()
	48	system.registerOnTermination(sseStream.cancel())
	49	waitForSSEConnect()
44	50	}
45	51
46	52	override def afterAll(): Unit = {
47	53	// intentionally cleaning up before we print out the stats.
48	54	super.afterAll()
49	55	marathonServer.close()
50	56	println()
51	57	DisplayAppScalingResults.displayMetrics(SingleAppScalingTest.metricsFile)
▲ Show 20 Lines • Show All 97 Lines • Show Last 20 Lines

View Options

project/Dependencies.scala

Show First 20 Lines • Show All 56 Lines • ▼ Show 20 Line(s)
57	57	// test
58	58	Test.diffson % "test",
59	59	Test.scalatest % "test",
60	60	Test.mockito % "test",
61	61	Test.akkaTestKit % "test",
62	62	Test.junit % "test",
63	63	Test.scalacheck % "test",
64	64	Test.wixAccordScalatest % "test",
65		Test.curatorTest % "test"
	65	Test.curatorTest % "test",
	66	Test.akkaSse % "test"
66	67	) ++ Kamon.all).map(
67	68	_.excludeAll(excludeSlf4jLog4j12)
68	69	.excludeAll(excludeLog4j)
69	70	.excludeAll(excludeJCL)
70	71	.excludeAll(excludeAkkaHttpExperimental)
71	72	)
72	73
73	74	val benchmark = Seq(
▲ Show 20 Lines • Show All 47 Lines • ▼ Show 20 Line(s)
121	122
122	123	val excludeMortbayJetty = ExclusionRule(organization = "org.mortbay.jetty")
123	124	val excludeJavaxServlet = ExclusionRule(organization = "javax.servlet")
124	125
125	126	val alpakkaS3 = "com.lightbend.akka" %% "akka-stream-alpakka-s3" % V.Alpakka
126	127	val akkaActor = "com.typesafe.akka" %% "akka-actor" % V.Akka
127	128	val akkaSlf4j = "com.typesafe.akka" %% "akka-slf4j" % V.Akka
128	129	val akkaStream = "com.typesafe.akka" %% "akka-stream" % V.Akka
129		val akkaHttp = "com.typesafe.akka" %% "akka-http" % "10.0.1"
	130	val akkaHttp = "com.typesafe.akka" %% "akka-http" % "10.0.5"
130	131	val akkaHttpPlayJson = "de.heikoseeberger" %% "akka-http-play-json" % "1.10.1"
131	132	val asyncAwait = "org.scala-lang.modules" %% "scala-async" % V.AsyncAwait
132	133	val sprayClient = "io.spray" %% "spray-client" % V.Spray
133	134	val sprayHttpx = "io.spray" %% "spray-httpx" % V.Spray
134	135	val playJson = "com.typesafe.play" %% "play-json" % V.PlayJson
135	136	val chaos = "mesosphere" %% "chaos" % V.Chaos exclude("org.glassfish.web", "javax.el")
136	137	val guava = "com.google.guava" % "guava" % V.Guava
137	138	val mesos = "org.apache.mesos" % "mesos" % V.Mesos
▲ Show 20 Lines • Show All 48 Lines • ▼ Show 20 Line(s)
186	187	val scalatest = "org.scalatest" %% "scalatest" % V.ScalaTest
187	188	val mockito = "org.mockito" % "mockito-all" % V.Mockito
188	189	val akkaTestKit = "com.typesafe.akka" %% "akka-testkit" % V.Akka
189	190	val diffson = "org.gnieh" %% "diffson" % V.Diffson
190	191	val junit = "junit" % "junit" % V.JUnit
191	192	val scalacheck = "org.scalacheck" %% "scalacheck" % V.ScalaCheck
192	193	val wixAccordScalatest = "com.wix" %% "accord-scalatest" % V.WixAccord
193	194	val curatorTest = "org.apache.curator" % "curator-test" % V.Curator
	195	val akkaSse = "de.heikoseeberger" %% "akka-sse" % "2.0.0"
194	196	}
195	197	}

View Options

src/test/scala/mesosphere/marathon/integration/LeaderIntegrationTest.scala

Show First 20 Lines • Show All 105 Lines • ▼ Show 20 Line(s)
106	106	"zk_timeout" -> s"$zkTimeout"
107	107	)
108	108
109	109	override val numAdditionalMarathons = 2
110	110
111	111	"Reelecting a leader" should {
112	112	"it survives a small reelection test" in {
113	113
114		for (_ <- 1 to 15) {
	114	for (_ <- 1 to 8) {
115	115	Given("a leader")
116	116	WaitTestSupport.waitUntil("a leader has been elected", 30.seconds) { firstRunningProcess.client.leader().code == 200 }
117	117
118	118	// pick the leader to communicate with because it's the only known survivor
119	119	val leader = firstRunningProcess.client.leader().value
120	120	val leadingProcess: LocalMarathon = leadingServerProcess(leader.leader)
121	121	val client = leadingProcess.client
122	122
Show All 32 Lines

View Options

src/test/scala/mesosphere/marathon/integration/MesosAppIntegrationTest.scala

Show First 20 Lines • Show All 117 Lines • ▼ Show 20 Line(s)
118	118	"deploy a simple pod with health checks" taggedAs WhenEnvSet(envVar) in {
119	119	val projectDir = sys.props.getOrElse("user.dir", ".")
120	120
121	121	Given("a pod with two tasks that are health checked")
122	122	val podId = testBasePath / s"healthypod-${currentAppId.incrementAndGet()}"
123	123	val containerDir = "/opt/marathon"
124	124
125	125	def appMockCommand(port: String) = """echo APP PROXY $$MESOS_TASK_ID RUNNING; /opt/marathon/python/app_mock.py """ +
126		s"""$port $podId v1 http://127.0.0.1:${callbackEndpoint.localAddress.getPort}/health$podId/v1"""
	126	s"""$port $podId v1 http://127.0.0.1:${healthEndpoint.localAddress.getPort}/health$podId/v1"""
127	127
128	128	val pod = PodDefinition(
129	129	id = podId,
130	130	containers = Seq(
131	131	MesosContainer(
132	132	name = "task1",
133	133	exec = Some(raml.MesosExec(raml.ShellCommand(appMockCommand("$ENDPOINT_TASK1")))),
134	134	resources = raml.Resources(cpus = 0.1, mem = 32.0),
▲ Show 20 Lines • Show All 241 Lines • Show Last 20 Lines

View Options

src/test/scala/mesosphere/marathon/integration/NetworkPartitionIntegrationTest.scala

Show First 20 Lines • Show All 78 Lines • ▼ Show 20 Line(s)
79	79	zkServer.start()
80	80
81	81	mesosCluster.masters(0).start()
82	82	mesosCluster.agents(0).start()
83	83
84	84	And("Marathon is restarted by Systemd")
85	85	// Simulate Systemd rebooting Marathon
86	86	marathonServer.start()
	87	waitForSSEConnect()
87	88	eventually {
88	89	marathonServer.isRunning should be(true)
89	90	}
90	91
91	92	Then("The task reappears as running")
92	93	waitForEventMatching("Task is declared running again") {
93	94	matchEvent("TASK_RUNNING", task)
94	95	}
95	96	}
96	97	}
97	98	def matchEvent(status: String, task: ITEnrichedTask): CallbackEvent => Boolean = { event =>
98	99	event.info.get("taskStatus").contains(status) &&
99	100	event.info.get("taskId").contains(task.id)
100	101	}
101	102	}

View Options

src/test/scala/mesosphere/marathon/integration/RestartIntegrationTest.scala

Show First 20 Lines • Show All 120 Lines • ▼ Show 20 Line(s)
121	121	val serviceFacade1 = ServiceMockFacade(f.marathon.tasks(appId).value) { task =>
122	122	task.version.contains(newVersion) && task.launched
123	123	}
124	124	And("We trigger the first new task to continue service migration")
125	125	serviceFacade1.continue()
126	126
127	127	When("we force the leader to abdicate to simulate a failover")
128	128	server.restart().futureValue
	129	f.waitForSSEConnect()
129	130
130	131	And("second updated task becomes healthy")
131	132	val serviceFacade2 = ServiceMockFacade(f.marathon.tasks(appId).value) { task =>
132	133	task != serviceFacade1.task && task.version.contains(newVersion) && task.launched
133	134	}
134	135	And("We trigger the second new task to continue service migration")
135	136	serviceFacade2.continue()
136	137
Show All 36 Lines

View Options

src/test/scala/mesosphere/marathon/integration/facades/MarathonFacade.scala

1	1		package mesosphere.marathon
2	2		package integration.facades
3	3
	4		import com.typesafe.scalalogging.StrictLogging
4	5		import java.io.File
5	6		import java.util.Date
6	7
	8		import akka.NotUsed
7	9		import akka.actor.ActorSystem
	10		import akka.stream.Materializer
	11		import akka.stream.scaladsl.Source
	12		import com.fasterxml.jackson.databind.ObjectMapper
	13		import com.fasterxml.jackson.module.scala.DefaultScalaModule
	14		import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
	15		import de.heikoseeberger.akkasse.EventStreamUnmarshalling
	16		import akka.actor.ActorSystem
	17		import akka.stream.scaladsl.Source
8	18		import mesosphere.marathon.core.event.{ EventSubscribers, Subscribe, Unsubscribe }
9	19		import mesosphere.marathon.core.pod.PodDefinition
10	20		import mesosphere.marathon.integration.setup.{ RestResult, SprayHttpResponse }
11	21		import mesosphere.marathon.raml.{ App, AppUpdate, GroupInfo, GroupUpdate, Pod, PodConversion, PodInstanceStatus, PodStatus, Raml }
12	22		import mesosphere.marathon.state._
13	23		import mesosphere.marathon.stream.Implicits._
14	24		import mesosphere.marathon.util.Retry
15	25		import org.slf4j.LoggerFactory
16	26		import play.api.libs.functional.syntax._
17	27		import play.api.libs.json.JsArray
18	28		import spray.client.pipelining._
19	29		import spray.http._
20	30		import spray.httpx.PlayJsonSupport
	31		import akka.http.scaladsl.{ Http => AkkaHttp }
	32		import akka.http.scaladsl.client.RequestBuilding.{ Get => AkkaGet }
	33		import akka.http.scaladsl.model.{ MediaType => AkkaMediaType }
	34		import akka.http.scaladsl.model.headers.{ Accept => AkkaAccept }
	35		import akka.http.scaladsl.unmarshalling.{ Unmarshal => AkkaUnmarshal }
	36		import de.heikoseeberger.akkasse.ServerSentEvent
21	37
22	38		import scala.collection.immutable.Seq
23	39		import scala.concurrent.Await.result
	40		import scala.concurrent.Future
24	41		import scala.concurrent.duration._
25	42		import scala.util.control.NonFatal
26	43
27	44		/**
28	45		* GET /apps will deliver something like Apps instead of List[App]
29	46		* Needed for dumb jackson.
30	47		*/
31	48		case class ITAppDefinition(app: App)
Show All 12 Lines
44	61		stagedAt: Option[Date],
45	62		state: String,
46	63		version: Option[String]) {
47	64
48	65		def launched: Boolean = startedAt.nonEmpty
49	66		def suspended: Boolean = startedAt.isEmpty
50	67		}
51	68		case class ITLeaderResult(leader: String) {
52			val port = leader.split(":")(1)
	69		val port: Integer = leader.split(":")(1).toInt
53	70		}
54	71
55	72		case class ITListDeployments(deployments: Seq[ITDeployment])
56	73
57	74		case class ITQueueDelay(timeLeftSeconds: Int, overdue: Boolean)
58	75		case class ITQueueItem(app: App, count: Int, delay: ITQueueDelay)
59	76		case class ITLaunchQueue(queue: List[ITQueueItem])
60	77
61	78		case class ITDeployment(id: String, affectedApps: Seq[String], affectedPods: Seq[String])
62	79
	80		sealed trait ITSSEEvent
	81		/** Used to signal that the SSE stream is connected */
	82		case object ITConnected extends ITSSEEvent
	83
	84		/** models each SSE published event */
	85		case class ITEvent(eventType: String, info: Map[String, Any]) extends ITSSEEvent
	86
63	87		/**
64	88		* The MarathonFacade offers the REST API of a remote marathon instance
65	89		* with all local domain objects.
66	90		*
67	91		* @param url the url of the remote marathon instance
68	92		*/
69			class MarathonFacade(val url: String, baseGroup: PathId, waitTime: Duration = 30.seconds)(implicit val system: ActorSystem)
	93		class MarathonFacade(
	94		val url: String, baseGroup: PathId, waitTime: Duration = 30.seconds)(
	95		implicit
	96		val system: ActorSystem, mat: Materializer)
70	97		extends PlayJsonSupport
		jeschkiesUnsubmitted Not Done What changed the formatting? I'm curious about the new style.
		timcharperAuthorUnsubmitted Not Done it was longer than 120 characters.
		timcharperAuthorUnsubmitted Not Done (and this is how scalariform does it :) ) I don't love it... but, I try and just accept the will of the formatting tool.
		jeschkiesUnsubmitted Not Done That's fine with me :).
71			with PodConversion {
	98		with PodConversion with StrictLogging {
72	99		implicit val scheduler = system.scheduler
73	100		import SprayHttpResponse._
74	101		import mesosphere.marathon.core.async.ExecutionContexts.global
75	102
76	103		require(baseGroup.absolute)
77	104
78	105		import mesosphere.marathon.api.v2.json.Formats._
79	106		import mesosphere.marathon.integration.setup.V2TestFormats._
Show All 32 Lines
112	139		def requireInBaseGroup(pathId: PathId): Unit = {
113	140		require(isInBaseGroup(pathId), s"pathId $pathId must be in baseGroup ($baseGroup)")
114	141		}
115	142
116	143		def marathonSendReceive: SendReceive = {
117	144		addHeader("Accept", "/") ~> sendReceive
118	145		}
119	146
	147		// we don't want to lose any events and the default maxEventSize is too small (8K)
	148		object EventUnmarshalling extends EventStreamUnmarshalling {
	149		override protected def maxEventSize: Int = Int.MaxValue
	150		override protected def maxLineSize: Int = Int.MaxValue
	151		}
	152
	153		/**
	154		* Connects to the Marathon SSE endpoint. Future completes when the http connection is established. Events are
	155		* streamed via the materializable-once Source.
	156		*/
	157		def events(): Future[Source[ITEvent, NotUsed]] = {
	158
	159		import EventUnmarshalling.fromEventStream
	160		val mapper = new ObjectMapper() with ScalaObjectMapper
	161		mapper.registerModule(DefaultScalaModule)
	162
	163		AkkaHttp().singleRequest(AkkaGet(s"$url/v2/events").withHeaders(AkkaAccept(AkkaMediaType.text("event-stream"))))
	164		.flatMap { response =>
	165		AkkaUnmarshal(response).to[Source[ServerSentEvent, NotUsed]]
		jeschkiesUnsubmitted Done What is the motivation to have the imports here instead of top of the file?
		timcharperAuthorUnsubmitted Done This is the only function that uses akka http. The rest uses Spray. I put them at the top and things failed to compile. Maybe it'd be better to do alias? What do you think?
		jeschkiesUnsubmitted Done It's fine with me here as long as the imports are only used once.
		timcharperAuthorUnsubmitted Not Done I went ahead and aliased them.
	166		}
	167		.map { stream =>
		jeschkiesUnsubmitted Not Done Do we have mappers for the objects that we send from Marathon?
		timcharperAuthorUnsubmitted Not Done Can you further clarify what you mean by "mappers"? As in the mapper used to read JSON?
	168		stream
	169		.map { event =>
		jasongilanfarrUnsubmitted Done nit: combine stream.map into one line?
		timcharperAuthorUnsubmitted Not Done We've had this style discussion before :) And I agreed on leading periods. Operation-per-line is pretty common practice for akka streams code.
	170		event.data.map { d =>
	171		val json = mapper.readValue[Map[String, Any]](d) // linter:ignore
	172		ITEvent(event.`type`.getOrElse("unknown"), json)
	173		}
		jasongilanfarrUnsubmitted Done Missing "."
	174		}
	175		.collect { case Some(event) => event }
	176		}
	177		}
	178
120	179		//app resource ----------------------------------------------
		jeschkiesUnsubmitted Done Would `collect { case Some(foo) => foo }` achieve the same?
		timcharperAuthorUnsubmitted Done Totally would. Great idea.
121	180
122	181		def listAppsInBaseGroup: RestResult[List[App]] = {
123	182		val pipeline = marathonSendReceive ~> read[ITListAppsResult]
124	183		val res = result(pipeline(Get(s"$url/v2/apps")), waitTime)
125	184		res.map(_.apps.filterAs(app => isInBaseGroup(PathId(app.id)))(collection.breakOut))
126	185		}
127	186
128	187		def app(id: PathId): RestResult[ITAppDefinition] = {
▲ Show 20 Lines • Show All 251 Lines • ▼ Show 20 Line(s)
380	439
381	440		def deleteArtifact(path: String): RestResult[HttpResponse] = {
382	441		val pipeline = marathonSendReceive ~> responseResult
383	442		result(pipeline(Delete(s"$url/v2/artifacts$path")), waitTime)
384	443		}
385	444
386	445		//leader ----------------------------------------------
387	446		def leader(): RestResult[ITLeaderResult] = {
	447		result(leaderAsync(), waitTime)
		jasongilanfarrUnsubmitted Done Missing the retry?
		timcharperAuthorUnsubmitted Not Done good catch
	448		}
	449
	450		def leaderAsync(): Future[RestResult[ITLeaderResult]] = {
388	451		val pipeline = marathonSendReceive ~> read[ITLeaderResult]
389			result(Retry("leader") { pipeline(Get(s"$url/v2/leader")) }, waitTime)
	452		Retry("leader") { pipeline(Get(s"$url/v2/leader")) }
390	453		}
391	454
392	455		def abdicate(): RestResult[HttpResponse] = {
393	456		val pipeline = marathonSendReceive ~> responseResult
394	457		result(Retry("abdicate") { pipeline(Delete(s"$url/v2/leader")) }, waitTime)
395	458		}
396	459
397	460		//info --------------------------------------------------
Show All 30 Lines

View Options

src/test/scala/mesosphere/marathon/integration/setup/MarathonTest.scala

1	1		package mesosphere.marathon
2	2		package integration.setup
3	3
	4		import akka.actor.Cancellable
4	5		import java.io.File
5	6		import java.nio.file.Files
6	7		import java.util.UUID
7	8		import java.util.concurrent.ConcurrentLinkedQueue
8	9
9	10		import akka.Done
10	11		import akka.actor.{ ActorSystem, Scheduler }
11	12		import akka.http.scaladsl.Http
12	13		import akka.http.scaladsl.client.RequestBuilding.Get
13	14		import akka.http.scaladsl.model.{ HttpRequest, HttpResponse, StatusCodes }
14	15		import akka.http.scaladsl.unmarshalling.FromRequestUnmarshaller
15	16		import akka.stream.Materializer
	17		import akka.stream.scaladsl.Sink
16	18		import com.fasterxml.jackson.databind.ObjectMapper
17	19		import com.fasterxml.jackson.module.scala.DefaultScalaModule
18	20		import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
19	21		import com.typesafe.scalalogging.StrictLogging
20	22		import mesosphere.AkkaUnitTestLike
21	23		import mesosphere.marathon.api.RestResource
22			import mesosphere.marathon.integration.facades.{ ITEnrichedTask, ITLeaderResult, MarathonFacade, MesosFacade }
	24		import mesosphere.marathon.integration.facades.{ ITEnrichedTask, ITConnected, ITEvent, ITSSEEvent, ITLeaderResult, MarathonFacade, MesosFacade }
23	25		import mesosphere.marathon.raml.{ App, AppHealthCheck, AppVolume, PodState, PodStatus, ReadMode }
24	26		import mesosphere.marathon.state.PathId
25	27		import mesosphere.marathon.test.ExitDisabledTest
26	28		import mesosphere.marathon.util.{ Lock, Retry }
27	29		import mesosphere.util.PortAllocator
28	30		import org.apache.commons.io.FileUtils
29	31		import org.scalatest.concurrent.PatienceConfiguration.Timeout
30	32		import org.scalatest.concurrent.{ Eventually, ScalaFutures }
31	33		import org.scalatest.time.{ Milliseconds, Span }
32	34		import org.scalatest.{ BeforeAndAfterAll, Suite }
33			import play.api.libs.json.{ JsObject, JsString, Json }
	35		import play.api.libs.json.{ JsObject, Json }
34	36
35	37		import scala.annotation.tailrec
36	38		import scala.async.Async.{ async, await }
37	39		import scala.collection.mutable
38	40		import scala.concurrent.duration._
39	41		import scala.concurrent.{ ExecutionContext, Future }
40	42		import scala.sys.process.Process
41	43		import scala.util.Try
▲ Show 20 Lines • Show All 145 Lines • ▼ Show 20 Line(s)
187	189		}
188	190
189	191		/**
190	192		* Base trait for tests that need a marathon
191	193		*/
192	194		trait MarathonTest extends StrictLogging with ScalaFutures with Eventually {
193	195		def marathonUrl: String
194	196		def marathon: MarathonFacade
	197		def leadingMarathon: Future[LocalMarathon]
195	198		def mesos: MesosFacade
196	199		val testBasePath: PathId
197	200		def suiteName: String
198	201
199	202		val appProxyIds = Lock(mutable.ListBuffer.empty[String])
200	203
201	204		implicit val system: ActorSystem
202	205		implicit val mat: Materializer
203	206		implicit val ctx: ExecutionContext
204	207		implicit val scheduler: Scheduler
205	208
206	209		case class CallbackEvent(eventType: String, info: Map[String, Any])
	210		object CallbackEvent {
	211		def apply(event: ITEvent): CallbackEvent = CallbackEvent(event.eventType, event.info)
	212		}
207	213
208	214		implicit class CallbackEventToStatusUpdateEvent(val event: CallbackEvent) {
209	215		def taskStatus: String = event.info.get("taskStatus").map(_.toString).getOrElse("")
210	216		def message: String = event.info("message").toString
211	217		def id: String = event.info("id").toString
212	218		def running: Boolean = taskStatus == "TASK_RUNNING"
213	219		def finished: Boolean = taskStatus == "TASK_FINISHED"
214	220		def failed: Boolean = taskStatus == "TASK_FAILED"
215	221		}
216	222
217	223		object StatusUpdateEvent {
218	224		def unapply(event: CallbackEvent): Option[CallbackEvent] = {
219	225		if (event.eventType == "status_update_event") Some(event)
220	226		else None
221	227		}
222	228		}
223	229
224			protected val events = new ConcurrentLinkedQueue[CallbackEvent]()
	230		protected val events = new ConcurrentLinkedQueue[ITSSEEvent]()
225	231		protected val healthChecks = Lock(mutable.ListBuffer.empty[IntegrationHealthCheck])
226	232
227	233		/**
228	234		* Note! This is declared as lazy in order to prevent eager evaluation of values on which it depends
229	235		* We initialize it during the before hook and wait for Marathon to respond.
230	236		*/
231			protected[setup] lazy val callbackEndpoint = {
	237		protected[setup] lazy val healthEndpoint = {
232	238		val route = {
233	239		import akka.http.scaladsl.server.Directives._
234	240		val mapper = new ObjectMapper() with ScalaObjectMapper
235	241		mapper.registerModule(DefaultScalaModule)
236	242
237	243		implicit val unmarshal = new FromRequestUnmarshaller[Map[String, Any]] {
238	244		override def apply(value: HttpRequest)(implicit ec: ExecutionContext, materializer: Materializer): Future[Map[String, Any]] = {
239	245		value.entity.toStrict(5.seconds)(materializer).map { entity =>
240	246		mapper.readValue[Map[String, Any]](entity.data.utf8String)
241	247		}(ec)
242	248		}
243	249		}
244	250
245			(post & entity(as[Map[String, Any]])) { event =>
246			val kind = event.get("eventType") match {
247			case Some(JsString(s)) => s
248			case Some(s) => s.toString
249			case None => "unknown"
250			}
251			logger.info(s"Received callback event: $kind with props $event")
252			events.add(CallbackEvent(kind, event))
253			complete(HttpResponse(status = StatusCodes.OK))
254			} ~ get {
	251		get {
255	252		path("health" / Segments) { uriPath =>
256	253		import PathId._
257	254		val (path, remaining) = uriPath.splitAt(uriPath.size - 2)
258	255		val (versionId, port) = (remaining.head, remaining.tail.head.toInt)
259	256		val appId = path.mkString("/").toRootPath
260	257
261	258		def instance = healthChecks(_.find { c => c.appId == appId && c.versionId == versionId && c.port == port })
262	259		def definition = healthChecks(_.find { c => c.appId == appId && c.versionId == versionId && c.port == 0 })
263	260		val state = instance.orElse(definition).fold(true)(_.healthy)
264	261
265	262		logger.info(s"Received health check request: app=$appId, version=$versionId appMockPort=$port reply=$state")
266	263		if (state) {
267	264		complete(HttpResponse(status = StatusCodes.OK))
268	265		} else {
269	266		complete(HttpResponse(status = StatusCodes.InternalServerError))
270	267		}
271	268		} ~ path(Remaining) { path =>
272	269		require(false, s"$path was unmatched!")
273	270		complete(HttpResponse(status = StatusCodes.InternalServerError))
274	271		}
275	272		}
276	273		}
277	274		val port = PortAllocator.ephemeralPort()
278	275		val server = Http().bindAndHandle(route, "localhost", port).futureValue
279			marathon.subscribe(s"http://localhost:$port")
280			logger.info(s"Listening for events on $port")
	276		logger.info(s"Listening for health events on $port")
281	277		server
282	278		}
283	279
284	280		protected[setup] def killAppProxies(): Unit = {
285	281		val PIDRE = """^\s(\d+)\s+(.)$""".r
286	282		val allJavaIds = Process("jps -lv").!!.split("\n")
287	283		val pids = allJavaIds.collect {
288	284		case PIDRE(pid, exec) if appProxyIds(_.exists(exec.contains)) => pid
Show All 22 Lines
311	307
312	308		def appProxy(appId: PathId, versionId: String, instances: Int,
313	309		healthCheck: Option[raml.AppHealthCheck] = Some(appProxyHealthCheck()),
314	310		dependencies: Set[PathId] = Set.empty): App = {
315	311
316	312		val projectDir = sys.props.getOrElse("user.dir", ".")
317	313		val appMock: File = new File(projectDir, "src/test/python/app_mock.py")
318	314		val cmd = Some(s"""echo APP PROXY $$MESOS_TASK_ID RUNNING; ${appMock.getAbsolutePath} """ +
319			s"""$$PORT0 $appId $versionId http://127.0.0.1:${callbackEndpoint.localAddress.getPort}/health$appId/$versionId""")
	315		s"""$$PORT0 $appId $versionId http://127.0.0.1:${healthEndpoint.localAddress.getPort}/health$appId/$versionId""")
320	316
321	317		App(
322	318		id = appId.toString,
323	319		cmd = cmd,
324	320		executor = "//cmd",
325	321		instances = instances,
326	322		cpus = 0.01, mem = 32.0,
327	323		healthChecks = healthCheck.toSet,
328	324		dependencies = dependencies.map(_.toString)
329	325		)
330	326		}
331	327
332	328		def dockerAppProxy(appId: PathId, versionId: String, instances: Int, healthCheck: Option[AppHealthCheck] = Some(appProxyHealthCheck()), dependencies: Set[PathId] = Set.empty): App = {
333	329		val projectDir = sys.props.getOrElse("user.dir", ".")
334	330		val containerDir = "/opt/marathon"
335	331
336	332		val cmd = Some("""echo APP PROXY $$MESOS_TASK_ID RUNNING; /opt/marathon/python/app_mock.py """ +
337			s"""$$PORT0 $appId $versionId http://127.0.0.1:${callbackEndpoint.localAddress.getPort}/health$appId/$versionId""")
	333		s"""$$PORT0 $appId $versionId http://127.0.0.1:${healthEndpoint.localAddress.getPort}/health$appId/$versionId""")
338	334
339	335		App(
340	336		id = appId.toString,
341	337		cmd = cmd,
342	338		container = Some(raml.Container(
343	339		`type` = raml.EngineType.Docker,
344	340		docker = Some(raml.DockerContainer(
345	341		image = "python:3.4.6-alpine",
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Line(s)
438	434		def waitForEventWith(
439	435		kind: String,
440	436		fn: CallbackEvent => Boolean, maxWait: FiniteDuration = patienceConfig.timeout.toMillis.millis): CallbackEvent = {
441	437		waitForEventMatching(s"event $kind to arrive", maxWait) { event =>
442	438		event.eventType == kind && fn(event)
443	439		}
444	440		}
445	441
	442		/**
	443		* Consumes the next event from the events queue within deadline. Does not throw. Returns None if unable to return an
	444		* event by that time.
	445		*
	446		* @param deadline The time after which to stop attempting to get an event and return None
	447		*/
	448		private def nextEvent(deadline: Deadline): Option[ITSSEEvent] = try {
	449		eventually(timeout(Span(deadline.timeLeft.toMillis, Milliseconds))) {
	450		val r = Option(events.poll)
	451		if (r.isEmpty)
	452		throw new NoSuchElementException
	453		r
	454		}
	455		} catch {
	456		case _: NoSuchElementException =>
	457		None
	458		}
	459
446	460		def waitForEventMatching(
447	461		description: String,
448	462		maxWait: FiniteDuration = patienceConfig.timeout.toMillis.millis)(fn: CallbackEvent => Boolean): CallbackEvent = {
	463		val deadline = maxWait.fromNow
449	464		@tailrec
450			def matchingEvent: Option[CallbackEvent] = if (events.isEmpty) None else {
451			val event = events.poll()
452			if (fn(event)) {
453			Some(event)
	465		def iter(): CallbackEvent = {
	466		nextEvent(deadline) match {
	467		case Some(ITConnected) =>
	468		throw new MarathonTest.UnexpectedConnect
	469		case Some(event: ITEvent) =>
	470		val cbEvent = CallbackEvent(event)
	471		if (fn(cbEvent)) {
	472		cbEvent
454	473		} else {
455	474		logger.info(s"Event $event did not match criteria skipping to next event")
456			matchingEvent
	475		iter()
457	476		}
	477		case None =>
	478		throw new RuntimeException(s"No events matched <$description>")
458	479		}
	480		}
	481		iter()
	482		}
459	483
460			eventually(timeout(Span(maxWait.toMillis, Milliseconds))) {
461			require(!events.isEmpty, s"No events matched <$description>")
462			matchingEvent.getOrElse(throw new RuntimeException("No matching events"))
	484		/**
	485		* Blocks until a single connected event is consumed. Discards any events up to that point.
	486		*
	487		* Not reasoning about SSE connection state will lead to flaky tests. If a master is killed, you should wait for the
	488		* SSE stream to reconnect before doing anything else, or you could miss events.
	489		*/
	490		def waitForSSEConnect(maxWait: FiniteDuration = patienceConfig.timeout.toMillis.millis): Unit = {
	491		@tailrec
	492		val deadline = maxWait.fromNow
	493		def iter(): Unit = {
	494		nextEvent(deadline) match {
	495		case Some(event: ITEvent) =>
	496		logger.info(s"Event ${event} was not a connected event; skipping")
	497		iter()
	498		case Some(ITConnected) =>
	499		logger.info("ITConnected event consumed")
	500		case None =>
	501		throw new RuntimeException("No connected events")
463	502		}
464	503		}
	504		iter()
	505		}
465	506
466	507		/**
467	508		* Wait for the events of the given kinds (=types).
468	509		*/
469	510		def waitForEvents(kinds: String*)(maxWait: FiniteDuration = patienceConfig.timeout.toMillis.millis): Map[String, Seq[CallbackEvent]] = {
470	511
471	512		val deadline = maxWait.fromNow
472	513
Show All 32 Lines
505	546		}
506	547		}
507	548
508	549		protected[setup] def teardown(): Unit = {
509	550		Try {
510	551		val frameworkId = marathon.info.entityJson.as[JsObject].value("frameworkId").as[String]
511	552		mesos.teardown(frameworkId).futureValue
512	553		}
513			Try(marathon.unsubscribe(s"http://localhost:${callbackEndpoint.localAddress.getPort}"))
514			Try(callbackEndpoint.unbind().futureValue)
	554		Try(healthEndpoint.unbind().futureValue)
515	555		Try(killAppProxies())
516	556		}
	557
	558		/**
	559		* Connects repeatedly to the Marathon SSE endpoint until cancelled.
	560		* Yields each event in order.
	561		*/
	562		def startEventSubscriber(): Cancellable = {
	563		@volatile var cancelled = false
	564		def iter(): Unit = {
	565		import akka.stream.scaladsl.Source
	566		logger.info("SSEStream: Connecting")
	567		Source.fromFuture(leadingMarathon)
	568		.mapAsync(1) { leader =>
	569		async {
	570		logger.info(s"SSEStream: Acquiring connection to ${leader.url}")
	571		val stream = await(leader.client.events())
	572		logger.info(s"SSEStream: Connection acquired to ${leader.url}")
	573
	574		/* A potentially impossible edge case exists in which we query the leader, and then before we get a connection
	575		* to that instance, it restarts and is no longer a leader.
	576		*
	577		* By checking the leader again once obtaining a connection to the SSE event stream, we have conclusive proof
	578		* that we are consuming from the current leader, and we keep our connected events as deterministic as
	579		* possible. */
	580		val leaderAfterConnection = await(leadingMarathon)
	581		logger.info(s"SSEStream: ${leader.url} is the leader")
	582		if (leader != leaderAfterConnection) {
	583		stream.runWith(Sink.cancelled)
	584		throw new RuntimeException("Leader status changed since first connecting to stream")
	585		} else {
	586		stream
517	587		}
	588		}
	589		}
	590		.flatMapConcat { stream =>
	591		// We prepend the ITConnected event here in order to avoid emitting an ITConnected event on failed connections
	592		stream.prepend(Source.single(ITConnected))
	593		}
	594		.runForeach { e: ITSSEEvent =>
	595		e match {
	596		case ITConnected =>
	597		logger.info(s"SSEStream: Connected")
	598		case event: ITEvent =>
	599		logger.info(s"SSEStream: Received callback event: ${event.eventType} with props ${event.info}")
	600		}
	601		events.offer(e)
	602		}
	603		.onComplete {
	604		case result =>
	605		if (!cancelled) {
	606		logger.info(s"SSEStream: Leader event stream was closed reason: ${result}")
	607		logger.info("Reconnecting")
	608		scheduler.scheduleOnce(patienceConfig.interval) { iter() }
	609		}
	610		}
	611		}
	612		iter()
	613		new Cancellable {
	614		override def cancel(): Boolean = {
	615		cancelled = true
	616		true
	617		}
	618		override def isCancelled: Boolean = cancelled
	619		}
	620		}
	621		}
518	622
	623		object MarathonTest extends StrictLogging {
	624		class UnexpectedConnect extends Exception("Received an unexpected SSE event stream Connection event. This is " +
	625		"considered an exception because not thinking about re-connection events properly can lead to race conditions in " +
	626		"the tests. You should call waitForSSEConnect() after killing a Marathon leader to ensure no events are dropped.")
	627		}
	628
519	629		/**
520	630		* Fixture that can be used for a single test case.
521	631		*/
522	632		trait MarathonFixture extends AkkaUnitTestLike with MesosClusterTest with ZookeeperServerTest {
523	633		def withMarathon[T](suiteName: String, marathonArgs: Map[String, String] = Map.empty)(f: (LocalMarathon, MarathonTest) => T): T = {
524	634		val marathonServer = LocalMarathon(autoStart = false, suiteName = suiteName, masterUrl = mesosMasterUrl,
525	635		zkUrl = s"zk://${zkServer.connectUri}/marathon-$suiteName", conf = marathonArgs)
526	636		marathonServer.start().futureValue
527	637
528	638		val marathonTest = new MarathonTest {
529	639		override def marathonUrl: String = s"http://localhost:${marathonServer.httpPort}"
530	640		override def marathon: MarathonFacade = marathonServer.client
531	641		override def mesos: MesosFacade = MarathonFixture.this.mesos
532	642		override val testBasePath: PathId = PathId("/")
533	643		override implicit val system: ActorSystem = MarathonFixture.this.system
534	644		override implicit val mat: Materializer = MarathonFixture.this.mat
535	645		override implicit val ctx: ExecutionContext = MarathonFixture.this.ctx
536	646		override implicit val scheduler: Scheduler = MarathonFixture.this.scheduler
537	647		override val suiteName: String = MarathonFixture.this.suiteName
538	648		override implicit def patienceConfig: PatienceConfig = PatienceConfig(MarathonFixture.this.patienceConfig.timeout, MarathonFixture.this.patienceConfig.interval)
	649		override def leadingMarathon = Future.successful(marathonServer)
539	650		}
	651		val sseStream = marathonTest.startEventSubscriber()
540	652		try {
541			marathonTest.callbackEndpoint
	653		marathonTest.healthEndpoint
	654		marathonTest.waitForSSEConnect()
542	655		f(marathonServer, marathonTest)
543	656		} finally {
	657		sseStream.cancel()
544	658		marathonTest.teardown()
545	659		marathonServer.stop()
546	660		}
547	661		}
548	662		}
549	663
550	664		object MarathonFixture extends MarathonFixture
551	665
Show All 20 Lines
572	686		zkUrl = s"zk://${zkServer.connectUri}/marathon",
573	687		conf = marathonArgs)
574	688		lazy val marathonUrl = s"http://localhost:${marathonServer.httpPort}"
575	689
576	690		val testBasePath: PathId = PathId("/")
577	691		lazy val marathon = marathonServer.client
578	692		lazy val appMock: AppMockFacade = new AppMockFacade()
579	693
	694		/**
	695		* Return the current leading Marathon
	696		* Expected to retry for a significant period of time until succeeds
	697		*/
	698		override def leadingMarathon: Future[LocalMarathon] =
	699		Future.successful(marathonServer)
	700
	701		@volatile private var sseStream: Option[Cancellable] = None
	702
580	703		abstract override def beforeAll(): Unit = {
581	704		super.beforeAll()
582	705		marathonServer.start().futureValue(Timeout(90.seconds))
583			callbackEndpoint
	706		sseStream = Some(startEventSubscriber())
	707		waitForSSEConnect()
584	708		}
585	709
586	710		abstract override def afterAll(): Unit = {
	711		sseStream.foreach(_.cancel)
587	712		teardown()
588	713		Try(marathonServer.close())
589	714		super.afterAll()
590	715		}
591	716		}
592	717
593	718		/**
594	719		* trait that has marathon, zk, and a mesos ready to go
595	720		*/
596	721		trait EmbeddedMarathonTest extends Suite with StrictLogging with ZookeeperServerTest with MesosClusterTest with LocalMarathonTest {
597	722		// disable failover timeout to assist with cleanup ops; terminated marathons are immediately removed from mesos's
598	723		// list of frameworks
599	724		override def marathonArgs: Map[String, String] = Map("failover_timeout" -> "0")
600	725		}
		jeschkiesUnsubmitted Done Could we use async/await here? What do you think, would it make things more readable?
		timcharperAuthorUnsubmitted Done async / await is a great idea. I'll make the change.
		jeschkiesUnsubmitted Done Sweet. I totally missed: stream.runWith(Sink.cancelled) throw new RuntimeException("Leader status changed since first connecting to stream") thanks. A quick questions about streams. Where does the exception end up? Do we see it as a failed future?
		timcharperAuthorUnsubmitted Done It fails the stream which is communicated through the materialized foreach sink value (`Future[Done]`). I flattened everything into a stream because I wanted all failures to flow to the same point for easier recovery.
601	726
602	727		/**
603	728		* Trait that has a Marathon cluster, zk, and Mesos via mesos-local ready to go.
604	729		*
605	730		* It provides multiple Marathon instances. This allows e.g. leadership rotation.
606	731		*/
607	732		trait MarathonClusterTest extends Suite with StrictLogging with ZookeeperServerTest with MesosClusterTest with LocalMarathonTest {
		jeschkiesUnsubmitted Done I like how you introduce SSE without breaking our current API. Do you think we can get rid of the event list completely and check the stream directly as a next step?
		timcharperAuthorUnsubmitted Done I thought about this. I'm a bit hesitant to do it personally because it seems like it would be a pretty large change and be difficult to backport, and an ultimate goal is to land this patch on 1.4
		jeschkiesUnsubmitted Done Totally agree. Could you create a JIRA for the next step (1.5 only) if we don't have one yet.
		timcharperAuthorUnsubmitted Done Would it be best to discuss with the team first to see what direction we should go?
608	733		val numAdditionalMarathons = 2
609	734		lazy val additionalMarathons = 0.until(numAdditionalMarathons).map { _ =>
610	735		LocalMarathon(autoStart = false, suiteName = suiteName, masterUrl = mesosMasterUrl,
611	736		zkUrl = s"zk://${zkServer.connectUri}/marathon",
612	737		conf = marathonArgs)
613	738		}
614	739		lazy val marathonFacades = marathon +: additionalMarathons.map(_.client)
	740		lazy val allMarathonServers = marathonServer +: additionalMarathons
		jasongilanfarrUnsubmitted Done Why in a second? why not faster?
		timcharperAuthorUnsubmitted Not Done ¯\_(ツ)_/¯ Good point. I'll use patienceConfig.interval. For... reasons.
615	741
	742		override def leadingMarathon: Future[LocalMarathon] = {
	743		val leader = Retry("querying leader", maxAttempts = 50, maxDelay = 1.second, maxDuration = patienceConfig.timeout) {
	744		Future.firstCompletedOf(marathonFacades.map(_.leaderAsync()))
	745		}
	746		leader.map { leader =>
	747		allMarathonServers.find { _.httpPort == leader.value.port }.head
	748		}
	749		}
	750
616	751		override def beforeAll(): Unit = {
617	752		super.beforeAll()
618	753		Future.sequence(additionalMarathons.map(_.start())).futureValue(Timeout(60.seconds))
619	754		}
620	755
621	756		override def afterAll(): Unit = {
622	757		Try(additionalMarathons.foreach(_.close()))
623	758		super.afterAll()
624	759		}
625	760
626	761		def nonLeader(leader: ITLeaderResult): MarathonFacade = {
627			marathonFacades.find(!_.url.contains(leader.port)).get
	762		marathonFacades.find(!_.url.contains(leader.port.toString)).get
628	763		}
629	764
630	765		override def cleanUp(withSubscribers: Boolean): Unit = {
631	766		Future.sequence(marathonServer.start() +: additionalMarathons.map(_.start())).futureValue
632	767		super.cleanUp(withSubscribers)
633	768		}
634	769		}
635	770

Switch tests to use SSE event stream instead of webhooksClosedAll UsersActions

Details

Diff Detail

Revision Contents

Diff 2460

mesos-simulation/src/test/scala/mesosphere/mesos/scale/SingleAppScalingTest.scala

project/Dependencies.scala

src/test/scala/mesosphere/marathon/integration/LeaderIntegrationTest.scala

src/test/scala/mesosphere/marathon/integration/MesosAppIntegrationTest.scala

src/test/scala/mesosphere/marathon/integration/NetworkPartitionIntegrationTest.scala

src/test/scala/mesosphere/marathon/integration/RestartIntegrationTest.scala

src/test/scala/mesosphere/marathon/integration/facades/MarathonFacade.scala

src/test/scala/mesosphere/marathon/integration/setup/MarathonTest.scala

Switch tests to use SSE event stream instead of webhooks
ClosedAll Users
Actions