Fix of http marathon healthchecks (#6115)

Reused some of smart akka-http code related to setting headers and filtering URIs

JIRA issues: MARATHON-8136

(cherry picked from commit 276ce1a)

Author: meln1k

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

@@ -3,12 +3,12 @@ package core.health.impl
 
 import java.net.{ InetSocketAddress, Socket }
 import java.security.cert.X509Certificate
-import javax.net.ssl.{ KeyManager, SSLContext, X509TrustManager }
 
+import javax.net.ssl.{ KeyManager, SSLContext, X509TrustManager }
 import akka.actor.{ Actor, PoisonPill }
 import akka.http.scaladsl.settings.ClientConnectionSettings
 import akka.http.scaladsl.client.RequestBuilding
-import akka.http.scaladsl.model.{ HttpRequest, HttpResponse }
+import akka.http.scaladsl.model.{ HttpRequest, HttpResponse, headers }
 import akka.http.scaladsl.{ ConnectionContext, Http }
 import akka.stream.{ ActorMaterializer, ActorMaterializerSettings, Materializer }
 import akka.stream.scaladsl.{ Sink, Source }
@@ -150,7 +150,7 @@ class HealthCheckWorkerActor(implicit mat: Materializer) extends Actor with Stri
     val url = s"https://$host:$port$absolutePath"
     logger.debug(s"Checking the health of [$url] for instance=${instance.instanceId} via HTTPS")
 
-    singleRequest(
+    singleRequestHttps(
       RequestBuilding.Get(url),
       check.timeout
     ).map { response =>
@@ -169,34 +169,45 @@ class HealthCheckWorkerActor(implicit mat: Materializer) extends Actor with Stri
   }
 
   def singleRequest(httpRequest: HttpRequest, timeout: FiniteDuration)(implicit mat: Materializer): Future[HttpResponse] = {
-    if (httpRequest.uri.scheme.equalsIgnoreCase("https")) {
-      // This is only a health check, so we are going to allow _very_ bad SSL configuration.
-      val disabledSslConfig = AkkaSSLConfig().mapSettings(s => s.withLoose {
-        s.loose.withAcceptAnyCertificate(true)
-          .withAllowLegacyHelloMessages(Some(true))
-          .withAllowUnsafeRenegotiation(Some(true))
-          .withAllowWeakCiphers(true)
-          .withAllowWeakProtocols(true)
-          .withDisableHostnameVerification(true)
-          .withDisableSNI(true)
-      })
-      val authority = httpRequest.uri.authority
-      val connectionFlow = Http().outgoingConnectionHttps(
-        authority.host.toString(),
-        authority.port,
-        ConnectionContext.https(disabledSslContext, sslConfig = Some(disabledSslConfig)),
-        settings = ClientConnectionSettings(system).withIdleTimeout(timeout)
-      )
-      Source.single(httpRequest).via(connectionFlow).runWith(Sink.head)
-    } else {
-      val authority = httpRequest.uri.authority
-      val connectionFlow = Http().outgoingConnection(
-        authority.host.toString(),
-        authority.port,
-        settings = ClientConnectionSettings(system).withIdleTimeout(timeout)
-      )
-      Source.single(httpRequest).via(connectionFlow).runWith(Sink.head)
-    }
+    val host = httpRequest.uri.authority.host.toString()
+    val port = httpRequest.uri.effectivePort
+    val hostHeader = headers.Host(host, port)
+    val effectiveRequest = httpRequest
+      .withUri(httpRequest.uri.toHttpRequestTargetOriginForm)
+      .withDefaultHeaders(hostHeader)
+
+    val connectionFlow = Http().outgoingConnection(
+      host,
+      port,
+      settings = ClientConnectionSettings(system).withIdleTimeout(timeout)
+    )
+    Source.single(effectiveRequest).via(connectionFlow).runWith(Sink.head)
+  }
+
+  def singleRequestHttps(httpRequest: HttpRequest, timeout: FiniteDuration)(implicit mat: Materializer): Future[HttpResponse] = {
+    val host = httpRequest.uri.authority.host.toString()
+    val port = httpRequest.uri.effectivePort
+    val hostHeader = headers.Host(host, port)
+    val effectiveRequest = httpRequest
+      .withUri(httpRequest.uri.toHttpRequestTargetOriginForm)
+      .withDefaultHeaders(hostHeader)
+    // This is only a health check, so we are going to allow _very_ bad SSL configuration.
+    val disabledSslConfig = AkkaSSLConfig().mapSettings(s => s.withLoose {
+      s.loose.withAcceptAnyCertificate(true)
+        .withAllowLegacyHelloMessages(Some(true))
+        .withAllowUnsafeRenegotiation(Some(true))
+        .withAllowWeakCiphers(true)
+        .withAllowWeakProtocols(true)
+        .withDisableHostnameVerification(true)
+        .withDisableSNI(true)
+    })
+    val connectionFlow = Http().outgoingConnectionHttps(
+      host,
+      port,
+      ConnectionContext.https(disabledSslContext, sslConfig = Some(disabledSslConfig)),
+      settings = ClientConnectionSettings(system).withIdleTimeout(timeout)
+    )
+    Source.single(effectiveRequest).via(connectionFlow).runWith(Sink.head)
   }
 }
 

src/test/scala/mesosphere/marathon/core/health/impl/HealthCheckWorkerActorTest.scala

@@ -100,7 +100,12 @@ class HealthCheckWorkerActorTest extends AkkaUnitTest with ImplicitSender {
             promise.success("success")
             complete(StatusCodes.OK)
           }
-        }
+        } ~
+          path("unhealthy"){
+            get {
+              complete(StatusCodes.InternalServerError)
+            }
+          }
 
       val binding = Http().bindAndHandle(route, "localhost", 0).futureValue
 
@@ -124,9 +129,14 @@ class HealthCheckWorkerActorTest extends AkkaUnitTest with ImplicitSender {
 
       val ref = system.actorOf(Props(classOf[HealthCheckWorkerActor], mat))
       ref ! HealthCheckJob(app, instance, MarathonHttpHealthCheck(port = Some(port), path = Some("/health")))
+      expectMsgClass(classOf[Healthy])
 
       promise.future.futureValue shouldEqual "success"
 
+      val unhealthy = system.actorOf(Props(classOf[HealthCheckWorkerActor], mat))
+      unhealthy ! HealthCheckJob(app, instance, MarathonHttpHealthCheck(port = Some(port), path = Some("/unhealthy")))
+      expectMsgClass(classOf[Unhealthy])
+
     }
   }
 }