Configuration option to enable gzip compression on k8s pods api for s…

…ervice discovery (#336)

* added configuration option to enable gzip compression on k8s pods api for service discovery

* allow various encodings to be configured for k8s discovery API call; address review findings

* adjust toString of settings to include httpRequestAcceptEncoding key; adjust "kubernetes-api-java" integration-test to use gzip compression

discovery-kubernetes-api/src/main/resources/reference.conf

@@ -39,5 +39,11 @@ pekko.discovery {
 
     # When set, validate the container is not in 'waiting' state
     container-name = ""
+
+    # Enables querying the Kubernetes HTTP API with 'Accept-Encoding: <http-request-accept-encoding>' header, e.g. in order to compress response payload
+    #
+    # The empty string "" will result in no encoding being requested. E.g. use "gzip" in order to use gzip compression.
+    # Supports all encodings of `org.apache.pekko.http.scaladsl.model.headers.HttpEncodings`.
+    http-request-accept-encoding = ""
   }
 }

discovery-kubernetes-api/src/main/scala/org/apache/pekko/discovery/kubernetes/KubernetesApiServiceDiscovery.scala

@@ -24,6 +24,8 @@ import scala.util.Try
 import scala.util.control.{ NoStackTrace, NonFatal }
 
 import org.apache.pekko
+import org.apache.pekko.http.javadsl.model.headers.AcceptEncoding
+import org.apache.pekko.http.scaladsl.coding.Coders
 import pekko.actor.ActorSystem
 import pekko.annotation.InternalApi
 import pekko.discovery.ServiceDiscovery.{ Resolved, ResolvedTarget }
@@ -34,7 +36,7 @@ import pekko.event.{ LogSource, Logging }
 import pekko.http.scaladsl.HttpsConnectionContext
 import pekko.http.scaladsl._
 import pekko.http.scaladsl.model._
-import pekko.http.scaladsl.model.headers.{ Authorization, OAuth2BearerToken }
+import pekko.http.scaladsl.model.headers.{ Authorization, HttpEncodings, OAuth2BearerToken }
 import pekko.http.scaladsl.unmarshalling.Unmarshal
 import pekko.pki.kubernetes.PemManagersProvider
 
@@ -141,7 +143,7 @@ class KubernetesApiServiceDiscovery(settings: Settings)(
         podRequest(apiToken, podNamespace, labelSelector),
         s"Unable to form request; check Kubernetes environment (expecting env vars ${settings.apiServiceHostEnvName}, ${settings.apiServicePortEnvName})")
 
-      response <- http.singleRequest(request, clientSslContext)
+      response <- http.singleRequest(request, clientSslContext).map(decodeResponse)
 
       entity <- response.entity.toStrict(resolveTimeout)
 
@@ -236,6 +238,21 @@ class KubernetesApiServiceDiscovery(settings: Settings)(
       val query = Uri.Query("labelSelector" -> labelSelector)
       val uri = Uri.from(scheme = "https", host = host, port = port).withPath(path).withQuery(query)
 
-      HttpRequest(uri = uri, headers = Seq(Authorization(OAuth2BearerToken(token))))
+      val authHeaders = Seq(Authorization(OAuth2BearerToken(token)))
+      val acceptEncodingHeader = HttpEncodings.getForKey(settings.httpRequestAcceptEncoding)
+        .map(httpEncoding => AcceptEncoding.create(httpEncoding))
+      HttpRequest(uri = uri, headers = authHeaders ++ acceptEncodingHeader)
     }
+
+  private def decodeResponse(response: HttpResponse): HttpResponse = {
+    val decoder = response.encoding match {
+      case HttpEncodings.gzip =>
+        Coders.Gzip
+      case HttpEncodings.deflate =>
+        Coders.Deflate
+      case _ =>
+        Coders.NoCoding
+    }
+    decoder.decodeMessage(response)
+  }
 }

discovery-kubernetes-api/src/main/scala/org/apache/pekko/discovery/kubernetes/Settings.scala

@@ -69,9 +69,11 @@ final class Settings(kubernetesApi: Config) extends Extension {
 
   val containerName: Option[String] = Some(kubernetesApi.getString("container-name")).filter(_.nonEmpty)
 
+  val httpRequestAcceptEncoding: String = kubernetesApi.getString("http-request-accept-encoding")
+
   override def toString =
     s"Settings($apiCaPath, $apiTokenPath, $apiServiceHostEnvName, $apiServicePortEnvName, " +
-    s"$podNamespacePath, $podNamespace, $podDomain)"
+    s"$podNamespacePath, $podNamespace, $podDomain, httpRequestAcceptEncoding=$httpRequestAcceptEncoding)"
 }
 
 object Settings extends ExtensionId[Settings] with ExtensionIdProvider {

integration-test/kubernetes-api-java/src/main/resources/application.conf

@@ -12,6 +12,8 @@ pekko.discovery {
   kubernetes-api {
     # in fact, this is already the default:
     pod-label-selector = "app=%s"
+    # ask for gzip compression as "Accept-Encoding" during the test:
+    http-request-accept-encoding = "gzip"
   }
 }