HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D859

Add timeout handling to http health checks and readiness checks
ClosedAll Users
Actions

Authored by unterstein on Jun 16 2017, 9:12 AM.

Details

Reviewers
zen-dog
jeschkies
timcharper
jenkins
kensipe
Commits
rMARATHONd06cca944ccc: Add timeout handling to http health checks and readiness checks
rMARATHONad4517f49aef: adapted feedback
rMARATHON9f85a4aa2473: Add timeout handling to http health checks and readiness checks
JIRA Issues
JIRA MARATHON-7520 1.5 Health Checks Broken
Summary

Due to the akka http introduction, proper timeout handling for http health checks and readiness checks were gone.
This commits re-introduces the semantically correct timeout for these checks.

Test Plan

sbt "integration:test-only *AppDeployIntegrationTest"

Diff Detail

Repository
rMARATHON marathon
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.
Changes from before your most recent comment are hidden. Show Older Changes
jeschkies accepted this revision.Jun 16 2017, 12:16 PM

Thanks!

This revision is now accepted and ready to land.Jun 16 2017, 12:16 PM
jenkins accepted this revision.Jun 16 2017, 12:31 PM
✔ Build of 3522 completed jenkins-public-marathon-phabricator-250.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-582-g1793076.tgz",
"sha1"" "5d8e93bb8197f4459a52ba900083123d85b7abdf"

＼\ ٩( ᐛ )و /／

zen-dog accepted this revision.Jun 16 2017, 1:12 PM

Thx!

src/main/scala/mesosphere/marathon/util/package.scala
31 ↗(On Diff #3521)

Very elegant :thumbup:

kensipe accepted this revision.Jun 16 2017, 4:06 PM
kensipe added a subscriber: kensipe.

nice work! and excellent work adding in the test!

kensipe added a comment.Jun 16 2017, 4:11 PM

BTW love the Timeout that @jeschkies recommended! super elegant!

zen-dog requested changes to this revision.Jun 16 2017, 5:12 PM

Wait, why did you change it back to that ugly Timeout.util thing? It should be the other way around! Throw the util thing out and replace it with your!

This revision now requires changes to proceed.Jun 16 2017, 5:12 PM
unterstein added a comment.Jun 19 2017, 5:18 PM

@zen-dog unfortunately it is not possible to superseed Timeout with FutureTimeoutLike because Timeout works with infinite durations and FutureTimeoutLike not :(

This revision is now accepted and ready to land.Jun 19 2017, 6:57 PM
Closed by commit rMARATHON9f85a4aa2473: Add timeout handling to http health checks and readiness checks (authored by unterstein). · Explain WhyJun 19 2017, 7:50 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents

Diff 3547

View Options

src/main/scala/mesosphere/marathon/core/health/impl/HealthCheckWorkerActor.scala

11package mesosphere.marathon
22package core.health.impl
33
44import java.net.{ InetSocketAddress, Socket }
55import java.security.cert.X509Certificate
66import javax.net.ssl.{ KeyManager, SSLContext, X509TrustManager }
77
88import akka.actor.{ Actor, PoisonPill }
99import akka.http.scaladsl.client.RequestBuilding
1010import akka.http.scaladsl.{ ConnectionContext, Http }
1111import akka.stream.Materializer
1212import com.typesafe.scalalogging.StrictLogging
1313import com.typesafe.sslconfig.akka.AkkaSSLConfig
1414import mesosphere.marathon.core.health._
1515import mesosphere.marathon.core.instance.Instance
1616import mesosphere.marathon.state.{ AppDefinition, Timestamp }
17import mesosphere.marathon.util.Timeout
1718import mesosphere.util.ThreadPoolContext
1819
1920import scala.concurrent.Future
2021import scala.util.{ Failure, Success }
2122
2223class HealthCheckWorkerActor(implicit mat: Materializer) extends Actor with StrictLogging {
2324
2425 import HealthCheckWorker._
2526
2627 private implicit val system = context.system
28 private implicit val scheduler = system.scheduler
2729 import context.dispatcher // execution context for futures
2830
2931 def receive: Receive = {
3032 case HealthCheckJob(app, instance, check) =>
3133 val replyTo = sender() // avoids closing over the volatile sender ref
3234
3335 doCheck(app, instance, check)
3436 .andThen {
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Line(s)
8587 check: MarathonHttpHealthCheck,
8688 host: String,
8789 port: Int): Future[Option[HealthResult]] = {
8890 val rawPath = check.path.getOrElse("")
8991 val absolutePath = if (rawPath.startsWith("/")) rawPath else s"/$rawPath"
9092 val url = s"http://$host:$port$absolutePath"
9193 logger.debug(s"Checking the health of [$url] for instance=${instance.instanceId} via HTTP")
9294
93 Http().singleRequest(
94 RequestBuilding.Get(url)).map { response =>
95 Timeout(check.timeout)(Http().singleRequest(
96 RequestBuilding.Get(url)
97 )).map { response =>
9598 response.discardEntityBytes() //forget about the body
9699 if (acceptableResponses.contains(response.status.intValue())) {
97100 Some(Healthy(instance.instanceId, instance.runSpecVersion))
98101 } else if (check.ignoreHttp1xx && (toIgnoreResponses.contains(response.status.intValue))) {
99102 logger.debug(s"Ignoring health check HTTP response ${response.status.intValue} for instance=${instance.instanceId}")
100103 None
101104 } else {
102105 logger.debug(s"Health check for instance=${instance.instanceId} responded with ${response.status}")
Show All 39 Lines
142145 .withAllowLegacyHelloMessages(Some(true))
143146 .withAllowUnsafeRenegotiation(Some(true))
144147 .withAllowWeakCiphers(true)
145148 .withAllowWeakProtocols(true)
146149 .withDisableHostnameVerification(true)
147150 .withDisableSNI(true)
148151 })
149152
150 Http().singleRequest(
153 Timeout(check.timeout)(Http().singleRequest(
151154 RequestBuilding.Get(url),
152 connectionContext = ConnectionContext.https(disabledSslContext, sslConfig = Some(disabledSslConfig))).map { response =>
155 connectionContext = ConnectionContext.https(disabledSslContext, sslConfig = Some(disabledSslConfig))
156 )).map { response =>
153157 response.discardEntityBytes() // forget about the body
154158 if (acceptableResponses.contains(response.status.intValue())) {
155159 Some(Healthy(instance.instanceId, instance.runSpecVersion))
156160 } else {
157161 logger.debug(s"Health check for ${instance.instanceId} responded with ${response.status}")
158162 Some(Unhealthy(instance.instanceId, instance.runSpecVersion, response.status.toString()))
159163 }
160164 }
Show All 26 Lines
View Options

src/main/scala/mesosphere/marathon/core/readiness/impl/ReadinessCheckExecutorImpl.scala

11package mesosphere.marathon
22package core.readiness.impl
33
44import akka.actor.ActorSystem
55import akka.http.scaladsl.Http
66import akka.http.scaladsl.model.headers.`Content-Type`
77import akka.http.scaladsl.client.RequestBuilding
8import akka.http.scaladsl.settings.ConnectionPoolSettings
98import akka.http.scaladsl.model.{ MediaTypes, StatusCodes, HttpResponse => AkkaHttpResponse }
109import akka.stream.Materializer
1110import mesosphere.marathon.core.readiness.ReadinessCheckExecutor.ReadinessCheckSpec
1211import mesosphere.marathon.core.readiness.{ HttpResponse, ReadinessCheckExecutor, ReadinessCheckResult }
12import mesosphere.marathon.util.Timeout
1313import org.slf4j.LoggerFactory
1414import rx.lang.scala.Observable
1515
1616import scala.concurrent.Future
1717import scala.concurrent.duration.FiniteDuration
1818import scala.util.Success
1919import scala.util.control.NonFatal
2020
2121/**
2222 * A Akka-HTTP-based implementation of a ReadinessCheckExecutor.
2323 */
2424private[readiness] class ReadinessCheckExecutorImpl(implicit actorSystem: ActorSystem, materializer: Materializer)
2525 extends ReadinessCheckExecutor {
2626
2727 import mesosphere.marathon.core.async.ExecutionContexts.global
2828 private[this] val log = LoggerFactory.getLogger(getClass)
29 private[this] val defaultSettings = ConnectionPoolSettings(actorSystem)
29 private implicit val scheduler = actorSystem.scheduler
3030
3131 override def execute(readinessCheckSpec: ReadinessCheckSpec): Observable[ReadinessCheckResult] = {
3232 def singleCheck(): Future[ReadinessCheckResult] = executeSingleCheck(readinessCheckSpec)
3333
3434 intervalObservable(readinessCheckSpec.interval)
3535 .scan(singleCheck()) {
3636 case (last, next) =>
3737 // Serialize readiness check execution and repeat last on ready.
Show All 36 Lines
7474 contentType = MediaTypes.`text/plain`.value,
7575 body = s"Marathon could not query ${check.url}: ${e.getMessage}"
7676 )
7777
7878 ReadinessCheckResult.forSpecAndResponse(check, response).copy(ready = false)
7979 }
8080
8181 private[impl] def akkaHttpGet(check: ReadinessCheckSpec): Future[AkkaHttpResponse] = {
82 val connectionSetting = defaultSettings.connectionSettings.withConnectingTimeout(check.timeout)
83 Http().singleRequest(
84 request = RequestBuilding.Get(check.url),
85 settings = defaultSettings.withConnectionSettings(connectionSetting)
86 )
82 Timeout(check.timeout)(Http().singleRequest(
83 request = RequestBuilding.Get(check.url)
84 ))
8785 }
8886}
View Options

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

Show First 20 Lines • Show All 319 Lines • ▼ Show 20 Line(s)
320320
321321 When("The app is deleted")
322322 val delete = marathon.deleteApp(id, force = true)
323323 delete should be(OK)
324324 waitForDeployment(delete)
325325 marathon.listAppsInBaseGroup.value should have size 0
326326 }
327327
328 "an unhealthy app fails to deploy because health checks takes too long to pass" in {
329 Given("a new app that is not healthy")
330 val id = appId()
331 registerAppProxyHealthCheck(id, "v1", state = true).withHealthAction(_ => Thread.sleep(20000))
332 val app = appProxy(id, "v1", instances = 1, healthCheck = Some(appProxyHealthCheck().copy(timeoutSeconds = 2)))
333
334 When("The app is deployed")
335 val create = marathon.createAppV2(app)
336
337 Then("We receive a deployment created confirmation")
338 create should be(Created)
339 extractDeploymentIds(create) should have size 1
340
341 And("a number of failed health events but the deployment does not succeed")
342
343 def interestingEvent() = waitForEventMatching("failed_health_check_event or deployment_success")(callbackEvent =>
344 callbackEvent.eventType == "deployment_success" ||
345 callbackEvent.eventType == "failed_health_check_event"
346 )
347
348 for (event <- Iterator.continually(interestingEvent()).take(10)) {
349 event.eventType should be("failed_health_check_event")
350 }
351
352 When("The app is deleted")
353 val delete = marathon.deleteApp(id, force = true)
354 delete should be(OK)
355 waitForDeployment(delete)
356 marathon.listAppsInBaseGroup.value should have size 0
357 }
358
328359 "update an app" in {
329360 Given("a new app")
330361 val id = appId()
331362 val v1 = appProxy(id, "v1", instances = 1, healthCheck = Some(appProxyHealthCheck()))
332363 val create = marathon.createAppV2(v1)
333364 create should be(Created)
334365 waitForDeployment(create)
335366 val before = marathon.tasks(id)
▲ Show 20 Lines • Show All 431 Lines • Show Last 20 Lines