Merge pull request #915 from saeltz/gzip-metrics

gzip compression

reporters/kamon-prometheus/src/main/scala/kamon/prometheus/embeddedhttp/SunEmbeddedHttpServer.scala

@@ -16,12 +16,18 @@
 
 package kamon.prometheus.embeddedhttp
 
-import java.net.{InetAddress, InetSocketAddress}
-import java.nio.charset.StandardCharsets
-
 import com.sun.net.httpserver.{HttpExchange, HttpHandler, HttpServer}
 import com.typesafe.config.Config
 import kamon.prometheus.ScrapeSource
+import kamon.prometheus.embeddedhttp.SunEmbeddedHttpServer.shouldUseCompression
+
+import java.io.OutputStream
+import java.net.{InetAddress, InetSocketAddress}
+import java.nio.charset.StandardCharsets
+import java.util
+import java.util.zip.GZIPOutputStream
+import scala.collection.JavaConverters._
+import scala.collection.mutable
 
 class SunEmbeddedHttpServer(hostname: String, port: Int, scrapeSource: ScrapeSource, config: Config) extends EmbeddedHttpServer(hostname, port, scrapeSource, config) {
   private val server = {
@@ -32,24 +38,42 @@ class SunEmbeddedHttpServer(hostname: String, port: Int, scrapeSource: ScrapeSou
         if (httpExchange.getRequestURI.getPath == "/metrics") {
           val data = scrapeSource.scrapeData()
           val bytes = data.getBytes(StandardCharsets.UTF_8)
-          httpExchange.sendResponseHeaders(200, bytes.length)
-          val os = httpExchange.getResponseBody
+          var os: OutputStream = null
           try {
-            os.write(bytes)
-          }
-          finally
-            os.close()
-        }
-        else {
-          httpExchange.sendResponseHeaders(404, -1)
-        }
+            if (shouldUseCompression(httpExchange)) {
+              httpExchange.getResponseHeaders.set("Content-Encoding", "gzip")
+              httpExchange.sendResponseHeaders(200, 0)
+              os = new GZIPOutputStream(httpExchange.getResponseBody)
+              os.write(bytes)
+            } else {
+              os = httpExchange.getResponseBody
+              httpExchange.sendResponseHeaders(200, bytes.length)
+              os.write(bytes)
+            }
+          } finally Option(os).map(_.close())
+        } else httpExchange.sendResponseHeaders(404, -1)
       }
-
     }
+
     s.createContext("/metrics", handler)
     s.start()
     s
   }
 
   def stop(): Unit = server.stop(0)
 }
+
+object SunEmbeddedHttpServer {
+  def shouldUseCompression(httpExchange: HttpExchange): Boolean = {
+    httpExchange.getRequestHeaders
+      .asScala.get("Accept-Encoding")
+      .map(extractEncodings)
+      .exists(_.contains("gzip"))
+  }
+
+  private def extractEncodings(headerList: util.List[String]): mutable.Buffer[String] = {
+    headerList.asScala
+      .flatMap(_.split(","))
+      .map(_.trim().toLowerCase())
+  }
+}

reporters/kamon-prometheus/src/test/scala/kamon/prometheus/EmbeddedHttpServerSpec.scala

@@ -2,13 +2,15 @@ package kamon.prometheus
 
 import java.io.FileNotFoundException
 import java.net.URL
-
 import com.typesafe.config.{Config, ConfigFactory}
 import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpec}
 
+import java.util.zip.GZIPInputStream
+
 class SunHttpServerSpecSuite extends EmbeddedHttpServerSpecSuite {
   override def testConfig: Config = ConfigFactory.load()
 }
+
 //class NanoHttpServerSpecSuite extends EmbeddedHttpServerSpecSuite {
 //  override def testConfig: Config = ConfigFactory.parseString(
 //    """
@@ -22,6 +24,7 @@ class SunHttpServerSpecSuite extends EmbeddedHttpServerSpecSuite {
 
 abstract class EmbeddedHttpServerSpecSuite extends WordSpec with Matchers with BeforeAndAfterAll with KamonTestSnapshotSupport {
   protected def testConfig: Config
+
   protected def port: Int = 9095
 
   private var testee: PrometheusReporter = _
@@ -74,13 +77,33 @@ abstract class EmbeddedHttpServerSpecSuite extends WordSpec with Matchers with B
         httpGetMetrics("/metricsss")
       }
     }
+
+    "respect gzip Content-Encoding headers" in {
+      //arrange
+      testee.reportPeriodSnapshot(counter("jvm.mem"))
+      //act
+      val metrics = httpGetMetrics("/metrics")
+      val gzippedMetrics = httpGetGzippedMetrics("/metrics")
+      //assert
+      metrics.length should be > gzippedMetrics.length
+    }
   }
 
   private def httpGetMetrics(endpoint: String): String = {
-    val src = scala.io.Source.fromURL(new URL(s"http://127.0.0.1:$port$endpoint"))
-    try
-      src.mkString
-    finally
-      src.close()
+    val url = new URL(s"http://127.0.0.1:$port$endpoint")
+    val src = scala.io.Source.fromURL(url)
+    try src.mkString
+    finally src.close()
+  }
+
+  private def httpGetGzippedMetrics(endpoint: String): String = {
+    val url = new URL(s"http://127.0.0.1:$port$endpoint")
+    val connection = url.openConnection
+    connection.setRequestProperty("Accept-Encoding", "gzip")
+    val gzipStream = new GZIPInputStream(connection.getInputStream)
+    val src = scala.io.Source.fromInputStream(gzipStream)
+    connection.getRequestProperty("Accept-Encoding") shouldBe "gzip"
+    try src.getLines.mkString
+    finally gzipStream.close()
   }
 }