+ kamon-okhttp: introduce the new module kamon-okhttp to provide inst…

…rumentation for OkHttp (#808)

* +: introduce instrumentation module kamon-okhttp
* update reference.conf on kamon-okhttp
* add a missing Apache 2 header

build.sbt

@@ -127,7 +127,8 @@ lazy val instrumentation = (project in file("instrumentation"))
     `kamon-system-metrics`,
     `kamon-akka`,
     `kamon-akka-http`,
-    `kamon-play`
+    `kamon-play`,
+    `kamon-okhttp`,
   )
 
 
@@ -406,6 +407,23 @@ lazy val `kamon-play` = (project in file("instrumentation/kamon-play"))
   )
 
 
+lazy val `kamon-okhttp` = (project in file("instrumentation/kamon-okhttp"))
+  .disablePlugins(AssemblyPlugin)
+  .enablePlugins(JavaAgent)
+  .settings(instrumentationSettings)
+  .settings(
+    libraryDependencies ++= Seq(
+      kanelaAgent % "provided",
+      "com.squareup.okhttp3"      % "okhttp"                    % "3.14.9" % "provided",
+
+      scalatest % "test",
+      logbackClassic % "test",
+      "org.eclipse.jetty"         % "jetty-server"              % "9.4.25.v20191220" % "test",
+      "org.eclipse.jetty"         % "jetty-servlet"             % "9.4.25.v20191220" % "test",
+    )
+  ).dependsOn(`kamon-core`, `kamon-executors`, `kamon-testkit` % "test")
+
+
 /**
   * Reporters
   */
@@ -618,5 +636,6 @@ val `kamon-bundle` = (project in file("bundle/kamon-bundle"))
     `kamon-system-metrics` % "shaded",
     `kamon-akka` % "shaded",
     `kamon-akka-http` % "shaded",
-    `kamon-play` % "shaded"
+    `kamon-play` % "shaded",
+    `kamon-okhttp` % "shaded",
   )

instrumentation/kamon-okhttp/src/main/resources/reference.conf

@@ -0,0 +1,104 @@
+# ============================================ #
+# kamon okhttp3 client reference configuration #
+# ============================================ #
+
+kamon.instrumentation.okhttp {
+
+  http-client {
+    #
+    # Configuration for HTTP context propagation.
+    #
+    propagation {
+
+      # Enables or disables HTTP context propagation on this HTTP server instrumentation. Please note that if
+      # propagation is disabled then some distributed tracing features will not be work as expected (e.g. Spans can
+      # be created and reported but will not be linked across boundaries nor take trace identifiers from tags).
+      enabled = yes
+
+      # HTTP propagation channel to b used by this instrumentation. Take a look at the kamon.propagation.http.default
+      # configuration for more details on how to configure the detault HTTP context propagation.
+      channel = "default"
+    }
+
+    tracing {
+
+      # Enables HTTP request tracing. When enabled the instrumentation will create Spans for outgoing requests
+      # and finish them when the response is received from the server.
+      enabled = yes
+
+      # Enables collection of span metrics using the `span.processing-time` metric.
+      span-metrics = on
+
+      # Select which tags should be included as span and span metric tags. The possible options are:
+      #   - span: the tag is added as a Span tag (i.e. using span.tag(...))
+      #   - metric: the tag is added a a Span metric tag (i.e. using span.tagMetric(...))
+      #   - off: the tag is not used.
+      #
+      tags {
+
+        # Use the http.url tag.
+        url = span
+
+        # Use the http.method tag.
+        method = metric
+
+        # Use the http.status_code tag.
+        status-code = metric
+
+        # Copy tags from the context into the Spans with the specified purpouse. For example, to copy a customer_type
+        # tag from the context into the HTTP Server Span created by the instrumentation, the following configuration
+        # should be added:
+        #
+        # from-context {
+        #   customer_type = span
+        # }
+        #
+        from-context {
+
+        }
+      }
+
+      operations {
+
+        # The default operation name to be used when creating Spans to handle the HTTP client requests. The HTTP
+        # Client instrumentation will always try to use the HTTP Operation Name Generator configured bellow to get
+        # a name, but if it fails to generate it then this name will be used.
+        default = "http.client.request"
+
+        # FQCN for a HttpOperationNameGenerator implementation, or ony of the following shorthand forms:
+        #   - hostname: Uses the request Host as the operation name.
+        #   - method: Uses the request HTTP method as the operation name.
+        #
+        name-generator = "kamon.okhttp3.instrumentation.OkHttpOperationNameGenerator"
+      }
+    }
+  }
+}
+
+kamon {
+  okhttp {
+    # Fully qualified name of the implementation of kamon.okhttp3.NameGenerator that will be used for assigning names
+    # names to Spans.
+    name-generator = kamon.okhttp3.DefaultNameGenerator
+    # Metrics for okhttp
+    metrics {
+      enabled = true
+    }
+  }
+}
+
+kanela {
+  modules {
+    okhttp-module {
+      name = "OkHttp Instrumentation Module"
+      description = "Provides context propagation, distributed tracing and HTTP client and server metrics for OkHttp"
+      stoppable = true
+      instrumentations = [
+        "kamon.okhttp3.instrumentation.OkHttpInstrumentation"
+      ]
+      within = [
+        "okhttp3..*"
+      ]
+    }
+  }
+}

instrumentation/kamon-okhttp/src/main/scala/kamon/okhttp3/instrumentation/KamonOkHttpTracing.scala

@@ -0,0 +1,95 @@
+/*
+ * =========================================================================================
+ * Copyright © 2013-2020 the kamon project <http://kamon.io/>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
+ * except in compliance with the License. You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the
+ * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the specific language governing permissions
+ * and limitations under the License.
+ * =========================================================================================
+ */
+
+package kamon.okhttp3.instrumentation
+
+import java.util
+
+import kamon.Kamon
+import kamon.context.HttpPropagation.HeaderWriter
+import kamon.instrumentation.http.{HttpClientInstrumentation, HttpMessage}
+import okhttp3.{Request, Response}
+
+import scala.collection.immutable.Map
+import scala.collection.{JavaConverters, mutable}
+
+object KamonOkHttpTracing {
+  private val httpClientConfig = Kamon.config.getConfig("kamon.instrumentation.okhttp.http-client")
+  private val instrumentation = HttpClientInstrumentation.from(httpClientConfig, "okhttp-client")
+
+  def withNewSpan(request: Request): HttpClientInstrumentation.RequestHandler[Request] = {
+    instrumentation.createHandler(getRequestBuilder(request), Kamon.currentContext)
+  }
+
+  def successContinuation(requestHandler: HttpClientInstrumentation.RequestHandler[Request], response: Response): Response = {
+    requestHandler.processResponse(toKamonResponse(response))
+    response
+  }
+
+  def failureContinuation(requestHandler: HttpClientInstrumentation.RequestHandler[Request], error: Throwable): Unit = {
+    requestHandler.span.fail(error)
+    requestHandler.span.finish()
+  }
+
+  def getRequestBuilder(request: Request): HttpMessage.RequestBuilder[Request] = new HttpMessage.RequestBuilder[Request]() {
+    private val _headers = mutable.Map[String, String]()
+
+    override def read(header: String): Option[String] = Option.apply(request.header(header))
+
+    override def readAll: Map[String, String] = {
+      JavaConverters
+        .mapAsScalaMapConverter(request.headers.toMultimap)
+        .asScala
+        .mapValues((values: util.List[String]) => values.get(0))
+        .toMap
+    }
+
+    override def url: String = request.url.toString
+
+    override def path: String = request.url.uri.getPath
+
+    override def method: String = request.method
+
+    override def host: String = request.url.host
+
+    override def port: Int = request.url.port
+
+    override def write(header: String, value: String): Unit = {
+      _headers += (header -> value)
+    }
+
+    override def build: Request = {
+      val newHeadersMap = request.headers.newBuilder
+      _headers.foreach { case (key, value) => newHeadersMap.add(key, value) }
+      request.newBuilder.headers(newHeadersMap.build).build
+    }
+  }
+
+  def toKamonResponse(response: Response): HttpMessage.Response = new HttpMessage.Response() {
+    override def statusCode: Int = response.code()
+  }
+
+  trait HeaderHandler extends HeaderWriter {
+    private val _headers = mutable.Map[String, String]()
+
+    override def write(header: String, value: String): Unit = {
+      _headers += (header -> value)
+    }
+
+    def headers: mutable.Map[String, String] = _headers
+  }
+
+}

instrumentation/kamon-okhttp/src/main/scala/kamon/okhttp3/instrumentation/KamonTracingInterceptor.scala

@@ -0,0 +1,35 @@
+/*
+ * =========================================================================================
+ * Copyright © 2013-2020 the kamon project <http://kamon.io/>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
+ * except in compliance with the License. You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the
+ * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the specific language governing permissions
+ * and limitations under the License.
+ * =========================================================================================
+ */
+
+package kamon.okhttp3.instrumentation
+
+import okhttp3.{Interceptor, Response}
+
+final class KamonTracingInterceptor extends Interceptor {
+
+  override def intercept(chain: Interceptor.Chain): Response = {
+    val clientRequestHandler = KamonOkHttpTracing.withNewSpan(chain.request)
+    val request = clientRequestHandler.request
+    try {
+      val response = chain.proceed(request)
+      KamonOkHttpTracing.successContinuation(clientRequestHandler, response)
+    } catch {
+      case error: Throwable =>
+        KamonOkHttpTracing.failureContinuation(clientRequestHandler, error)
+        throw error
+    }
+  }
+}

instrumentation/kamon-okhttp/src/main/scala/kamon/okhttp3/instrumentation/OkHttpInstrumentation.scala

@@ -0,0 +1,50 @@
+/*
+ * =========================================================================================
+ * Copyright © 2013-2020 the kamon project <http://kamon.io/>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
+ * except in compliance with the License. You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the
+ * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the specific language governing permissions
+ * and limitations under the License.
+ * =========================================================================================
+ */
+
+package kamon.okhttp3.instrumentation
+
+import kanela.agent.api.instrumentation.InstrumentationBuilder
+import kanela.agent.libs.net.bytebuddy.asm.Advice
+import okhttp3.OkHttpClient
+
+class OkHttpInstrumentation extends InstrumentationBuilder {
+
+  /**
+    * Instrument:
+    *
+    * okhttp3.OkHttpClient::constructor
+    */
+  onType("okhttp3.OkHttpClient")
+    .advise(isConstructor() and takesOneArgumentOf("okhttp3.OkHttpClient$Builder"), classOf[OkHttpClientBuilderAdvisor])
+}
+
+/**
+  * Avisor for okhttp3.OkHttpClient::constructor(OkHttpClient.Builder)
+  */
+class OkHttpClientBuilderAdvisor
+
+object OkHttpClientBuilderAdvisor {
+
+  import scala.collection.JavaConverters._
+
+  @Advice.OnMethodEnter(suppress = classOf[Throwable])
+  def addKamonInterceptor(@Advice.Argument(0) builder: OkHttpClient.Builder): Unit = {
+    val interceptors = builder.networkInterceptors.asScala
+    if (!interceptors.exists(_.isInstanceOf[KamonTracingInterceptor])) {
+      builder.addNetworkInterceptor(new KamonTracingInterceptor)
+    }
+  }
+}

instrumentation/kamon-okhttp/src/main/scala/kamon/okhttp3/instrumentation/OkHttpOperationNameGenerator.scala

@@ -0,0 +1,25 @@
+/*
+ * =========================================================================================
+ * Copyright © 2013-2020 the kamon project <http://kamon.io/>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
+ * except in compliance with the License. You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the
+ * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the specific language governing permissions
+ * and limitations under the License.
+ * =========================================================================================
+ */
+
+package kamon.okhttp3.instrumentation
+
+import kamon.instrumentation.http.{HttpMessage, HttpOperationNameGenerator}
+
+class OkHttpOperationNameGenerator extends HttpOperationNameGenerator {
+  override def name(request: HttpMessage.Request): Option[String] = {
+    Option(request.url)
+  }
+}

instrumentation/kamon-okhttp/src/test/resources/logback.xml

@@ -0,0 +1,12 @@
+<configuration>
+    <statusListener class="ch.qos.logback.core.status.NopStatusListener"/>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT"/>
+    </root>
+</configuration>