Merge pull request #521 from newrelic/revert-akka

Revert akka
newrelic · Oct 28, 2021 · 14b29f2 · 14b29f2
2 parents ebf39df + b234cc6
commit 14b29f2
Show file tree

Hide file tree

Showing 28 changed files with 185 additions and 462 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -32,7 +32,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 
 ### Fixes
-Restores transactions for Akka and Akka-Http [#497](https://github.com/newrelic/newrelic-java-agent/pull/497)
 The existing MongoDB sync client instrumentation was incorrectly applying when MongoDB reactive or async client was being used, which could lead to segment timeouts and long transaction response times. [#476](https://github.com/newrelic/newrelic-java-agent/pull/476)
 
 ### Deprecations/Removed Features

diff --git a/instrumentation/akka-http-core-2.11_10.0.11/build.gradle b/instrumentation/akka-http-core-2.11_10.0.11/build.gradle
@@ -10,7 +10,6 @@ dependencies {
     implementation(project(":agent-bridge"))
     implementation("com.typesafe.akka:akka-http-core_2.11:10.0.11")
     implementation("com.typesafe.akka:akka-stream_2.11:2.5.11")
-    testImplementation("com.typesafe.akka:akka-http_2.11:10.0.11")
 }
 
 verifyInstrumentation {

diff --git a/...on/akka-http-core-2.11_10.0.11/src/main/scala/akka/http/scaladsl/FlowRequestHandler.scala b/...on/akka-http-core-2.11_10.0.11/src/main/scala/akka/http/scaladsl/FlowRequestHandler.scala
diff --git a/...akka-http-core-2.11_10.0.11/src/main/scala/akka/http/scaladsl/HttpExtInstrumentation.java b/...akka-http-core-2.11_10.0.11/src/main/scala/akka/http/scaladsl/HttpExtInstrumentation.java
@@ -13,7 +13,6 @@
 import akka.http.scaladsl.settings.ConnectionPoolSettings;
 import akka.http.scaladsl.settings.ServerSettings;
 import akka.stream.Materializer;
-import akka.stream.scaladsl.Flow;
 import com.newrelic.api.agent.NewRelic;
 import com.newrelic.api.agent.Segment;
 import com.newrelic.api.agent.weaver.MatchType;
@@ -26,11 +25,29 @@
 @Weave(type = MatchType.ExactClass, originalName = "akka.http.scaladsl.HttpExt")
 public class HttpExtInstrumentation {
 
-    public Future<HttpInstrumentation.ServerBinding> bindAndHandle(Flow<HttpRequest, HttpResponse, ?> handler,
-                                                                   String _interface,
-                                                                   int port, ConnectionContext connectionContext,
-                                                                   ServerSettings settings, LoggingAdapter log, Materializer fm) {
-        handler = FlowRequestHandler$.MODULE$.instrumentFlow(handler);
+    public Future<HttpInstrumentation.ServerBinding> bindAndHandleAsync(
+            Function1<HttpRequest, Future<HttpResponse>> handler,
+            String interfaceString, int port,
+            ConnectionContext connectionContext,
+            ServerSettings settings, int parallelism,
+            LoggingAdapter adapter, Materializer mat) {
+
+        AsyncRequestHandler wrapperHandler = new AsyncRequestHandler(handler, mat.executionContext());
+        handler = wrapperHandler;
+
+        return Weaver.callOriginal();
+    }
+
+    public Future<HttpInstrumentation.ServerBinding> bindAndHandleSync(
+            Function1<HttpRequest, HttpResponse> handler,
+            String interfaceString, int port,
+            ConnectionContext connectionContext,
+            ServerSettings settings,
+            LoggingAdapter adapter, Materializer mat) {
+
+        SyncRequestHandler wrapperHandler = new SyncRequestHandler(handler);
+        handler = wrapperHandler;
+
         return Weaver.callOriginal();
     }
 

diff --git a/...ore-2.11_10.0.11/src/main/scala/com/nr/instrumentation/akkahttpcore/ResponseWrapper.scala b/...ore-2.11_10.0.11/src/main/scala/com/nr/instrumentation/akkahttpcore/ResponseWrapper.scala
@@ -9,11 +9,7 @@ package com.nr.instrumentation.akkahttpcore
 
 import akka.http.scaladsl.model.HttpResponse
 import akka.http.scaladsl.model.headers.RawHeader
-import com.newrelic.agent.bridge.{AgentBridge, Token}
-import com.newrelic.api.agent.weaver.Weaver
-import com.newrelic.api.agent.{HeaderType, Response, Transaction}
-
-import scala.concurrent.{ExecutionContext, Future}
+import com.newrelic.api.agent.{HeaderType, Response}
 
 class ResponseWrapper(var response: HttpResponse) extends Response {
 
@@ -35,41 +31,3 @@ class ResponseWrapper(var response: HttpResponse) extends Response {
     response = response.addHeader(new RawHeader(name, value))
   }
 }
-
-object ResponseWrapper {
-  def wrapAsyncResponse(token: Token)(implicit ec: ExecutionContext): HttpResponse => Future[HttpResponse] = {
-    response: HttpResponse => Future(wrapResponse(token, response))
-  }
-
-
-  def wrapResponse(token: Token, response: HttpResponse): HttpResponse = {
-    val localToken = token
-
-    try {
-      val txn: Transaction = localToken.getTransaction
-      if (txn != null) {
-        val wrappedResponse = new ResponseWrapper(response)
-        txn.setWebResponse(wrappedResponse)
-        txn.addOutboundResponseHeaders()
-        txn.markResponseSent()
-        val updatedResponse = wrappedResponse.response
-        localToken.expire()
-        updatedResponse
-      } else {
-        response
-      }
-
-    } catch {
-      case t: Throwable => AgentBridge.instrumentation.noticeInstrumentationError(t, Weaver.getImplementationTitle)
-        try {
-          localToken.expire()
-          response
-        }
-        catch {
-          case t: Throwable =>
-            AgentBridge.instrumentation.noticeInstrumentationError(t, Weaver.getImplementationTitle)
-            response
-        }
-    }
-  }
-}
diff --git a/...11_10.0.11/src/test/scala/com/agent/instrumentation/akka/http/core/AkkaHttpCoreTest.scala b/...11_10.0.11/src/test/scala/com/agent/instrumentation/akka/http/core/AkkaHttpCoreTest.scala
@@ -92,20 +92,6 @@ class AkkaHttpCoreTest {
     Assert.assertEquals("WebTransaction/AkkaHttpCore/akkaHandler", txName)
   }
 
-  @Test
-  def asyncHandlerPlayFlowServerTest(): Unit = {
-    playServer.startFromFlow(port)
-
-    Http().singleRequest(HttpRequest(uri = baseUrl + "/asyncPing"))
-
-    val introspector: Introspector = InstrumentationTestRunner.getIntrospector
-    awaitFinishedTx(introspector, 1)
-    playServer.stop()
-    Assert.assertEquals(1, introspector.getFinishedTransactionCount())
-    val txName = introspector.getTransactionNames.iterator.next
-    Assert.assertEquals("WebTransaction/AkkaHttpCore/akkaHandler", txName)
-  }
-
   @Test
   def syncHandlerPlayServerCatTest(): Unit = {
     playServer.start(port, async = false)

diff --git a/...ore-2.11_10.0.11/src/test/scala/com/agent/instrumentation/akka/http/core/PlayServer.scala b/...ore-2.11_10.0.11/src/test/scala/com/agent/instrumentation/akka/http/core/PlayServer.scala
@@ -8,14 +8,14 @@
 package com.agent.instrumentation.akka.http.core
 
 import akka.actor.ActorSystem
+import akka.event.Logging
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.model.HttpMethods._
 import akka.http.scaladsl.model._
-import akka.http.scaladsl.server.Directives.{complete, get, onSuccess, path}
-import akka.http.scaladsl.server.Route
 import akka.stream.ActorMaterializer
-import akka.stream.scaladsl._
+import akka.stream.scaladsl.{Source, _}
 import akka.util.Timeout
+import com.typesafe.config.ConfigFactory
 
 import scala.concurrent.duration._
 import scala.concurrent.{Await, Future}
@@ -27,31 +27,36 @@ class PlayServer() {
   implicit val materializer = ActorMaterializer()
   implicit val timeout: Timeout = 3 seconds
 
+  val config = ConfigFactory.load()
+  val logger = Logging(system, getClass)
+
   var bindingFuture: Future[Http.ServerBinding] = _
 
   def start(port: Int, async: Boolean) = {
-    val requestHandler: HttpRequest => HttpResponse = {
-      case HttpRequest(GET, Uri.Path("/ping"), _, _, _) => HttpResponse(entity = "Boops!")
-    }
-    val asyncRequestHandler: HttpRequest => Future[HttpResponse] = {
-      case HttpRequest(GET, Uri.Path("/asyncPing"), _, _, _) => Future(HttpResponse(entity = "Hoops!"))
+
+    if (async) {
+
+      val asyncRequestHandler: HttpRequest => Future[HttpResponse] = {
+        case HttpRequest(GET, Uri.Path("/asyncPing"), _, _, _) =>
+          Future[HttpResponse](HttpResponse(entity = "Hoops!"))
+      }
+
+      bindingFuture = Http().bindAndHandleAsync(asyncRequestHandler, interface = "localhost", port)
+
     }
-    bindingFuture = if (async)
-      Http().bindAndHandleAsync(asyncRequestHandler, interface = "localhost", port)
-    else
-      Http().bindAndHandleSync(requestHandler, interface = "localhost", port)
-    Await.ready(bindingFuture, timeout.duration)
-  }
+    else {
 
-  def startFromFlow(port: Int) = {
-    val routeFlow: Flow[HttpRequest, HttpResponse, _] = Route.handlerFlow(
-      path("ping") {
-        get(onSuccess(Future("Hoops"))(complete(_)))
+      val requestHandler: HttpRequest => HttpResponse = {
+        case HttpRequest(GET, Uri.Path("/ping"), _, _, _) =>
+          HttpResponse(entity = "Boops!")
       }
-    )
-    bindingFuture = Http()
-      .bindAndHandle(routeFlow, "localhost", port)
-    Await.ready(bindingFuture, timeout.duration)
+
+      bindingFuture = Http().bindAndHandleSync(requestHandler, interface = "localhost", port)
+    }
+
+    Await.ready({
+      bindingFuture
+    }, timeout.duration)
   }
 
   def stop() = {

diff --git a/instrumentation/akka-http-core-2.13_10.1.8/build.gradle b/instrumentation/akka-http-core-2.13_10.1.8/build.gradle
@@ -20,8 +20,6 @@ dependencies {
     implementation(project(":agent-bridge"))
     implementation("com.typesafe.akka:akka-http-core_2.13:10.1.8")
     implementation("com.typesafe.akka:akka-stream_2.13:2.5.23")
-    testImplementation("com.typesafe.akka:akka-http-spray-json_2.13:10.1.8")
-    testImplementation("com.typesafe.akka:akka-http_2.13:10.1.8")
 }
 
 verifyInstrumentation {

diff --git a/...ion/akka-http-core-2.13_10.1.8/src/main/scala/akka/http/scaladsl/FlowRequestHandler.scala b/...ion/akka-http-core-2.13_10.1.8/src/main/scala/akka/http/scaladsl/FlowRequestHandler.scala
diff --git a/.../akka-http-core-2.13_10.1.8/src/main/scala/akka/http/scaladsl/HttpExtInstrumentation.java b/.../akka-http-core-2.13_10.1.8/src/main/scala/akka/http/scaladsl/HttpExtInstrumentation.java
@@ -13,7 +13,6 @@
 import akka.http.scaladsl.settings.ConnectionPoolSettings;
 import akka.http.scaladsl.settings.ServerSettings;
 import akka.stream.Materializer;
-import akka.stream.scaladsl.Flow;
 import com.newrelic.api.agent.NewRelic;
 import com.newrelic.api.agent.Segment;
 import com.newrelic.api.agent.weaver.MatchType;
@@ -26,11 +25,29 @@
 @Weave(type = MatchType.ExactClass, originalName = "akka.http.scaladsl.HttpExt")
 public class HttpExtInstrumentation {
 
-    public Future<HttpInstrumentation.ServerBinding> bindAndHandle(Flow<HttpRequest, HttpResponse, ?> handler,
-                                                                   String _interface,
-                                                                   int port, ConnectionContext connectionContext,
-                                                                   ServerSettings settings, LoggingAdapter log, Materializer fm) {
-        handler = FlowRequestHandler$.MODULE$.instrumentFlow(handler);
+    public Future<HttpInstrumentation.ServerBinding> bindAndHandleAsync(
+            Function1<HttpRequest, Future<HttpResponse>> handler,
+            String interfaceString, int port,
+            ConnectionContext connectionContext,
+            ServerSettings settings, int parallelism,
+            LoggingAdapter adapter, Materializer mat) {
+
+        AsyncRequestHandler wrapperHandler = new AsyncRequestHandler(handler, mat.executionContext());
+        handler = wrapperHandler;
+
+        return Weaver.callOriginal();
+    }
+
+    public Future<HttpInstrumentation.ServerBinding> bindAndHandleSync(
+            Function1<HttpRequest, HttpResponse> handler,
+            String interfaceString, int port,
+            ConnectionContext connectionContext,
+            ServerSettings settings,
+            LoggingAdapter adapter, Materializer mat) {
+
+        SyncRequestHandler wrapperHandler = new SyncRequestHandler(handler);
+        handler = wrapperHandler;
+
         return Weaver.callOriginal();
     }