akka-http sync and async tests
MODIFY
dd-java-agent/instrumentation/akka-http-10.0/akka-http-10.0.gradle
dd-java-agent/instrumentation/akka-http-10.0/akka-http-10.0.gradle
@@ -5,13 +5,16 @@ apply plugin: 'org.unbroken-dome.test-sets'
testSets {
lagomTest
}
-compileLagomTestGroovy {
- classpath = classpath.plus(files(compileLagomTestScala.destinationDir))
- dependsOn compileLagomTestScala
+
+compileLagomTestJava {
+ sourceCompatibility = 1.8
+ targetCompatibility = 1.8
}
+compileLagomTestGroovy {
sourceCompatibility = 1.8
targetCompatibility = 1.8
+}
dependencies {
compileOnly group: 'com.typesafe.akka', name: 'akka-http_2.11', version: '10.0.0'
@@ -37,3 +40,4 @@ dependencies {
}
test.dependsOn lagomTest
+testJava8Minimum += '*Test*.class'

MODIFY
dd-java-agent/instrumentation/akka-http-10.0/src/lagomTest/groovy/LagomTest.groovy
dd-java-agent/instrumentation/akka-http-10.0/src/lagomTest/groovy/LagomTest.groovy
import akka.NotUsed
import akka.stream.javadsl.Source
-import akka.stream.testkit.TestPublisher
import akka.stream.testkit.javadsl.TestSink
-import com.lightbend.lagom.javadsl.testkit.ServiceTest
-import datadog.opentracing.DDSpan
import org.junit.After
-import scala.concurrent.duration.FiniteDuration
-import static java.util.concurrent.TimeUnit.SECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS
import datadog.trace.agent.test.AgentTestRunner
import play.inject.guice.GuiceApplicationBuilder
import spock.lang.Shared
@@ -17,8 +13,13 @@ import akka.stream.testkit.TestSubscriber.Probe
import java.util.function.Function
import static com.lightbend.lagom.javadsl.testkit.ServiceTest.*
+import static datadog.trace.agent.test.ListWriterAssert.assertTraces
class LagomTest extends AgentTestRunner {
+ static {
+ System.setProperty("dd.integration.akkahttp.enabled", "true")
+ }
+
@Shared
private TestServer server
@@ -54,10 +55,8 @@ class LagomTest extends AgentTestRunner {
def "normal request traces" () {
setup:
- EchoService service = server.client(EchoService.class)
+ EchoService service = server.client(EchoService)
- // Use a source that never terminates (concat Source.maybe) so we
- // don't close the upstream, which would close the downstream
Source<String, NotUsed> input =
Source.from(Arrays.asList("msg1", "msg2", "msg3"))
.concat(Source.maybe())
@@ -70,57 +69,64 @@ class LagomTest extends AgentTestRunner {
probe.expectNext("msg3")
probe.cancel()
- TEST_WRITER.waitForTraces(1)
- DDSpan[] akkaTrace = TEST_WRITER.get(0)
- DDSpan root = akkaTrace[0]
-
expect:
- TEST_WRITER.size() == 1
- akkaTrace.size() == 2
-
- root.serviceName == "unnamed-java-app"
- root.operationName == "akkahttp.request"
- root.resourceName == "GET ws://?/echo"
- !root.context().getErrorFlag()
- root.context().tags["http.status_code"] == 101
- root.context().tags["http.url"] == "ws://localhost:${server.port()}/echo"
- root.context().tags["http.method"] == "GET"
- root.context().tags["span.kind"] == "server"
- root.context().tags["component"] == "akkahttp-action"
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 2) {
+ span(0) {
+ serviceName "unnamed-java-app"
+ operationName "akkahttp.request"
+ resourceName "GET ws://?/echo"
+ errored false
+ tags {
+ defaultTags()
+ "http.status_code" 101
+ "http.url" "ws://localhost:${server.port()}/echo"
+ "http.method" "GET"
+ "span.kind" "server"
+ "span.type" "web"
+ "component" "akkahttp-action"
+ }
+ }
+ span(1) {
+ childOf span(0)
+ }
+ }
+ }
}
def "error traces" () {
setup:
- EchoService service = server.client(EchoService.class)
+ EchoService service = server.client(EchoService)
- // Use a source that never terminates (concat Source.maybe) so we
- // don't close the upstream, which would close the downstream
Source<String, NotUsed> input =
Source.from(Arrays.asList("msg1", "msg2", "msg3"))
.concat(Source.maybe())
try {
- Source<String, NotUsed> output = service.error().invoke(input)
- .toCompletableFuture().get(5, SECONDS)
+ service.error().invoke(input).toCompletableFuture().get(5, SECONDS)
} catch (Exception e) {
}
- TEST_WRITER.waitForTraces(1)
- DDSpan[] akkaTrace = TEST_WRITER.get(0)
- DDSpan root = akkaTrace[0]
-
expect:
- TEST_WRITER.size() == 1
- akkaTrace.size() == 1
-
- root.serviceName == "unnamed-java-app"
- root.operationName == "akkahttp.request"
- root.resourceName == "GET ws://?/error"
- root.context().getErrorFlag()
- root.context().tags["http.status_code"] == 500
- root.context().tags["http.url"] == "ws://localhost:${server.port()}/error"
- root.context().tags["http.method"] == "GET"
- root.context().tags["span.kind"] == "server"
- root.context().tags["component"] == "akkahttp-action"
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 1) {
+ span(0) {
+ serviceName "unnamed-java-app"
+ operationName "akkahttp.request"
+ resourceName "GET ws://?/error"
+ errored true
+ tags {
+ defaultTags()
+ "http.status_code" 500
+ "http.url" "ws://localhost:${server.port()}/error"
+ "http.method" "GET"
+ "span.kind" "server"
+ "span.type" "web"
+ "component" "akkahttp-action"
+ "error" true
+ }
+ }
+ }
+ }
}
}

DELETE
dd-java-agent/instrumentation/akka-http-10.0/src/lagomTest/scala/LagomServer.scala
-// import com.lightbend.lagom.scaladsl.testkit.ServiceTest
-
-object LagomServer {
- // lagom application
- // val server: ServiceTest.TestServer = ServiceTest.startServer(ServiceTest.defaultSetup())
-
- def start(): Unit = {
- }
-}

MODIFY
dd-java-agent/instrumentation/akka-http-10.0/src/main/java/datadog/trace/instrumentation/akkahttp/AkkaHttpInstrumentation.java
dd-java-agent/instrumentation/akka-http-10.0/src/main/java/datadog/trace/instrumentation/akkahttp/AkkaHttpInstrumentation.java
@@ -38,11 +38,10 @@ public final class AkkaHttpInstrumentation extends Instrumenter.Configurable {
super("akkahttp");
}
- // TODO: sync vs async testing
- // TODO: Use test DSL
- // TODO: Merge into play testing
- // TODO: Disable Instrumentation by default
- // TODO: remove testWithScala from lagom
+ @Override
+ protected boolean defaultEnabled() {
+ return false;
+ }
private static final HelperInjector akkaHttpHelperInjector =
new HelperInjector(
@@ -58,6 +57,12 @@ public final class AkkaHttpInstrumentation extends Instrumenter.Configurable {
.type(named("akka.http.scaladsl.HttpExt"))
.transform(DDTransformers.defaultTransformers())
.transform(akkaHttpHelperInjector)
+ // Insturmenting akka-streams bindAndHandle api was previously attempted.
+ // This proved difficult as there was no clean way to close the async scope
+ // in the graph logic after the user's requst handler completes.
+ //
+ // Instead, we're instrumenting the bindAndHandle function helpers by
+ // wrapping the scala functions with our own handlers.
.transform(
DDAdvice.create()
.advice(

MODIFY
dd-java-agent/instrumentation/akka-http-10.0/src/test/groovy/AkkaHttpInstrumentationTest.groovy
dd-java-agent/instrumentation/akka-http-10.0/src/test/groovy/AkkaHttpInstrumentationTest.groovy
-import datadog.opentracing.DDSpan
import datadog.trace.agent.test.AgentTestRunner
import okhttp3.OkHttpClient
import okhttp3.Request
import spock.lang.Shared
+import static datadog.trace.agent.test.ListWriterAssert.assertTraces
+
class AkkaHttpInstrumentationTest extends AgentTestRunner {
+ static {
+ System.setProperty("dd.integration.akkahttp.enabled", "true")
+ }
+
@Shared
- int port
+ int asyncPort
+ @Shared
+ int syncPort
def setupSpec() {
- WebServer.start()
- port = WebServer.port()
+ AkkaHttpTestAsyncWebServer.start()
+ asyncPort = AkkaHttpTestAsyncWebServer.port()
+ AkkaHttpTestSyncWebServer.start()
+ syncPort = AkkaHttpTestSyncWebServer.port()
}
def cleanupSpec() {
- WebServer.stop()
+ AkkaHttpTestAsyncWebServer.stop()
+ AkkaHttpTestSyncWebServer.stop()
}
- def "200 request trace" () {
+ def "#server 200 request trace" () {
setup:
OkHttpClient client = new OkHttpClient.Builder().build()
def request = new Request.Builder()
@@ -27,32 +37,43 @@ class AkkaHttpInstrumentationTest extends AgentTestRunner {
.get()
.build()
def response = client.newCall(request).execute()
- TEST_WRITER.waitForTraces(1)
- DDSpan[] akkaTrace = TEST_WRITER.get(0)
- DDSpan root = akkaTrace[0]
+
expect:
response.code() == 200
- TEST_WRITER.size() == 1
- akkaTrace.size() == 2
- akkaTrace[1].operationName == 'WebServer$.tracedMethod'
- akkaTrace[1].parentId == akkaTrace[0].spanId
-
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 2) {
+ span(0) {
+ traceId 123
+ parentId 456
+ serviceName "unnamed-java-app"
+ operationName "akkahttp.request"
+ resourceName "GET /test"
+ errored false
+ tags {
+ defaultTags()
+ "http.status_code" 200
+ "http.url" "http://localhost:$port/test"
+ "http.method" "GET"
+ "span.kind" "server"
+ "span.type" "web"
+ "component" "akkahttp-action"
+ }
+ }
+ span(1) {
+ childOf span(0)
+ assert span(1).operationName.endsWith('.tracedMethod')
+ }
+ }
+ }
- root.traceId == 123
- root.parentId == 456
- root.serviceName == "unnamed-java-app"
- root.operationName == "akkahttp.request"
- root.resourceName == "GET /test"
- !root.context().getErrorFlag()
- root.context().tags["http.status_code"] == 200
- root.context().tags["http.url"] == "http://localhost:$port/test"
- root.context().tags["http.method"] == "GET"
- root.context().tags["span.kind"] == "server"
- root.context().tags["component"] == "akkahttp-action"
+ where:
+ server | port
+ "async" | asyncPort
+ "sync" | syncPort
}
- def "exceptions trace for #endpoint" () {
+ def "#server exceptions trace for #endpoint" () {
setup:
OkHttpClient client = new OkHttpClient.Builder().build()
def request = new Request.Builder()
@@ -60,34 +81,42 @@ class AkkaHttpInstrumentationTest extends AgentTestRunner {
.get()
.build()
def response = client.newCall(request).execute()
- TEST_WRITER.waitForTraces(1)
- DDSpan[] akkaTrace = TEST_WRITER.get(0)
- DDSpan root = akkaTrace[0]
expect:
response.code() == 500
- TEST_WRITER.size() == 1
-
- root.operationName == "akkahttp.request"
- root.resourceName == "GET /$endpoint"
- root.context().getErrorFlag()
- root.context().getTags()["error.type"] == RuntimeException.name
- root.context().getTags()["error.stack"].toString().startsWith("java.lang.RuntimeException: $errorMessage")
-
- root.context().tags["http.status_code"] == 500
- root.context().tags["http.url"] == "http://localhost:$port/$endpoint"
- root.context().tags["http.method"] == "GET"
- root.context().tags["span.kind"] == "server"
- root.context().tags["component"] == "akkahttp-action"
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 1) {
+ span(0) {
+ serviceName "unnamed-java-app"
+ operationName "akkahttp.request"
+ resourceName "GET /$endpoint"
+ errored true
+ tags {
+ defaultTags()
+ "http.status_code" 500
+ "http.url" "http://localhost:$port/$endpoint"
+ "http.method" "GET"
+ "span.kind" "server"
+ "span.type" "web"
+ "component" "akkahttp-action"
+ "error" true
+ "error.type" RuntimeException.name
+ "error.msg" errorMessage
+ "error.stack" tag("error.stack")
+ }
+ }
+ }
+ }
where:
- endpoint | errorMessage
- "throw-handler" | "Oh no handler"
- "throw-callback" | "Oh no callback"
+ 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 "5xx trace" () {
+ def "#server 5xx trace" () {
setup:
OkHttpClient client = new OkHttpClient.Builder().build()
def request = new Request.Builder()
@@ -95,29 +124,38 @@ class AkkaHttpInstrumentationTest extends AgentTestRunner {
.get()
.build()
def response = client.newCall(request).execute()
- TEST_WRITER.waitForTraces(1)
- DDSpan[] akkaTrace = TEST_WRITER.get(0)
- DDSpan root = akkaTrace[0]
+
expect:
response.code() == 500
- TEST_WRITER.size() == 1
- akkaTrace.size() == 1
-
- root.context().getErrorFlag()
- root.context().getTags()["error.stack"] == null
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 1) {
+ span(0) {
+ serviceName "unnamed-java-app"
+ operationName "akkahttp.request"
+ resourceName "GET /server-error"
+ errored true
+ tags {
+ defaultTags()
+ "http.status_code" 500
+ "http.url" "http://localhost:$port/server-error"
+ "http.method" "GET"
+ "span.kind" "server"
+ "span.type" "web"
+ "component" "akkahttp-action"
+ "error" true
+ }
+ }
+ }
+ }
- root.serviceName == "unnamed-java-app"
- root.operationName == "akkahttp.request"
- root.resourceName == "GET /server-error"
- root.context().tags["http.status_code"] == 500
- root.context().tags["http.url"] == "http://localhost:$port/server-error"
- root.context().tags["http.method"] == "GET"
- root.context().tags["span.kind"] == "server"
- root.context().tags["component"] == "akkahttp-action"
+ where:
+ server | port
+ "async" | asyncPort
+ "sync" | syncPort
}
- def "4xx trace" () {
+ def "#server 4xx trace" () {
setup:
OkHttpClient client = new OkHttpClient.Builder().build()
def request = new Request.Builder()
@@ -125,26 +163,33 @@ class AkkaHttpInstrumentationTest extends AgentTestRunner {
.get()
.build()
def response = client.newCall(request).execute()
- TEST_WRITER.waitForTraces(1)
- DDSpan[] akkaTrace = TEST_WRITER.get(0)
- DDSpan root = akkaTrace[0]
+
expect:
response.code() == 404
- TEST_WRITER.size() == 1
- akkaTrace.size() == 1
-
- !root.context().getErrorFlag()
- root.context().getTags()["error.stack"] == null
-
- root.serviceName == "unnamed-java-app"
- root.operationName == "akkahttp.request"
- root.resourceName == "404"
- root.context().tags["http.status_code"] == 404
- root.context().tags["http.url"] == "http://localhost:$port/not-found"
- root.context().tags["http.method"] == "GET"
- root.context().tags["span.kind"] == "server"
- root.context().tags["component"] == "akkahttp-action"
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 1) {
+ span(0) {
+ serviceName "unnamed-java-app"
+ operationName "akkahttp.request"
+ resourceName "404"
+ errored false
+ tags {
+ defaultTags()
+ "http.status_code" 404
+ "http.url" "http://localhost:$port/not-found"
+ "http.method" "GET"
+ "span.kind" "server"
+ "span.type" "web"
+ "component" "akkahttp-action"
+ }
+ }
}
}
+ where:
+ server | port
+ "async" | asyncPort
+ "sync" | syncPort
+ }
+}

RENAME
dd-java-agent/instrumentation/akka-http-10.0/src/test/scala/WebServer.scala
dd-java-agent/instrumentation/akka-http-10.0/src/test/scala/AkkaHttpTestWebServer.scala
@@ -3,16 +3,13 @@ import akka.http.scaladsl.Http
import akka.http.scaladsl.Http.ServerBinding
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.HttpMethods.GET
-import akka.http.scaladsl.server.Directives._
import akka.stream.ActorMaterializer
import datadog.trace.agent.test.TestUtils
import datadog.trace.api.Trace
-import datadog.trace.instrumentation.akkahttp.AkkaHttpInstrumentation.{DatadogGraph, DatadogLogic}
import scala.concurrent.{Await, Future}
-// TODO: rename -> AkkaHttpTestServer
-object WebServer {
+object AkkaHttpTestAsyncWebServer {
val port = TestUtils.randomOpenPort()
implicit val system = ActorSystem("my-system")
implicit val materializer = ActorMaterializer()
@@ -61,3 +58,43 @@ object WebServer {
def tracedMethod(): Unit = {
}
}
+
+object AkkaHttpTestSyncWebServer {
+ val port = TestUtils.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 {
+ 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
+ }
+ }
+
+ @Trace
+ def tracedMethod(): Unit = {
+ }
+}

MODIFY
dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/SpanAssert.groovy
dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/SpanAssert.groovy
@@ -42,6 +42,14 @@ class SpanAssert {
assert span.parentId == 0
}
+ def parentId(long parentId) {
+ assert span.parentId == parentId
+ }
+
+ def traceId(long traceId) {
+ assert span.traceId == traceId
+ }
+
def childOf(DDSpan parent) {
assert span.parentId == parent.spanId
assert span.traceId == parent.traceId

MODIFY
dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/TagsAssert.groovy
dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/TagsAssert.groovy
@@ -61,4 +61,8 @@ class TagsAssert {
void assertTagsAllVerified() {
assert tags.keySet() == assertedTags
}
+
+ Object tag(String tagName) {
+ return tags[tagName]
+ }
}

akka-http sync and async tests