Merge pull request #934 from SimunKaracic/akka-http-upgrade

Akka http upgrade

Author: SimunKaracic

.github/workflows/ci.yml

@@ -11,6 +11,6 @@ jobs:
       - run: git fetch --depth=1 origin '+refs/tags/*:refs/tags/*'
       - uses: olafurpg/setup-scala@v10
         with:
-          java-version: [email protected]242
+          java-version: [email protected]252
       - name: Test
         run: csbt +test

build.sbt

@@ -392,6 +392,12 @@ lazy val `kamon-akka` = (project in file("instrumentation/kamon-akka"))
   )
 
 
+def akkaHttpVersion(scalaVersion: String) = scalaVersion match {
+  case "2.11" => "10.1.12"
+  case _      => "10.2.3"
+}
+
+
 lazy val `kamon-akka-http` = (project in file("instrumentation/kamon-akka-http"))
   .enablePlugins(JavaAgent)
   .disablePlugins(AssemblyPlugin)
@@ -401,18 +407,18 @@ lazy val `kamon-akka-http` = (project in file("instrumentation/kamon-akka-http")
     javaAgents += "org.mortbay.jetty.alpn" % "jetty-alpn-agent" % "2.0.10" % "test",
     libraryDependencies ++= Seq(
       kanelaAgent % "provided",
-      "com.typesafe.akka" %% "akka-http"            % "10.1.12" % "provided",
-      "com.typesafe.akka" %% "akka-http2-support"   % "10.1.12" % "provided",
-      "com.typesafe.akka" %% "akka-stream"          % "2.5.31" % "provided",
+      "com.typesafe.akka" %% "akka-http"            % akkaHttpVersion(scalaBinaryVersion.value) % "provided",
+      "com.typesafe.akka" %% "akka-http2-support"   % akkaHttpVersion(scalaBinaryVersion.value)  % "provided",
+      "com.typesafe.akka" %% "akka-stream"          % "2.5.32" % "provided",
 
       scalatest % "test",
       slf4jApi % "test",
       slf4jnop % "test",
       okHttp % "test",
-      "com.typesafe.akka" %% "akka-http-testkit"    % "10.1.12" % "test",
+      "com.typesafe.akka" %% "akka-http-testkit"    % akkaHttpVersion(scalaBinaryVersion.value) % "test",
       "de.heikoseeberger" %% "akka-http-json4s"     % "1.27.0" % "test",
       "org.json4s"        %% "json4s-native"        % "3.6.7" % "test",
-    )
+    ),
   )).dependsOn(`kamon-akka`, `kamon-testkit` % "test")
 
 

instrumentation/kamon-akka-http/src/main/scala-2.11/kamon/instrumentation/akka/http/AkkaHttpServerInstrumentation.scala

@@ -353,4 +353,4 @@ object Http2BlueprintInterceptor {
         zuper.call()
     }
   }
-}
+}

instrumentation/kamon-akka-http/src/main/scala-2.12/kamon/instrumentation/akka/http/AkkaHttpServerInstrumentation.scala

@@ -58,6 +58,7 @@ class AkkaHttpServerInstrumentation extends InstrumentationBuilder {
     * operation name before the request processing hits the routing tree, we are delaying the sampling decision to the
     * point at which we have some operation name.
     */
+
   onType("akka.http.scaladsl.HttpExt")
     .advise(method("bindAndHandle"), classOf[HttpExtBindAndHandleAdvice])
 
@@ -66,7 +67,8 @@ class AkkaHttpServerInstrumentation extends InstrumentationBuilder {
     * creation happen at different times we are wrapping the handler with the interface/port data and reading that
     * information when turning the handler function into a flow and wrapping it the same way we would for HTTP/1.
     */
-  onType("akka.http.scaladsl.Http2Ext")
+
+  onType("akka.http.impl.engine.http2.Http2Ext")
     .advise(method("bindAndHandleAsync") and isPublic(), classOf[Http2ExtBindAndHandleAdvice])
 
   onType("akka.http.impl.engine.http2.Http2Blueprint$")
@@ -102,12 +104,20 @@ class AkkaHttpServerInstrumentation extends InstrumentationBuilder {
     */
   onTypes("akka.http.scaladsl.util.FastFuture$FulfilledFuture", "akka.http.scaladsl.util.FastFuture$ErrorFuture")
     .mixin(classOf[HasContext.MixinWithInitializer])
-      .advise(method("transform"), InvokeWithCapturedContext)
-      .advise(method("transformWith"), InvokeWithCapturedContext)
-      .advise(method("onComplete"), InvokeWithCapturedContext)
+    .advise(method("transform"), InvokeWithCapturedContext)
+    .advise(method("transformWith"), InvokeWithCapturedContext)
+    .advise(method("onComplete"), InvokeWithCapturedContext)
 
   onType("akka.http.scaladsl.util.FastFuture$")
     .intercept(method("transformWith$extension1"), FastFutureTransformWithAdvice)
+
+
+  /**
+    * Akka-http 10.1.x compatibility.
+    */
+
+  onType("akka.http.scaladsl.Http2Ext")
+    .advise(method("bindAndHandleAsync") and isPublic(), classOf[Http2ExtBindAndHandleAdvice])
 }
 
 trait HasMatchingContext {
@@ -369,4 +379,4 @@ object Http2BlueprintInterceptor {
         zuper.call()
     }
   }
-}
+}

instrumentation/kamon-akka-http/src/main/scala-2.13/kamon/instrumentation/akka/http/AkkaHttpServerInstrumentation.scala

@@ -32,7 +32,6 @@ import scala.collection.immutable
 
 
 class AkkaHttpServerInstrumentation extends InstrumentationBuilder {
-
   /**
     * When instrumenting bindAndHandle what we do is wrap the Flow[HttpRequest, HttpResponse, NotUsed] provided by
     * the user and add all the processing there. This is the part of the instrumentation that performs Context
@@ -49,8 +48,9 @@ class AkkaHttpServerInstrumentation extends InstrumentationBuilder {
     * For the HTTP/2 instrumentation, since the parts where we can capture the interface/port and the actual flow
     * creation happen at different times we are wrapping the handler with the interface/port data and reading that
     * information when turning the handler function into a flow and wrapping it the same way we would for HTTP/1.
+    *
     */
-  onType("akka.http.scaladsl.Http2Ext")
+  onType("akka.http.impl.engine.http2.Http2Ext")
     .advise(method("bindAndHandleAsync") and isPublic(), classOf[Http2ExtBindAndHandleAdvice])
 
   onType("akka.http.impl.engine.http2.Http2Blueprint$")
@@ -92,6 +92,13 @@ class AkkaHttpServerInstrumentation extends InstrumentationBuilder {
 
   onType("akka.http.scaladsl.util.FastFuture$")
     .intercept(method("transformWith$extension").and(takesArguments(4)), FastFutureTransformWithAdvice)
+
+  /**
+    * Akka-http 10.1.x compatibility.
+    */
+
+  onType("akka.http.scaladsl.Http2Ext")
+    .advise(method("bindAndHandleAsync") and isPublic(), classOf[Http2ExtBindAndHandleAdvice])
 }
 
 trait HasMatchingContext {
@@ -359,4 +366,4 @@ object Http2BlueprintInterceptor {
         zuper.call()
     }
   }
-}
+}

instrumentation/kamon-akka-http/src/main/scala/kamon/instrumentation/akka/http/AkkaHttpClientInstrumentation.scala

@@ -27,14 +27,22 @@ import kanela.agent.api.instrumentation.InstrumentationBuilder
 import scala.concurrent.Future
 import scala.util.{Failure, Success}
 
-class AkkaHttpClientInstrumentation extends InstrumentationBuilder {
+class AkkaHttpClientInstrumentation extends InstrumentationBuilder with VersionFiltering {
 
   /**
     * Simply modifies the requests as they are submitted. This does not cover connection pooling, just requests sent
     * via the Http.singleRequest mechanism.
     */
-  onType("akka.http.scaladsl.HttpExt")
-    .advise(method("singleRequestImpl"), classOf[HttpExtSingleRequestAdvice])
+
+  onAkkaHttp("10.1") {
+    onType("akka.http.scaladsl.HttpExt")
+      .advise(method("singleRequestImpl"), classOf[HttpExtSingleRequestAdvice])
+  }
+
+  onAkkaHttp("10.2") {
+    onType("akka.http.scaladsl.HttpExt")
+      .advise(method("singleRequest"), classOf[HttpExtSingleRequestAdvice])
+  }
 
   onType("akka.http.impl.engine.client.PoolMaster")
     .advise(method("dispatchRequest"), classOf[PoolMasterDispatchRequestAdvice])
@@ -58,4 +66,4 @@ object AkkaHttpClientInstrumentation {
 
     responseFuture
   }
-}
+}

instrumentation/kamon-akka-http/src/main/scala/kamon/instrumentation/akka/http/VersionFiltering.scala

@@ -0,0 +1,8 @@
+package kamon.instrumentation.akka.http
+
+trait VersionFiltering {
+  def onAkkaHttp(version: String)(block: => Unit): Unit = {
+    if(akka.http.Version.current.startsWith(version))
+      block
+  }
+}

instrumentation/kamon-instrumentation-common/src/main/scala/kamon/instrumentation/http/HttpOperationNameGenerator.scala

@@ -37,11 +37,12 @@ object HttpOperationNameGenerator {
     * Uses the request Host to assign a name.
     */
   object Hostname extends HttpOperationNameGenerator {
-    override def name(request: Request): Option[String] =
-      Option(request.host)
+    override def name(request: Request): Option[String] = {
+      Option(request.host).filter(_.nonEmpty) orElse request.read("host").map(_.takeWhile(_ != ':'))
+    }
   }
- 
- /**
+
+  /**
     * Uses the request Host and Port to assign a name.
     */
   object HostnameAndPort extends HttpOperationNameGenerator {
@@ -61,7 +62,7 @@ object HttpOperationNameGenerator {
   /**
     * Uses a static name.
     */
-  class Static(name:String) extends HttpOperationNameGenerator {
+  class Static(name: String) extends HttpOperationNameGenerator {
     override def name(request: Request): Option[String] =
       Option(name)
   }