Merge pull request #929 from kaizo-com/deflate-compression

Datadog 'Deflate' compression

Author: SimunKaracic

reporters/kamon-datadog/src/main/resources/reference.conf

@@ -62,6 +62,9 @@ kamon {
       connect-timeout = 5 seconds
       read-timeout = 5 seconds
       write-timeout = 5 seconds
+
+      # Use 'Deflate' compression when posting to the Datadog API
+      compression = false
     }
 
 
@@ -111,5 +114,3 @@ kamon {
     }
   }
 }
-
-

reporters/kamon-datadog/src/main/scala/kamon/datadog/DeflateInterceptor.scala

@@ -0,0 +1,40 @@
+package kamon.datadog
+
+import okhttp3.Interceptor.Chain
+import okhttp3.{Interceptor, MediaType, RequestBody, Response}
+import okio.{BufferedSink, DeflaterSink, Okio}
+
+import java.util.zip.Deflater
+
+class DeflateInterceptor extends Interceptor {
+
+  def intercept(chain: Chain): Response = {
+    val originalRequest = chain.request
+
+    Option(originalRequest.body) match {
+      case Some(body) if Option(originalRequest.header("Content-Encoding")).isEmpty =>
+        val compressedRequest = originalRequest
+          .newBuilder
+          .header("Content-Encoding", "deflate")
+          .method(
+            originalRequest.method,
+            encodeDeflate(body)
+          )
+          .build
+
+        chain.proceed(compressedRequest)
+      case _ =>
+        chain.proceed(originalRequest)
+    }
+  }
+
+  private def encodeDeflate(body: RequestBody): RequestBody = new RequestBody() {
+    def contentType: MediaType = body.contentType
+
+    def writeTo(sink: BufferedSink): Unit = {
+      val deflateSink = Okio.buffer(new DeflaterSink(sink, new Deflater()))
+      body.writeTo(deflateSink)
+      deflateSink.close
+    }
+  }
+}

reporters/kamon-datadog/src/main/scala/kamon/datadog/package.scala

@@ -36,7 +36,7 @@ package object datadog {
     }
   }
 
-  private[datadog] case class HttpClient(apiUrl: String, apiKey: Option[String], usingAgent: Boolean, connectTimeout: Duration,
+  private[datadog] case class HttpClient(apiUrl: String, apiKey: Option[String], usingCompression: Boolean, usingAgent: Boolean, connectTimeout: Duration,
                                          readTimeout: Duration, writeTimeout: Duration) {
 
     val httpClient: OkHttpClient = createHttpClient()
@@ -45,6 +45,7 @@ package object datadog {
       this(
         config.getString("api-url"),
         if (usingAgent) None else Some(config.getString("api-key")),
+        if (usingAgent) false else config.getBoolean("compression"),
         usingAgent,
         config.getDuration("connect-timeout"),
         config.getDuration("read-timeout"),
@@ -97,12 +98,14 @@ package object datadog {
 
     // Apparently okhttp doesn't require explicit closing of the connection
     private def createHttpClient(): OkHttpClient = {
-      new OkHttpClient.Builder()
+      val builder = new OkHttpClient.Builder()
         .connectTimeout(connectTimeout.toMillis, TimeUnit.MILLISECONDS)
         .readTimeout(readTimeout.toMillis, TimeUnit.MILLISECONDS)
         .writeTimeout(writeTimeout.toMillis, TimeUnit.MILLISECONDS)
         .retryOnConnectionFailure(false)
-        .build()
+
+      if (usingCompression) builder.addInterceptor(new DeflateInterceptor).build()
+      else builder.build()
     }
   }
 

reporters/kamon-datadog/src/test/resources/application.conf

@@ -18,5 +18,8 @@ kamon {
       include-instance = "no"
       include-host = "no"
     }
+    api {
+      compression = false
+    }
   }
 }

reporters/kamon-datadog/src/test/scala/kamon/datadog/DatadogAPIReporterSpec.scala

@@ -1,17 +1,18 @@
 package kamon.datadog
 
 import java.time.Instant
-
 import kamon.Kamon
 import kamon.metric.Distribution.Percentile
 import kamon.metric._
 import kamon.module.ModuleFactory
 import kamon.tag.TagSet
 import kamon.testkit.Reconfigure
 import okhttp3.mockwebserver.MockResponse
+import okio.{Buffer, InflaterSource, Okio}
 import org.scalatest.Matchers
 import play.api.libs.json.Json
 
+import java.util.zip.Inflater
 import scala.concurrent.ExecutionContext
 
 class DatadogAPIReporterSpec extends AbstractHttpReporter with Matchers with Reconfigure {
@@ -20,10 +21,45 @@ class DatadogAPIReporterSpec extends AbstractHttpReporter with Matchers with Rec
     val reporter = new DatadogAPIReporterFactory().create(ModuleFactory.Settings(Kamon.config(), ExecutionContext.global))
     val now = Instant.ofEpochMilli(1523395554)
 
+    "sends metrics - compressed" in {
+      val baseUrl = mockResponse("/test", new MockResponse().setStatus("HTTP/1.1 200 OK"))
+      applyConfig("kamon.datadog.api.api-url = \"" + baseUrl + "\"")
+      applyConfig("kamon.datadog.api.api-key = \"dummy\"")
+      applyConfig("kamon.datadog.api.compression = true")
+      reporter.reconfigure(Kamon.config())
+
+      reporter.reportPeriodSnapshot(
+        PeriodSnapshot.apply(
+          now.minusMillis(1000),
+          now,
+          MetricSnapshot.ofValues[Long](
+            "test.counter",
+            "test",
+            Metric.Settings.ForValueInstrument(MeasurementUnit.none, java.time.Duration.ZERO),
+            Instrument.Snapshot.apply(TagSet.of("tag1", "value1"), 0L) :: Nil
+          ) :: Nil,
+          Nil,
+          Nil,
+          Nil,
+          Nil
+        )
+      )
+
+      val request = server.takeRequest()
+
+      val decompressedBody = Okio.buffer(new InflaterSource(request.getBody.buffer(), new Inflater())).readByteString().utf8()
+
+      Json.parse(decompressedBody) shouldEqual Json
+        .parse(
+          """{"series":[{"metric":"test.counter","interval":1,"points":[[1523394,0]],"type":"count","host":"test","tags":["env:staging","service:kamon-application","tag1:value1"]}]}"""
+        )
+    }
+
     "sends counter metrics" in {
       val baseUrl = mockResponse("/test", new MockResponse().setStatus("HTTP/1.1 200 OK"))
       applyConfig("kamon.datadog.api.api-url = \"" + baseUrl + "\"")
       applyConfig("kamon.datadog.api.api-key = \"dummy\"")
+      applyConfig("kamon.datadog.api.compression = false")
       reporter.reconfigure(Kamon.config())
 
       reporter.reportPeriodSnapshot(
@@ -56,6 +92,7 @@ class DatadogAPIReporterSpec extends AbstractHttpReporter with Matchers with Rec
       val baseUrl = mockResponse("/test", new MockResponse().setStatus("HTTP/1.1 200 OK"))
       applyConfig("kamon.datadog.api.api-url = \"" + baseUrl + "\"")
       applyConfig("kamon.datadog.api.api-key = \"dummy\"")
+      applyConfig("kamon.datadog.api.compression = false")
       reporter.reconfigure(Kamon.config())
 
       val distribution = new Distribution {