Fix grpc tests with java-concurrent.
MODIFY
dd-java-agent/instrumentation/grpc-1.5/grpc-1.5.gradle
dd-java-agent/instrumentation/grpc-1.5/grpc-1.5.gradle
@@ -47,6 +47,9 @@ dependencies {
testCompile group: 'io.grpc', name: 'grpc-netty', version: grpcVersion
testCompile group: 'io.grpc', name: 'grpc-protobuf', version: grpcVersion
testCompile group: 'io.grpc', name: 'grpc-stub', version: grpcVersion
+ testCompile group: 'javax.annotation', name: 'javax.annotation-api', version: '1.3.2'
+
+ testCompile project(':dd-java-agent:instrumentation:java-concurrent')
latestDepTestCompile sourceSets.test.output // include the protobuf generated classes
latestDepTestCompile group: 'io.grpc', name: 'grpc-netty', version: '+'

MODIFY
dd-java-agent/instrumentation/grpc-1.5/src/main/java/datadog/trace/instrumentation/grpc/server/TracingServerInterceptor.java
dd-java-agent/instrumentation/grpc-1.5/src/main/java/datadog/trace/instrumentation/grpc/server/TracingServerInterceptor.java
@@ -74,9 +74,9 @@ public class TracingServerInterceptor implements ServerInterceptor {
public void close(final Status status, final Metadata trailers) {
DECORATE.onClose(span, status);
try (final AgentScope scope = activateSpan(span, false)) {
- scope.setAsyncPropagation(true);
+ // Using async propagate here breaks the tests which use InProcessTransport.
+ // It also seems logical to not need it at all, so removing it.
delegate().close(status, trailers);
- scope.setAsyncPropagation(false);
} catch (final Throwable e) {
DECORATE.onError(span, e);
throw e;

MODIFY
dd-java-agent/instrumentation/grpc-1.5/src/test/groovy/GrpcStreamingTest.groovy
dd-java-agent/instrumentation/grpc-1.5/src/test/groovy/GrpcStreamingTest.groovy
+import datadog.opentracing.scopemanager.ContinuableScope
import datadog.trace.agent.test.AgentTestRunner
import datadog.trace.api.DDSpanTypes
import datadog.trace.bootstrap.instrumentation.api.Tags
@@ -9,6 +10,7 @@ import io.grpc.Server
import io.grpc.inprocess.InProcessChannelBuilder
import io.grpc.inprocess.InProcessServerBuilder
import io.grpc.stub.StreamObserver
+import io.opentracing.noop.NoopSpan
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.TimeUnit
@@ -29,22 +31,44 @@ class GrpcStreamingTest extends AgentTestRunner {
return new StreamObserver<Helloworld.Response>() {
@Override
void onNext(Helloworld.Response value) {
+
serverReceived << value.message
(1..msgCount).each {
+ if ((testTracer.scopeManager().active() as ContinuableScope).isAsyncPropagating()) {
+ // The InProcessTransport calls the client response in process, so we have to disable async propagation.
+ def tempScope = testTracer.scopeManager().activate(NoopSpan.INSTANCE)
observer.onNext(value)
+ tempScope.close()
+ } else {
+ observer.onError(new IllegalStateException("not async propagating!"))
+ }
}
}
@Override
void onError(Throwable t) {
+ if ((testTracer.scopeManager().active() as ContinuableScope).isAsyncPropagating()) {
+ // The InProcessTransport calls the client response in process, so we have to disable async propagation.
+ def tempScope = testTracer.scopeManager().activate(NoopSpan.INSTANCE)
error.set(t)
observer.onError(t)
+ tempScope.close()
+ } else {
+ observer.onError(new IllegalStateException("not async propagating!"))
+ }
}
@Override
void onCompleted() {
+ if ((testTracer.scopeManager().active() as ContinuableScope).isAsyncPropagating()) {
+ // The InProcessTransport calls the client response in process, so we have to disable async propagation.
+ def tempScope = testTracer.scopeManager().activate(NoopSpan.INSTANCE)
observer.onCompleted()
+ tempScope.close()
+ } else {
+ observer.onError(new IllegalStateException("not async propagating!"))
+ }
}
}
}
@@ -58,17 +82,29 @@ class GrpcStreamingTest extends AgentTestRunner {
def observer = client.conversation(new StreamObserver<Helloworld.Response>() {
@Override
void onNext(Helloworld.Response value) {
+ if ((testTracer.scopeManager().active() as ContinuableScope).isAsyncPropagating()) {
clientReceived << value.message
+ } else {
+ error.set(new IllegalStateException("not async propagating!"))
+ }
}
@Override
void onError(Throwable t) {
+ if ((testTracer.scopeManager().active() as ContinuableScope).isAsyncPropagating()) {
error.set(t)
+ } else {
+ error.set(new IllegalStateException("not async propagating!"))
+ }
}
@Override
void onCompleted() {
+ if ((testTracer.scopeManager().active() as ContinuableScope).isAsyncPropagating()) {
TEST_WRITER.waitForTraces(1)
+ } else {
+ error.set(new IllegalStateException("not async propagating!"))
+ }
}
})
@@ -80,6 +116,10 @@ class GrpcStreamingTest extends AgentTestRunner {
then:
error.get() == null
+ TEST_WRITER.waitForTraces(2)
+ error.get() == null
+ serverReceived == clientRange.collect { "call $it" }
+ clientReceived == serverRange.collect { clientRange.collect { "call $it" } }.flatten().sort()
assertTraces(2) {
trace(0, clientMessageCount + 1) {
@@ -148,9 +188,6 @@ class GrpcStreamingTest extends AgentTestRunner {
}
}
- serverReceived == clientRange.collect { "call $it" }
- clientReceived == serverRange.collect { clientRange.collect { "call $it" } }.flatten().sort()
-
cleanup:
channel?.shutdownNow()?.awaitTermination(10, TimeUnit.SECONDS)
server?.shutdownNow()?.awaitTermination()

MODIFY
dd-java-agent/instrumentation/grpc-1.5/src/test/groovy/GrpcTest.groovy
dd-java-agent/instrumentation/grpc-1.5/src/test/groovy/GrpcTest.groovy
+import datadog.common.exec.CommonTaskExecutor
import datadog.trace.agent.test.AgentTestRunner
import datadog.trace.api.DDSpanTypes
import datadog.trace.bootstrap.instrumentation.api.Tags
@@ -16,6 +17,9 @@ import io.grpc.stub.StreamObserver
import java.util.concurrent.TimeUnit
+import static datadog.trace.agent.test.utils.TraceUtils.basicSpan
+import static datadog.trace.agent.test.utils.TraceUtils.runUnderTrace
+
class GrpcTest extends AgentTestRunner {
def "test request-response"() {
@@ -25,17 +29,27 @@ class GrpcTest extends AgentTestRunner {
void sayHello(
final Helloworld.Request req, final StreamObserver<Helloworld.Response> responseObserver) {
final Helloworld.Response reply = Helloworld.Response.newBuilder().setMessage("Hello $req.name").build()
+ CommonTaskExecutor.INSTANCE.execute {
+ if (testTracer.activeSpan() == null) {
+ responseObserver.onError(new IllegalStateException("no active span"))
+ } else {
responseObserver.onNext(reply)
responseObserver.onCompleted()
}
}
+ }
+ }
Server server = InProcessServerBuilder.forName(getClass().name).addService(greeter).directExecutor().build().start()
ManagedChannel channel = InProcessChannelBuilder.forName(getClass().name).build()
GreeterGrpc.GreeterBlockingStub client = GreeterGrpc.newBlockingStub(channel)
when:
- def response = client.sayHello(Helloworld.Request.newBuilder().setName(name).build())
+ def response = runUnderTrace("parent") {
+ def resp = client.sayHello(Helloworld.Request.newBuilder().setName(name).build())
+ TEST_WRITER.waitForTraces(1) // Wait for the server span to be reported.
+ return resp
+ }
then:
response.message == "Hello $name"
@@ -46,7 +60,7 @@ class GrpcTest extends AgentTestRunner {
operationName "grpc.server"
resourceName "example.Greeter/SayHello"
spanType DDSpanTypes.RPC
- childOf trace(1).get(0)
+ childOf trace(1).get(1)
errored false
tags {
"$Tags.COMPONENT" "grpc-server"
@@ -70,13 +84,14 @@ class GrpcTest extends AgentTestRunner {
}
}
}
- trace(1, 2) {
- span(0) {
+ trace(1, 3) {
+ basicSpan(it, 0, "parent")
+ span(1) {
serviceName "unnamed-java-app"
operationName "grpc.client"
resourceName "example.Greeter/SayHello"
spanType DDSpanTypes.RPC
- parent()
+ childOf span(0)
errored false
tags {
"$Tags.COMPONENT" "grpc-client"
@@ -85,12 +100,12 @@ class GrpcTest extends AgentTestRunner {
defaultTags()
}
}
- span(1) {
+ span(2) {
serviceName "unnamed-java-app"
operationName "grpc.message"
resourceName "grpc.message"
spanType DDSpanTypes.RPC
- childOf span(0)
+ childOf span(1)
errored false
tags {
"$Tags.COMPONENT" "grpc-client"

DELETE
dd-java-agent/instrumentation/grpc-1.5/src/test/groovy/util/BlockingInterceptor.groovy
-package util
-
-import io.grpc.CallOptions
-import io.grpc.Channel
-import io.grpc.ClientCall
-import io.grpc.ClientInterceptor
-import io.grpc.ForwardingClientCall
-import io.grpc.Metadata
-import io.grpc.MethodDescriptor
-
-import java.util.concurrent.Phaser
-
-/**
- * Interceptor that blocks client from returning until server trace is reported.
- */
-class BlockingInterceptor implements ClientInterceptor {
- private final Phaser phaser
-
- BlockingInterceptor(Phaser phaser) {
- this.phaser = phaser
- phaser.register()
- }
-
- @Override
- <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
- return new ForwardingClientCall.SimpleForwardingClientCall(next.newCall(method, callOptions)) {
- @Override
- void start(final ClientCall.Listener responseListener, final Metadata headers) {
- super.start(new BlockingListener(responseListener, phaser), headers)
- }
- }
- }
-}

DELETE
dd-java-agent/instrumentation/grpc-1.5/src/test/groovy/util/BlockingListener.groovy
-package util
-
-import io.grpc.ClientCall
-import io.grpc.ForwardingClientCallListener
-import io.grpc.Metadata
-import io.grpc.Status
-
-import java.util.concurrent.Phaser
-
-class BlockingListener<RespT> extends ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT> {
- private final Phaser phaser
-
- BlockingListener(ClientCall.Listener<RespT> delegate, Phaser phaser) {
- super(delegate)
- this.phaser = phaser
- }
-
- @Override
- void onClose(final Status status, final Metadata trailers) {
- delegate().onClose(status, trailers)
- phaser.arriveAndAwaitAdvance()
- }
-}

Fix grpc tests with java-concurrent.