Migrate Akka-Http Server tests and add ignoreParent()
MODIFY
dd-java-agent/instrumentation/akka-http-10.0/src/main/java/datadog/trace/instrumentation/akkahttp/AkkaHttpServerInstrumentation.java
dd-java-agent/instrumentation/akka-http-10.0/src/main/java/datadog/trace/instrumentation/akkahttp/AkkaHttpServerInstrumentation.java
@@ -104,6 +104,7 @@ public final class AkkaHttpServerInstrumentation extends Instrumenter.Default {
final Scope scope =
GlobalTracer.get()
.buildSpan("akka-http.request")
+ .ignoreActiveSpan()
.asChildOf(extractedContext)
.startActive(false);

MODIFY
dd-java-agent/instrumentation/akka-http-10.0/src/test/groovy/AkkaHttpServerInstrumentationTest.groovy
dd-java-agent/instrumentation/akka-http-10.0/src/test/groovy/AkkaHttpServerInstrumentationTest.groovy
-import datadog.trace.agent.test.AgentTestRunner
-import datadog.trace.agent.test.utils.OkHttpUtils
+import datadog.trace.agent.test.asserts.TraceAssert
+import datadog.trace.agent.test.base.HttpServerTest
import datadog.trace.api.DDSpanTypes
+import datadog.trace.instrumentation.akkahttp.AkkaHttpServerDecorator
import io.opentracing.tag.Tags
-import okhttp3.Request
-import spock.lang.Shared
-class AkkaHttpServerInstrumentationTest extends AgentTestRunner {
+import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
+import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.SUCCESS
- @Shared
- int asyncPort
- @Shared
- int syncPort
+abstract class AkkaHttpServerInstrumentationTest extends HttpServerTest<AkkaHttpServerDecorator> {
- @Shared
- def client = OkHttpUtils.client()
-
- def setupSpec() {
- AkkaHttpTestAsyncWebServer.start()
- asyncPort = AkkaHttpTestAsyncWebServer.port()
- AkkaHttpTestSyncWebServer.start()
- syncPort = AkkaHttpTestSyncWebServer.port()
+ @Override
+ AkkaHttpServerDecorator decorator() {
+ return AkkaHttpServerDecorator.DECORATE
}
- def cleanupSpec() {
- AkkaHttpTestAsyncWebServer.stop()
- AkkaHttpTestSyncWebServer.stop()
+ @Override
+ String expectedOperationName() {
+ return "akka-http.request"
}
- def "#server 200 request trace"() {
- setup:
- def request = new Request.Builder()
- .url("http://localhost:$port/test")
- .header("x-datadog-trace-id", "123")
- .header("x-datadog-parent-id", "456")
- .get()
- .build()
- def response = client.newCall(request).execute()
-
- expect:
- response.code() == 200
-
- assertTraces(1) {
- trace(0, 2) {
- span(0) {
- traceId "123"
- parentId "456"
- serviceName "unnamed-java-app"
- operationName "akka-http.request"
- resourceName "GET /test"
- spanType DDSpanTypes.HTTP_SERVER
- errored false
- tags {
- defaultTags(true)
- "$Tags.HTTP_STATUS.key" 200
- "$Tags.HTTP_URL.key" "http://localhost:$port/test"
- "$Tags.HTTP_METHOD.key" "GET"
- "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
- "$Tags.COMPONENT.key" "akka-http-server"
- }
- }
- span(1) {
- childOf span(0)
- assert span(1).operationName.endsWith('.tracedMethod')
- }
- }
- }
-
- where:
- server | port
- "async" | asyncPort
- "sync" | syncPort
+ @Override
+ boolean hasExceptionBody() {
+ false
}
- def "#server exceptions trace for #endpoint"() {
- setup:
- def request = new Request.Builder()
- .url("http://localhost:$port/$endpoint")
- .get()
- .build()
- def response = client.newCall(request).execute()
+// FIXME: This doesn't work because we don't support bindAndHandle.
+// @Override
+// void startServer(int port) {
+// AkkaHttpTestWebServer.start(port)
+// }
+//
+// @Override
+// void stopServer() {
+// AkkaHttpTestWebServer.stop()
+// }
- expect:
- response.code() == 500
-
- assertTraces(1) {
- trace(0, 1) {
- span(0) {
- serviceName "unnamed-java-app"
- operationName "akka-http.request"
- resourceName "GET /$endpoint"
+ void serverSpan(TraceAssert trace, int index, String traceID = null, String parentID = null, String method = "GET", ServerEndpoint endpoint = SUCCESS) {
+ trace.span(index) {
+ serviceName expectedServiceName()
+ operationName expectedOperationName()
+ resourceName endpoint.status == 404 ? "404" : "$method ${endpoint.resolve(address).path}"
spanType DDSpanTypes.HTTP_SERVER
- errored true
- tags {
- defaultTags()
- "$Tags.HTTP_STATUS.key" 500
- "$Tags.HTTP_URL.key" "http://localhost:$port/$endpoint"
- "$Tags.HTTP_METHOD.key" "GET"
- "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
- "$Tags.COMPONENT.key" "akka-http-server"
- errorTags RuntimeException, errorMessage
- }
+ errored endpoint.errored
+ if (parentID != null) {
+ traceId traceID
+ parentId parentID
+ } else {
+ parent()
}
- }
- }
-
- where:
- server | port | endpoint | errorMessage
- "async" | asyncPort | "throw-handler" | "Oh no handler"
- "async" | asyncPort | "throw-callback" | "Oh no callback"
- "sync" | syncPort | "throw-handler" | "Oh no handler"
- }
-
- def "#server 5xx trace"() {
- setup:
- def request = new Request.Builder()
- .url("http://localhost:$port/server-error")
- .get()
- .build()
- def response = client.newCall(request).execute()
-
- expect:
- response.code() == 500
-
- assertTraces(1) {
- trace(0, 1) {
- span(0) {
- serviceName "unnamed-java-app"
- operationName "akka-http.request"
- resourceName "GET /server-error"
- spanType DDSpanTypes.HTTP_SERVER
- errored true
tags {
- defaultTags()
- "$Tags.HTTP_STATUS.key" 500
- "$Tags.HTTP_URL.key" "http://localhost:$port/server-error"
- "$Tags.HTTP_METHOD.key" "GET"
+ defaultTags(true)
+ "$Tags.COMPONENT.key" serverDecorator.component()
+ if (endpoint.errored) {
+ "$Tags.ERROR.key" endpoint.errored
+ "error.msg" { it == null || it == EXCEPTION.body }
+ "error.type" { it == null || it == Exception.name }
+ "error.stack" { it == null || it instanceof String }
+ }
+ "$Tags.HTTP_STATUS.key" endpoint.status
+ "$Tags.HTTP_URL.key" "${endpoint.resolve(address)}"
+ "$Tags.HTTP_METHOD.key" method
"$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
- "$Tags.COMPONENT.key" "akka-http-server"
- "$Tags.ERROR.key" true
}
}
}
}
- where:
- server | port
- "async" | asyncPort
- "sync" | syncPort
+class AkkaHttpServerInstrumentationTestSync extends AkkaHttpServerInstrumentationTest {
+ @Override
+ void startServer(int port) {
+ AkkaHttpTestSyncWebServer.start(port)
}
- def "#server 4xx trace"() {
- setup:
- def request = new Request.Builder()
- .url("http://localhost:$port/not-found")
- .get()
- .build()
- def response = client.newCall(request).execute()
-
- expect:
- response.code() == 404
-
- assertTraces(1) {
- trace(0, 1) {
- span(0) {
- serviceName "unnamed-java-app"
- operationName "akka-http.request"
- resourceName "404"
- spanType DDSpanTypes.HTTP_SERVER
- errored false
- tags {
- defaultTags()
- "$Tags.HTTP_STATUS.key" 404
- "$Tags.HTTP_URL.key" "http://localhost:$port/not-found"
- "$Tags.HTTP_METHOD.key" "GET"
- "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
- "$Tags.COMPONENT.key" "akka-http-server"
- }
+ @Override
+ void stopServer() {
+ AkkaHttpTestSyncWebServer.stop()
}
}
+
+class AkkaHttpServerInstrumentationTestAsync extends AkkaHttpServerInstrumentationTest {
+ @Override
+ void startServer(int port) {
+ AkkaHttpTestAsyncWebServer.start(port)
}
- where:
- server | port
- "async" | asyncPort
- "sync" | syncPort
+ @Override
+ void stopServer() {
+ AkkaHttpTestAsyncWebServer.stop()
}
}

ADD
dd-java-agent/instrumentation/akka-http-10.0/src/test/groovy/AkkaHttpTestInstrumentation.java
+import static net.bytebuddy.matcher.ElementMatchers.named;
+
+import com.google.auto.service.AutoService;
+import datadog.trace.agent.test.base.HttpServerTestAdvice;
+import datadog.trace.agent.tooling.Instrumenter;
+import net.bytebuddy.agent.builder.AgentBuilder;
+
+@AutoService(Instrumenter.class)
+public class AkkaHttpTestInstrumentation implements Instrumenter {
+
+ @Override
+ public AgentBuilder instrument(final AgentBuilder agentBuilder) {
+ return agentBuilder
+ .type(named("akka.http.impl.engine.server.HttpServerBluePrint$PrepareRequests$$anon$1"))
+ .transform(
+ new AgentBuilder.Transformer.ForAdvice()
+ .advice(named("onPush"), HttpServerTestAdvice.ServerEntryAdvice.class.getName()));
+ }
+}

ADD
dd-java-agent/instrumentation/akka-http-10.0/src/test/scala/AkkaHttpTestAsyncWebServer.scala
+import akka.actor.ActorSystem
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.Http.ServerBinding
+import akka.http.scaladsl.model.HttpMethods.GET
+import akka.http.scaladsl.model._
+import akka.stream.ActorMaterializer
+import datadog.trace.agent.test.base.HttpServerTest
+import datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint._
+import groovy.lang.Closure
+
+import scala.concurrent.{Await, Future}
+
+object AkkaHttpTestAsyncWebServer {
+ implicit val system = ActorSystem("my-system")
+ implicit val materializer = ActorMaterializer()
+ // needed for the future flatMap/onComplete in the end
+ implicit val executionContext = system.dispatcher
+ val asyncHandler: HttpRequest => Future[HttpResponse] = {
+ case HttpRequest(GET, uri: Uri, _, _, _) => {
+ Future {
+ val endpoint = HttpServerTest.ServerEndpoint.forPath(uri.path.toString())
+ HttpServerTest.controller(endpoint, new Closure[HttpResponse]() {
+ def doCall(): HttpResponse = {
+ val resp = HttpResponse(status = endpoint.getStatus) //.withHeaders(headers.Type)resp.contentType = "text/plain"
+ endpoint match {
+ case SUCCESS => resp.withEntity(endpoint.getBody)
+ case REDIRECT => resp.withHeaders(headers.Location(endpoint.getBody))
+ case ERROR => resp.withEntity(endpoint.getBody)
+ case EXCEPTION => throw new Exception(endpoint.getBody)
+ case _ => HttpResponse(status = NOT_FOUND.getStatus).withEntity(NOT_FOUND.getBody)
+ }
+ }
+ })
+ }
+ }
+ }
+
+ private var binding: ServerBinding = null
+
+ def start(port: Int): Unit = synchronized {
+ if (null == binding) {
+ import scala.concurrent.duration._
+ binding = Await.result(Http().bindAndHandleAsync(asyncHandler, "localhost", port), 10 seconds)
+ }
+ }
+
+ def stop(): Unit = synchronized {
+ if (null != binding) {
+ binding.unbind()
+ system.terminate()
+ binding = null
+ }
+ }
+}

ADD
dd-java-agent/instrumentation/akka-http-10.0/src/test/scala/AkkaHttpTestSyncWebServer.scala
+import akka.actor.ActorSystem
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.Http.ServerBinding
+import akka.http.scaladsl.model.HttpMethods.GET
+import akka.http.scaladsl.model._
+import akka.stream.ActorMaterializer
+import datadog.trace.agent.test.base.HttpServerTest
+import datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint._
+import groovy.lang.Closure
+
+import scala.concurrent.Await
+
+object AkkaHttpTestSyncWebServer {
+ implicit val system = ActorSystem("my-system")
+ implicit val materializer = ActorMaterializer()
+ // needed for the future flatMap/onComplete in the end
+ implicit val executionContext = system.dispatcher
+ val syncHandler: HttpRequest => HttpResponse = {
+ case HttpRequest(GET, uri: Uri, _, _, _) => {
+ val endpoint = HttpServerTest.ServerEndpoint.forPath(uri.path.toString())
+ HttpServerTest.controller(endpoint, new Closure[HttpResponse]() {
+ def doCall(): HttpResponse = {
+ val resp = HttpResponse(status = endpoint.getStatus)
+ endpoint match {
+ case SUCCESS => resp.withEntity(endpoint.getBody)
+ case REDIRECT => resp.withHeaders(headers.Location(endpoint.getBody))
+ case ERROR => resp.withEntity(endpoint.getBody)
+ case EXCEPTION => throw new Exception(endpoint.getBody)
+ case _ => HttpResponse(status = NOT_FOUND.getStatus).withEntity(NOT_FOUND.getBody)
+ }
+ }
+ })
+ }
+ }
+
+ private var binding: ServerBinding = null
+
+ def start(port: Int): Unit = synchronized {
+ if (null == binding) {
+ import scala.concurrent.duration._
+ binding = Await.result(Http().bindAndHandleSync(syncHandler, "localhost", port), 10 seconds)
+ }
+ }
+
+ def stop(): Unit = synchronized {
+ if (null != binding) {
+ binding.unbind()
+ system.terminate()
+ binding = null
+ }
+ }
+}

MODIFY
dd-java-agent/instrumentation/akka-http-10.0/src/test/scala/AkkaHttpTestWebServer.scala
dd-java-agent/instrumentation/akka-http-10.0/src/test/scala/AkkaHttpTestWebServer.scala
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.Http.ServerBinding
-import akka.http.scaladsl.model.HttpMethods.GET
import akka.http.scaladsl.model._
+import akka.http.scaladsl.server.Directives._
+import akka.http.scaladsl.server.ExceptionHandler
import akka.stream.ActorMaterializer
-import datadog.trace.agent.test.utils.PortUtils
-import datadog.trace.api.Trace
+import datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint._
-import scala.concurrent.{Await, Future}
+import scala.concurrent.Await
-object AkkaHttpTestAsyncWebServer {
- val port = PortUtils.randomOpenPort()
+// FIXME: This doesn't work because we don't support bindAndHandle.
+object AkkaHttpTestWebServer {
implicit val system = ActorSystem("my-system")
implicit val materializer = ActorMaterializer()
// needed for the future flatMap/onComplete in the end
implicit val executionContext = system.dispatcher
- val asyncHandler: HttpRequest => Future[HttpResponse] = {
- case HttpRequest(GET, Uri.Path("/test"), _, _, _) =>
- Future {
- tracedMethod()
- HttpResponse(entity = "Hello unit test.")
- }
- case HttpRequest(GET, Uri.Path("/throw-handler"), _, _, _) =>
- sys.error("Oh no handler")
- case HttpRequest(GET, Uri.Path("/throw-callback"), _, _, _) =>
- Future {
- sys.error("Oh no callback")
- }
- case HttpRequest(GET, Uri.Path("/server-error"), _, _, _) =>
- Future {
- HttpResponse(entity = "Error unit test.", status = StatusCodes.InternalServerError)
- }
- case _ =>
- Future {
- HttpResponse(entity = "Not found unit test.", status = StatusCodes.NotFound)
- }
- }
-
- private var binding: ServerBinding = null
- def start(): Unit = synchronized {
- if (null == binding) {
- import scala.concurrent.duration._
- binding = Await.result(Http().bindAndHandleAsync(asyncHandler, "localhost", port), 10 seconds)
- }
- }
-
- def stop(): Unit = synchronized {
- if (null != binding) {
- binding.unbind()
- system.terminate()
- binding = null
- }
+ val exceptionHandler = ExceptionHandler {
+ case ex: Exception => complete(HttpResponse(status = EXCEPTION.getStatus).withEntity(ex.getMessage))
}
- @Trace
- def tracedMethod(): Unit = {
+ val route = { //handleExceptions(exceptionHandler) {
+ path(SUCCESS.rawPath) {
+ complete(HttpResponse(status = SUCCESS.getStatus).withEntity(SUCCESS.getBody))
+ } ~ path(REDIRECT.rawPath) {
+ redirect(Uri(REDIRECT.getBody), StatusCodes.Found)
+ } ~ path(ERROR.rawPath) {
+ complete(HttpResponse(status = ERROR.getStatus).withEntity(ERROR.getBody))
+ } ~ path(EXCEPTION.rawPath) {
+ failWith(new Exception(EXCEPTION.getBody))
}
}
-object AkkaHttpTestSyncWebServer {
- val port = PortUtils.randomOpenPort()
- implicit val system = ActorSystem("my-system")
- implicit val materializer = ActorMaterializer()
- // needed for the future flatMap/onComplete in the end
- implicit val executionContext = system.dispatcher
- val syncHandler: HttpRequest => HttpResponse = {
- case HttpRequest(GET, Uri.Path("/test"), _, _, _) =>
- tracedMethod()
- HttpResponse(entity = "Hello unit test.")
- case HttpRequest(GET, Uri.Path("/throw-handler"), _, _, _) =>
- sys.error("Oh no handler")
- case HttpRequest(GET, Uri.Path("/server-error"), _, _, _) =>
- HttpResponse(entity = "Error unit test.", status = StatusCodes.InternalServerError)
- case _ =>
- HttpResponse(entity = "Not found unit test.", status = StatusCodes.NotFound)
- }
-
private var binding: ServerBinding = null
- def start(): Unit = synchronized {
+ def start(port: Int): Unit = synchronized {
if (null == binding) {
import scala.concurrent.duration._
- binding = Await.result(Http().bindAndHandleSync(syncHandler, "localhost", port), 10 seconds)
+ binding = Await.result(Http().bindAndHandle(route, "localhost", port), 10 seconds)
}
}
@@ -93,8 +48,4 @@ object AkkaHttpTestSyncWebServer {
binding = null
}
}
-
- @Trace
- def tracedMethod(): Unit = {
- }
}

MODIFY
dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTest.groovy
dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTest.groovy
@@ -48,12 +48,14 @@ abstract class HttpServerTest<DECORATOR extends HttpServerDecorator> extends Age
def setupSpec() {
startServer(port)
+ println "Http server started at: http://localhost:$port/"
}
abstract void startServer(int port)
def cleanupSpec() {
stopServer()
+ println "Http server stopped at: http://localhost:$port/"
}
abstract void stopServer()
@@ -70,7 +72,11 @@ abstract class HttpServerTest<DECORATOR extends HttpServerDecorator> extends Age
true
}
- enum ServerEndpoint {
+ boolean hasExceptionBody() {
+ true
+ }
+
+ public enum ServerEndpoint {
SUCCESS("success", 200, "success"),
ERROR("error", 500, "controller error"),
EXCEPTION("exception", 500, "controller exception"),
@@ -90,10 +96,14 @@ abstract class HttpServerTest<DECORATOR extends HttpServerDecorator> extends Age
this.errored = status >= 500
}
- String getPath() {
+ public String getPath() {
return "/$path"
}
+ public String rawPath() {
+ return path
+ }
+
URI resolve(URI address) {
return address.resolve(path)
}
@@ -113,9 +123,9 @@ abstract class HttpServerTest<DECORATOR extends HttpServerDecorator> extends Age
static <T> T controller(ServerEndpoint endpoint, Closure<T> closure) {
if (endpoint == NOT_FOUND) {
- closure()
+ return closure()
} else {
- runUnderTrace("controller", closure)
+ return runUnderTrace("controller", closure)
}
}
@@ -204,7 +214,9 @@ abstract class HttpServerTest<DECORATOR extends HttpServerDecorator> extends Age
expect:
response.code() == EXCEPTION.status
- response.body().string() == EXCEPTION.body
+ if (hasExceptionBody()) {
+ assert response.body().string() == EXCEPTION.body
+ }
and:
cleanAndAssertTraces(1) {

MODIFY
dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTestAdvice.java
dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTestAdvice.java
package datadog.trace.agent.test.base;
import datadog.trace.api.DDTags;
+import datadog.trace.context.TraceScope;
import io.opentracing.Scope;
import io.opentracing.Tracer;
import io.opentracing.noop.NoopScopeManager;
@@ -24,10 +25,13 @@ public abstract class HttpServerTestAdvice {
if (tracer.activeSpan() != null) {
return NoopScopeManager.NoopScope.INSTANCE;
} else {
- return tracer
+ final Scope scope =
+ tracer
.buildSpan("TEST_SPAN")
.withTag(DDTags.RESOURCE_NAME, "ServerEntry")
.startActive(true);
+ ((TraceScope) scope).setAsyncPropagation(true);
+ return scope;
}
}

Migrate Akka-Http Server tests and add ignoreParent()