Merge remote-tracking branch 'cobli/pr-add-span-reporter'

Author: ivantopo

.travis.yml

@@ -1,12 +1,23 @@
+# Use container-based infrastructure
+sudo: false
+
 language: scala
-script:
-  - ./travis-test.sh
+
+jdk: oraclejdk8
+
 scala:
-   - 2.11.8
-jdk:
-   - oraclejdk8
-before_script:
-  - mkdir $TRAVIS_BUILD_DIR/tmp
-  - export SBT_OPTS="-Djava.io.tmpdir=$TRAVIS_BUILD_DIR/tmp"
-sudo: false
-  
+  - 2.12.8
+
+# These directories are cached to S3 at the end of the build
+cache:
+  directories:
+    - $HOME/.ivy2/cache
+    - $HOME/.sbt/boot/
+
+before_cache:
+  # Cleanup the cached directories to avoid unnecessary cache updates
+  - rm -fv $HOME/.ivy2/.sbt.ivy.lock
+  - find $HOME/.ivy2/cache -name "ivydata-*.properties" -print -delete
+  - find $HOME/.sbt        -name "*.lock"               -print -delete
+
+

README.md

@@ -32,13 +32,19 @@ Kamon.addReporter(new DatadogAgentReporter())
 Kamon.addReporter(new DatadogAPIReporter())
 ```
 
+In order to get [APM tracing](https://docs.datadoghq.com/tracing/) working, add
+
+```
+Kamon.addReporter(new DatadogSpanReporter())
+```
+
 Configuration
 -------------
 
 #### Agent Reporter
 
 By default, the Agent reporter assumes that you have an instance of the Datadog Agent running in localhost and listening on
-port 8125. If that is not the case the you can use the `kamon.datadog.agent.hostname` and `kamon.datadog.agent.port` configuration
+port 8125. If that is not the case, you can use the `kamon.datadog.agent.hostname` and `kamon.datadog.agent.port` configuration
 keys to point the module at your Datadog Agent installation.
 
 #### API Reporter
@@ -56,6 +62,15 @@ follows:
 
 You can refer to the [Datadog documentation](https://docs.datadoghq.com/developers/metrics/#histograms) for more details.
 
+
+#### Span Reporter
+As [recommended](https://docs.datadoghq.com/api/?lang=python#tracing) , the Span reporter assumes that you have an instance of the Datadog Agent running in localhost and listening on
+port 8125. If that is not the case, you can use the `kamon.datadog.trace.http.api-url` configuration
+key to point the module at your Datadog Agent installation.
+
+Don't forget to configure [actors filtering for message tracing](https://kamon.io/docs/latest/instrumentation/akka/tracing/) 
+
+
 ### Metric Units ###
 
 Kamon keeps all timing measurements in nanoseconds and memory measurements in bytes. In order to scale those to other

build.sbt

@@ -13,29 +13,42 @@
  * =========================================================================================
  */
 
-import scalariform.formatter.preferences._
 import com.typesafe.sbt.SbtScalariform.ScalariformKeys
+import scalariform.formatter.preferences._
+
 
-val kamonCore         = "io.kamon"               %% "kamon-core"          % "1.1.6"
-val kamonTestKit      = "io.kamon"               %% "kamon-testkit"       % "1.1.6"
-val asyncHttpClient   = "com.squareup.okhttp3"    % "okhttp"              % "3.10.0"
+val kamonCore             = "io.kamon"               %% "kamon-core"          % "1.1.6"
+val kamonTestKit          = "io.kamon"               %% "kamon-testkit"       % "1.1.6"
+val asyncHttpClient       = "com.squareup.okhttp3"    % "okhttp"              % "3.10.0"
+val asyncHttpClientMock   = "com.squareup.okhttp3"    % "mockwebserver"       % "3.10.0"
 
 lazy val root = (project in file("."))
   .settings(name := "kamon-datadog")
   .settings(
     libraryDependencies ++=
-      compileScope(kamonCore, asyncHttpClient, scalaCompact.value) ++ 
-      testScope(scalatest, slf4jApi, slf4jnop, kamonCore, kamonTestKit),
+      compileScope(kamonCore, asyncHttpClient, scalaCompact.value, playJsonVersion.value) ++
+        testScope(scalatest, slf4jApi, slf4jnop, kamonCore, kamonTestKit, asyncHttpClientMock),
     ScalariformKeys.preferences := formatSettings(ScalariformKeys.preferences.value))
 
 
+def playJsonVersion = Def.setting {
+  scalaBinaryVersion.value match {
+    case "2.10"          => "com.typesafe.play"      %% "play-json"          % "2.4.11"
+    case "2.12" | "2.11" => "com.typesafe.play"      %% "play-json"          % "2.6.9"
+  }
+}
+
+
 def scalaCompact = Def.setting {
   scalaBinaryVersion.value match {
     case "2.10" | "2.11" => "org.scala-lang.modules" %% "scala-java8-compat" % "0.5.0"
     case "2.12"          => "org.scala-lang.modules" %% "scala-java8-compat" % "0.8.0"
   }
 }
 
+/* Changing Kamon configuration in real-time seems to turn tests unstable */ 
+parallelExecution in Test := false
+
 def formatSettings(prefs: IFormattingPreferences) = prefs
   .setPreference(AlignParameters, true)
   .setPreference(AlignSingleLineCaseStatements, true)

project/build.properties

@@ -1 +1 @@
-sbt.version=1.2.8
+sbt.version=1.2.8

src/main/resources/reference.conf

@@ -28,18 +28,39 @@ kamon {
     #
     http {
 
-      # Datadog API key to use to send metrics to datadog directly over HTTPS. 
+      api-url = "https://app.datadoghq.com/api/v1/series"
+
+
+      # Datadog API key to use to send metrics to datadog directly over HTTPS.
       # If this is not set, metrics are sent as statsd packets over UDP to dogstatsd.
       api-key = ""
 
-      # The Datadog api endpoint to send the data to
-      api-url = "https://app.datadoghq.com/api/v1/series?api_key="
+      using-agent = false
 
       connect-timeout = 5 seconds
       read-timeout = 5 seconds
       request-timeout = 5 seconds
     }
 
+
+    #
+    # Settings relevant to the DatadogSpanReporter
+    #
+    trace.http {
+
+      # Default to agent URL (https://docs.datadoghq.com/api/?lang=python#tracing)
+      api-url = "http://localhost:8126/v0.3/traces"
+
+      api-key = ${kamon.datadog.http.api-key}
+
+      using-agent = true
+
+      connect-timeout = ${kamon.datadog.http.connect-timeout}
+      read-timeout = ${kamon.datadog.http.read-timeout}
+      request-timeout = ${kamon.datadog.http.request-timeout}
+    }
+
+
     # All time values are collected in nanoseconds,
     # to scale before sending to datadog set "time-units" to "s" or "ms" or "µs".
     # Value "n" is equivalent to omitting the setting

src/main/scala/kamon/datadog/DatadogAPIReporter.scala

@@ -17,32 +17,30 @@
 package kamon.datadog
 
 import java.lang.StringBuilder
+import java.nio.charset.StandardCharsets
 import java.text.{ DecimalFormat, DecimalFormatSymbols }
 import java.time.Duration
 import java.util.Locale
-import java.util.concurrent.TimeUnit
 
 import com.typesafe.config.Config
-import kamon.metric._
-import kamon.metric.MeasurementUnit
 import kamon.metric.MeasurementUnit.Dimension.{ Information, Time }
-import kamon.metric.MeasurementUnit.{ information, time }
+import kamon.metric.{ MeasurementUnit, MetricDistribution, MetricValue, PeriodSnapshot }
 import kamon.util.{ EnvironmentTagBuilder, Matcher }
 import kamon.{ Kamon, MetricReporter }
-import okhttp3.{ MediaType, OkHttpClient, Request, RequestBody }
 import org.slf4j.LoggerFactory
 
+import scala.util.{ Failure, Success }
+
 class DatadogAPIReporter extends MetricReporter {
   import DatadogAPIReporter._
 
   private val logger = LoggerFactory.getLogger(classOf[DatadogAPIReporter])
   private val symbols = DecimalFormatSymbols.getInstance(Locale.US)
   symbols.setDecimalSeparator('.') // Just in case there is some weird locale config we are not aware of.
 
-  private val jsonType = MediaType.parse("application/json; charset=utf-8")
   private val valueFormat = new DecimalFormat("#0.#########", symbols)
   private var configuration = readConfiguration(Kamon.config())
-  private var httpClient: OkHttpClient = createHttpClient(configuration)
+  private var httpClient: HttpClient = new HttpClient(configuration.httpConfig)
 
   override def start(): Unit = {
     logger.info("Started the Datadog API reporter.")
@@ -54,24 +52,20 @@ class DatadogAPIReporter extends MetricReporter {
 
   override def reconfigure(config: Config): Unit = {
     val newConfiguration = readConfiguration(config)
-    httpClient = createHttpClient(readConfiguration(Kamon.config()))
     configuration = newConfiguration
+    httpClient = new HttpClient(configuration.httpConfig)
   }
 
   override def reportPeriodSnapshot(snapshot: PeriodSnapshot): Unit = {
-    val url = configuration.apiUrl + configuration.apiKey
-    val body = RequestBody.create(jsonType, buildRequestBody(snapshot))
-    val request = new Request.Builder().url(url).post(body).build
-    val response = httpClient.newCall(request).execute()
-
-    if (!response.isSuccessful()) {
-      logger.error(s"Failed to POST metrics to Datadog ${url} with status code [${response.code()}], Body: [${response.body().string()}]")
+    httpClient.doPost("application/json; charset=utf-8", buildRequestBody(snapshot)) match {
+      case Failure(e) =>
+        logger.error(e.getMessage)
+      case Success(response) =>
+        logger.info(response)
     }
-
-    response.close()
   }
 
-  private[datadog] def buildRequestBody(snapshot: PeriodSnapshot): String = {
+  private[datadog] def buildRequestBody(snapshot: PeriodSnapshot): Array[Byte] = {
     val timestamp = snapshot.from.getEpochSecond.toString
 
     val host = Kamon.environment.host
@@ -111,7 +105,9 @@ class DatadogAPIReporter extends MetricReporter {
     seriesBuilder
       .insert(0, "{\"series\":[")
       .append("]}")
-      .toString
+      .toString()
+      .getBytes(StandardCharsets.UTF_8)
+
   }
 
   private def scale(value: Long, unit: MeasurementUnit): Double = unit.dimension match {
@@ -124,23 +120,10 @@ class DatadogAPIReporter extends MetricReporter {
     case _ => value.toDouble
   }
 
-  // Apparently okhttp doesn't require explicit closing of the connection
-  private def createHttpClient(config: Configuration): OkHttpClient = {
-    new OkHttpClient.Builder()
-      .connectTimeout(config.connectTimeout.toMillis, TimeUnit.MILLISECONDS)
-      .readTimeout(config.connectTimeout.toMillis, TimeUnit.MILLISECONDS)
-      .writeTimeout(config.connectTimeout.toMillis, TimeUnit.MILLISECONDS).build()
-  }
-
   private def readConfiguration(config: Config): Configuration = {
     val datadogConfig = config.getConfig("kamon.datadog")
-
     Configuration(
-      apiUrl = datadogConfig.getString("http.api-url"),
-      apiKey = datadogConfig.getString("http.api-key"),
-      connectTimeout = datadogConfig.getDuration("http.connect-timeout"),
-      readTimeout = datadogConfig.getDuration("http.read-timeout"),
-      requestTimeout = datadogConfig.getDuration("http.request-timeout"),
+      datadogConfig.getConfig("http"),
       timeUnit = readTimeUnit(datadogConfig.getString("time-unit")),
       informationUnit = readInformationUnit(datadogConfig.getString("information-unit")),
       // Remove the "host" tag since it gets added to the datadog payload separately
@@ -154,8 +137,7 @@ private object DatadogAPIReporter {
   val count = "count"
   val gauge = "gauge"
 
-  case class Configuration(apiUrl: String, apiKey: String, connectTimeout: Duration, readTimeout: Duration, requestTimeout: Duration,
-                           timeUnit: MeasurementUnit, informationUnit: MeasurementUnit, extraTags: Map[String, String], tagFilter: Matcher)
+  case class Configuration(httpConfig: Config, timeUnit: MeasurementUnit, informationUnit: MeasurementUnit, extraTags: Map[String, String], tagFilter: Matcher)
 
   implicit class QuoteInterp(val sc: StringContext) extends AnyVal {
     def quote(args: Any*): String = "\"" + sc.s(args: _*) + "\""

src/main/scala/kamon/datadog/DatadogSpanReporter.scala

@@ -0,0 +1,89 @@
+package kamon.datadog
+
+import java.time.Duration
+
+import com.typesafe.config.Config
+import kamon.trace.IdentityProvider.Identifier
+import kamon.trace.{ IdentityProvider, Span }
+import kamon.{ Kamon, SpanReporter }
+import org.slf4j.LoggerFactory
+import play.api.libs.json.{ JsObject, Json }
+
+trait KamonDataDogTranslator {
+  def translate(span: Span.FinishedSpan): DdSpan
+}
+
+object KamonDataDogTranslatorDefault$ extends KamonDataDogTranslator {
+  def translate(span: Span.FinishedSpan): DdSpan = {
+    val traceId = BigInt(span.context.traceID.string, 16)
+    val spanId = BigInt(span.context.spanID.string, 16)
+
+    val parentId = span.context.parentID match {
+      case IdentityProvider.NoIdentifier => None
+      case Identifier(value, _)          => Some(BigInt(value, 16))
+    }
+
+    val name = span.tags.get("component")
+      .getOrElse(Span.TagValue.String("kamon.trace"))
+      .asInstanceOf[Span.TagValue.String]
+      .string
+    val resource = span.operationName
+    val service = Kamon.environment.service
+    val from = span.from
+    val start = from.getEpochNano
+    val duration = Duration.between(from, span.to)
+    val marks = span.marks.map { m => m.key -> m.instant.getEpochNano }.toMap
+    val tags = span.tags.map { m =>
+      m._1 -> {
+        m._2 match {
+          case v: Span.TagValue.Boolean => {
+            v.text match {
+              case "true"  => true
+              case "false" => false
+            }
+          }
+          case v: Span.TagValue.String => v.string
+          case v: Span.TagValue.Number => v.number
+          case null                    => null
+        }
+      }
+    }
+    val error: Boolean = tags.get("error") match {
+      case Some(v: Boolean) => v
+      case _                => false
+    }
+    val meta = marks ++ tags
+    new DdSpan(traceId, spanId, parentId, name, resource, service, "custom", start, duration, meta, error)
+
+  }
+}
+
+class DatadogSpanReporter extends SpanReporter {
+  private val translator: KamonDataDogTranslator = KamonDataDogTranslatorDefault$
+  private val logger = LoggerFactory.getLogger(classOf[DatadogAPIReporter])
+  final private val httpConfigPath = "kamon.datadog.trace.http"
+  private var httpClient = new HttpClient(Kamon.config().getConfig(httpConfigPath))
+
+  override def reportSpans(spans: Seq[Span.FinishedSpan]): Unit = if (spans.nonEmpty) {
+    val spanList: List[Seq[JsObject]] = spans
+      .map(span => translator.translate(span).toJson())
+      .groupBy { _.\("trace_id").get.toString() }
+      .values
+      .toList
+    httpClient.doJsonPut(Json.toJson(spanList))
+  }
+
+  override def start(): Unit = {
+    logger.info("Started the Kamon DataDog reporter")
+  }
+
+  override def stop(): Unit = {
+    logger.info("Stopped the Kamon DataDog reporter")
+  }
+
+  override def reconfigure(config: Config): Unit = {
+    logger.info("Reconfigured the Kamon DataDog reporter")
+    httpClient = new HttpClient(config.getConfig(httpConfigPath))
+  }
+
+}