rollback Future instrumentation to Kamon 1.x behavior (#1035)

Author: ivantopo

instrumentation/kamon-akka-http/src/main/resources/reference.conf

@@ -238,6 +238,8 @@ kanela.modules {
     name = "Akka HTTP Instrumentation"
     description = "Provides context propagation, distributed tracing and HTTP client and server metrics for Akka HTTP"
 
+    # Add "kamon.instrumentation.akka.http.FastFutureInstrumentation" to the instrumentation list if you need to
+    # use the Future Chaining instrumentation in Kamon 2.2.x. It will be fully removed in Kamon 2.3.0.
     instrumentations = [
       "kamon.instrumentation.akka.http.AkkaHttpServerInstrumentation"
       "kamon.instrumentation.akka.http.AkkaHttpClientInstrumentation"

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

@@ -78,6 +78,9 @@ class AkkaHttpServerInstrumentation extends InstrumentationBuilder {
     .intercept(method("redirect"), classOf[ResolveOperationNameOnRouteInterceptor])
     .intercept(method("failWith"), classOf[ResolveOperationNameOnRouteInterceptor])
 
+}
+
+class FastFutureInstrumentation extends InstrumentationBuilder {
 
   /**
     * This allows us to keep the right Context when Futures go through Akka HTTP's FastFuture and transformantions made
@@ -304,7 +307,9 @@ object FastFutureTransformWithAdvice {
       catch { case NonFatal(e) => FastFuture.failed(e) }
 
     // If we get a FulfilledFuture or ErrorFuture, those will have the HasContext mixin,
-    // otherwise we are getting a regular Future which has the context mixed into its value.
+    // otherwise we are getting a regular Future that might have a context mixed into its
+    // value if the Future Chaining instrumentation is manually enabled (it was deprecated
+    // in Kamon 2.1.21).
     if(future.isInstanceOf[HasContext])
       zuper.call()
     else {
@@ -316,18 +321,25 @@ object FastFutureTransformWithAdvice {
             case Failure(e) => p completeWith strictTransform(e, f)
           }(ec)
           p.future
+
         case Some(value) =>
-          // This is possible because of the Future's instrumentation
-          val futureContext = value.asInstanceOf[HasContext].context
-          val scope = Kamon.storeContext(futureContext)
+          value match {
+            case hc: HasContext =>
+              // The Future's value (Try[A]) might have a Context stored if the
+              // Future Chaining instrumentation is enabled.
+              val scope = Kamon.storeContext(hc.context)
+
+              val transformedFuture = value match {
+                case Success(a) => strictTransform(a, s)
+                case Failure(e) => strictTransform(e, f)
+              }
+
+              scope.close()
+              transformedFuture
 
-          val transformedFuture = value match {
             case Success(a) => strictTransform(a, s)
             case Failure(e) => strictTransform(e, f)
           }
-
-          scope.close()
-          transformedFuture
       }
     }
   }

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

@@ -96,9 +96,18 @@ class AkkaHttpServerInstrumentation extends InstrumentationBuilder {
     .intercept(method("redirect"), classOf[ResolveOperationNameOnRouteInterceptor])
     .intercept(method("failWith"), classOf[ResolveOperationNameOnRouteInterceptor])
 
+  /**
+    * Akka-http 10.1.x compatibility.
+    */
+
+  onType("akka.http.scaladsl.Http2Ext")
+    .advise(method("bindAndHandleAsync") and isPublic(), classOf[Http2ExtBindAndHandleAdvice])
+}
+
+class FastFutureInstrumentation extends InstrumentationBuilder {
 
   /**
-    * This allows us to keep the right Context when Futures go through Akka HTTP's FastFuture and transformantions made
+    * This allows us to keep the right Context when Futures go through Akka HTTP's FastFuture and transformations made
     * to them. Without this, it might happen that when a Future is already completed and used on any of the Futures
     * directives we might get a Context mess up.
     */
@@ -110,14 +119,6 @@ class AkkaHttpServerInstrumentation extends InstrumentationBuilder {
 
   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 {
@@ -329,7 +330,9 @@ object FastFutureTransformWithAdvice {
       catch { case NonFatal(e) => FastFuture.failed(e) }
 
     // If we get a FulfilledFuture or ErrorFuture, those will have the HasContext mixin,
-    // otherwise we are getting a regular Future which has the context mixed into its value.
+    // otherwise we are getting a regular Future that might have a context mixed into its
+    // value if the Future Chaining instrumentation is manually enabled (it was deprecated
+    // in Kamon 2.1.21).
     if(future.isInstanceOf[HasContext])
       zuper.call()
     else {
@@ -341,18 +344,25 @@ object FastFutureTransformWithAdvice {
             case Failure(e) => p completeWith strictTransform(e, f)
           }(ec)
           p.future
+
         case Some(value) =>
-          // This is possible because of the Future's instrumentation
-          val futureContext = value.asInstanceOf[HasContext].context
-          val scope = Kamon.storeContext(futureContext)
+          value match {
+            case hc: HasContext =>
+              // The Future's value (Try[A]) might have a Context stored if the
+              // Future Chaining instrumentation is enabled.
+              val scope = Kamon.storeContext(hc.context)
+
+              val transformedFuture = value match {
+                case Success(a) => strictTransform(a, s)
+                case Failure(e) => strictTransform(e, f)
+              }
+
+              scope.close()
+              transformedFuture
 
-          val transformedFuture = value match {
             case Success(a) => strictTransform(a, s)
             case Failure(e) => strictTransform(e, f)
           }
-
-          scope.close()
-          transformedFuture
       }
     }
   }

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

@@ -79,6 +79,16 @@ class AkkaHttpServerInstrumentation extends InstrumentationBuilder {
     .intercept(method("failWith"), classOf[ResolveOperationNameOnRouteInterceptor])
 
 
+  /**
+    * Akka-http 10.1.x compatibility.
+    */
+
+  onType("akka.http.scaladsl.Http2Ext")
+    .advise(method("bindAndHandleAsync") and isPublic(), classOf[Http2ExtBindAndHandleAdvice])
+}
+
+class FastFutureInstrumentation extends InstrumentationBuilder {
+
   /**
     * This allows us to keep the right Context when Futures go through Akka HTTP's FastFuture and transformantions made
     * to them. Without this, it might happen that when a Future is already completed and used on any of the Futures
@@ -92,13 +102,6 @@ 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 {
@@ -328,18 +331,25 @@ object FastFutureTransformWithAdvice {
             case Failure(e) => p completeWith strictTransform(e, f)
           }(ec)
           p.future
+
         case Some(value) =>
-          // This is possible because of the Future's instrumentation
-          val futureContext = value.asInstanceOf[HasContext].context
-          val scope = Kamon.storeContext(futureContext)
+          value match {
+            case hc: HasContext =>
+              // The Future's value (Try[A]) might have a Context stored if the
+              // Future Chaining instrumentation is enabled.
+              val scope = Kamon.storeContext(hc.context)
+
+              val transformedFuture = value match {
+                case Success(a) => strictTransform(a, s)
+                case Failure(e) => strictTransform(e, f)
+              }
+
+              scope.close()
+              transformedFuture
 
-          val transformedFuture = value match {
             case Success(a) => strictTransform(a, s)
             case Failure(e) => strictTransform(e, f)
           }
-
-          scope.close()
-          transformedFuture
       }
     }
   }

instrumentation/kamon-akka-http/src/test/scala/kamon/akka/http/FastFutureInstrumentationSpec.scala

@@ -18,7 +18,16 @@ import scala.util.Try
 
 class FastFutureInstrumentationSpec extends WordSpec with Matchers {
 
-  "the FastFuture instrumentation" should {
+  /**
+    *  DEPRECATED
+    *
+    *  This spec is ignored since Kamon 2.2.0, along with the deprecation of the Future Chaining instrumentation, and
+    *  should be completely removed before releasing Kamon 2.3.0.
+    *
+    *  We are keeping this spec only for the rare case that we might need to fix a bug on the Future Chaining
+    *  instrumentation while we keep it in maintenance mode.
+    */
+  "the FastFuture instrumentation" ignore {
     "keep the Context captured by the Future from which it was created" when {
       "calling .map/.flatMap/.onComplete and the original Future has not completed yet" in {
         val completeSignal = new CountDownLatch(1)

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

@@ -27,7 +27,6 @@ kanela.modules {
       "^com.sun.tools.*",
       "^sbt.internal.*",
       "^com.intellij.rt.*",
-      "^scala.concurrent.*",
       "^org.jboss.netty.*",
       "^com.google.common.base.internal.Finalizer",
       "^kamon.module.*",

instrumentation/kamon-scala-future/src/main/resources/reference.conf

@@ -21,10 +21,16 @@ kamon.instrumentation.futures.scala {
 }
 
 kanela.modules {
+  executor-service {
+    within += "scala.concurrent.impl.CallbackRunnable"
+    within += "scala.concurrent.impl.Future\\$PromiseCompletingRunnable"
+    within += "scala.concurrent.impl.Promise\\$Transformation"
+  }
 
   scala-future {
-    name = "Scala Future Instrumentation"
+    name = "Scala Future Chaining Instrumentation (Deprecated)"
     order = 1
+    enabled = false
     description = "Deprecated since Kamon 2.1.21, will be removed in Kamon 2.3.0. Provides automatic context propagation to the thread executing a Scala Future's body and callbacks"
     instrumentations = [
       "kamon.instrumentation.futures.scala.FutureChainingInstrumentation"