pekko: support mTLS in PekkoHttpClient

Add a create overload that accepts an HttpsConnectionContext so the
client presents a client certificate for mutual TLS. The context is
threaded through singleRequest and used as the default for superPool,
with a per-call ClientConfig.connectionContext still taking precedence.
Existing callers are unaffected (default is None).

Author: brharrington

atlas-pekko/src/main/scala/com/netflix/atlas/pekko/PekkoHttpClient.scala

@@ -89,6 +89,35 @@ object PekkoHttpClient {
     new HttpClientImpl(name)(system)
   }
 
+  /**
+    * Create a new client instance that uses the provided connection context by default for its
+    * requests. A per-call `ClientConfig.connectionContext` passed to `superPool` still takes
+    * precedence over this default. This can be used to make mutual TLS (mTLS) calls. The context
+    * is typically built from an `SSLContext` configured with the client certificate and trust
+    * store, e.g.:
+    *
+    * {{{
+    * val connectionContext = ConnectionContext.httpsClient(sslContext)
+    * PekkoHttpClient.create("myapp", system, connectionContext)
+    * }}}
+    *
+    * @param name
+    *     Name to use for access logging and metrics.
+    * @param system
+    *     Actor system to use for processing the requests.
+    * @param connectionContext
+    *     Connection context used to establish the client connections.
+    * @return
+    *     New client instance.
+    */
+  def create(
+    name: String,
+    system: ActorSystem,
+    connectionContext: HttpsConnectionContext
+  ): PekkoHttpClient = {
+    new HttpClientImpl(name, Some(connectionContext))(system)
+  }
+
   /**
     * Create a client instance that will return a fixed response for every request. Mainly
     * used for testing.
@@ -153,14 +182,18 @@ object PekkoHttpClient {
   private def isConnectException(t: Throwable): Boolean = t.isInstanceOf[ConnectException]
 
   /** Default implementation based on Pekko `Http()`. */
-  private[pekko] class HttpClientImpl(name: String)(implicit val system: ActorSystem)
+  private[pekko] class HttpClientImpl(
+    name: String,
+    defaultConnectionContext: Option[HttpsConnectionContext] = None
+  )(implicit val system: ActorSystem)
       extends PekkoHttpClient {
 
     private implicit val ec: ExecutionContext = system.dispatcher
     private val http = Http()
 
     protected def doSingleRequest(request: HttpRequest): Future[HttpResponse] = {
-      http.singleRequest(request)
+      val connectionContext = defaultConnectionContext.getOrElse(http.defaultClientHttpsContext)
+      http.singleRequest(request, connectionContext)
     }
 
     override def singleRequest(request: HttpRequest): Future[HttpResponse] = {
@@ -178,7 +211,9 @@ object PekkoHttpClient {
     override def superPool[C](
       config: ClientConfig
     ): Flow[(HttpRequest, C), (Try[HttpResponse], C), NotUsed] = {
-      val connectionContext = config.connectionContext.getOrElse(http.defaultClientHttpsContext)
+      val connectionContext = config.connectionContext
+        .orElse(defaultConnectionContext)
+        .getOrElse(http.defaultClientHttpsContext)
       val settings = config.settings.getOrElse(ConnectionPoolSettings(system))
 
       // All retries will be handled in this flow, disable in the pekko pool

atlas-pekko/src/test/scala/com/netflix/atlas/pekko/PekkoHttpClientSuite.scala

@@ -24,6 +24,7 @@ import com.netflix.spectator.ipc.IpcStatus
 import munit.FunSuite
 import org.apache.pekko.NotUsed
 import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.http.scaladsl.ConnectionContext
 import org.apache.pekko.http.scaladsl.HttpsConnectionContext
 import org.apache.pekko.http.scaladsl.model.HttpMethods
 import org.apache.pekko.http.scaladsl.model.HttpRequest
@@ -38,6 +39,7 @@ import org.apache.pekko.stream.scaladsl.Source
 import java.util.concurrent.ArrayBlockingQueue
 import java.util.concurrent.TimeUnit
 import java.util.stream.Collectors
+import javax.net.ssl.SSLContext
 import scala.concurrent.Await
 import scala.concurrent.Future
 import scala.concurrent.duration.Duration
@@ -119,7 +121,14 @@ class PekkoHttpClientSuite extends FunSuite {
       .withMaxRetries(3)
       .withBaseConnectionBackoff(FiniteDuration(0, TimeUnit.MILLISECONDS))
       .withMaxConnectionBackoff(FiniteDuration(0, TimeUnit.MILLISECONDS))
-    val config = ClientConfig(settings = Some(settings))
+    flowRequest(client, request, ClientConfig(settings = Some(settings)))
+  }
+
+  private def flowRequest(
+    client: PekkoHttpClient,
+    request: HttpRequest,
+    config: ClientConfig
+  ): Try[HttpResponse] = {
     val future = Source
       .single(request)
       .map(r => r -> NotUsed)
@@ -165,6 +174,22 @@ class PekkoHttpClientSuite extends FunSuite {
     }
   }
 
+  test("superPool: uses default connection context from create") {
+    val ctx = ConnectionContext.httpsClient(SSLContext.getDefault)
+    val client = new ContextCapturingClient(system, Some(ctx))
+    flowRequest(client, HttpRequest(HttpMethods.GET, Uri("/test")))
+    assertEquals(client.lastContext, ctx)
+  }
+
+  test("superPool: config connection context overrides default") {
+    val defaultCtx = ConnectionContext.httpsClient(SSLContext.getDefault)
+    val overrideCtx = ConnectionContext.httpsClient(SSLContext.getDefault)
+    val client = new ContextCapturingClient(system, Some(defaultCtx))
+    val config = ClientConfig(connectionContext = Some(overrideCtx))
+    flowRequest(client, HttpRequest(HttpMethods.GET, Uri("/test")), config)
+    assertEquals(client.lastContext, overrideCtx)
+  }
+
   test("superPool: ipc metrics on throttled") {
     val responses = List(
       Success(HttpResponse(StatusCodes.TooManyRequests)),
@@ -185,6 +210,25 @@ class PekkoHttpClientSuite extends FunSuite {
 
 object PekkoHttpClientSuite {
 
+  /** Records the connection context passed to the super pool so it can be asserted on. */
+  class ContextCapturingClient(
+    system: ActorSystem,
+    defaultContext: Option[HttpsConnectionContext]
+  ) extends PekkoHttpClient.HttpClientImpl("test", defaultContext)(system) {
+
+    @volatile var lastContext: HttpsConnectionContext = _
+
+    override protected def superPoolFlow[C](
+      connectionContext: HttpsConnectionContext,
+      settings: ConnectionPoolSettings
+    ): Flow[(HttpRequest, C), (Try[HttpResponse], C), NotUsed] = {
+      lastContext = connectionContext
+      Flow[(HttpRequest, C)].map {
+        case (_, context) => Success(HttpResponse(StatusCodes.OK)) -> context
+      }
+    }
+  }
+
   class TestHttpClient(system: ActorSystem, responses: List[Try[HttpResponse]])
       extends PekkoHttpClient.HttpClientImpl("test")(system) {