8000 Let propagate unhandled events and fix tests · DataDog/dd-trace-java@1ab91db · GitHub
[go: up one dir, main page]
Skip to content

Commit 1ab91db

Browse files
amarzialiamarziali
committed
Let propagate unhandled events and fix tests
1 parent e619f2a commit 1ab91db

File tree

4 files changed

+46
-71
lines changed

4 files changed

+46
-71
lines changed

‎dd-java-agent/instrumentation/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/websocket/WebSocketServerRequestTracingHandler.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,5 +99,7 @@ public void channelRead(ChannelHandlerContext ctx, Object frame) {
9999
}
100100
}
101101
}
102+
// can be other messages we do not handle like ping, pong or continuations
103+
ctx.fireChannelRead(frame);
102104
}
103105
}

‎dd-java-agent/instrumentation/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/websocket/WebSocketServerResponseTracingHandler.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -94,5 +94,7 @@ public void write(ChannelHandlerContext ctx, Object frame, ChannelPromise promis
9494
}
9595
}
9696
}
97+
// can be other messages we do not handle like ping, pong or continuations
98+
ctx.write(frame, promise);
9799
}
98100
}

‎dd-java-agent/instrumentation/netty-4.1/src/test/groovy/Netty41ServerTest.groovy

Lines changed: 40 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -1,29 +1,3 @@
1-
import io.netty.channel.ChannelFutureListener
2-
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame
3-
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame
4-
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame
5-
import io.netty.handler.codec.http.websocketx.WebSocketFrame
6-
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler
7-
8-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.BODY_URLENCODED
9-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.ERROR
10-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
11-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.FORWARDED
12-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.NOT_FOUND
13-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.QUERY_ENCODED_BOTH
14-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.QUERY_ENCODED_QUERY
15-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.QUERY_PARAM
16-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.REDIRECT
17-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.SUCCESS
18-
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.USER_BLOCK
19-
import static datadog.trace.agent.test.utils.TraceUtils.runUnderTrace
20-
import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH
21-
import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE
22-
import static io.netty.handler.codec.http.HttpResponseStatus.CONTINUE
23-
import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR
24-
import static io.netty.handler.codec.http.HttpUtil.is100ContinueExpected
25-
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1
26-
271
import datadog.appsec.api.blocking.Blocking
282
import datadog.trace.agent.test.base.HttpServer
293
import datadog.trace.agent.test.base.HttpServerTest
@@ -34,6 +8,7 @@ import datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator
348
import io.netty.bootstrap.ServerBootstrap
359
import io.netty.buffer.ByteBuf
3610
import io.netty.buffer.Unpooled
11+
import io.netty.channel.ChannelFutureListener
3712
import io.netty.channel.ChannelHandlerContext
3813
import io.netty.channel.ChannelInitializer
3914
import io.netty.channel.ChannelPipeline
@@ -51,10 +26,33 @@ import io.netty.handler.codec.http.HttpResponseStatus
5126
import io.netty.handler.codec.http.HttpServerCodec
5227
import io.netty.handler.codec.http.multipart.Attribute
5328
import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder
29+
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame
30+
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame
31+
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame
32+
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler
5433
import io.netty.handler.logging.LogLevel
5534
import io.netty.handler.logging.LoggingHandler
5635
import io.netty.util.CharsetUtil
5736

37+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.BODY_URLENCODED
38+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.ERROR
39+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
40+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.FORWARDED
41+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.NOT_FOUND
42+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.QUERY_ENCODED_BOTH
43+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.QUERY_ENCODED_QUERY
44+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.QUERY_PARAM
45+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.REDIRECT
46+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.SUCCESS
47+
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.USER_BLOCK
48+
import static datadog.trace.agent.test.utils.TraceUtils.runUnderTrace
49+
import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH
50+
import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE
51+
import static io.netty.handler.codec.http.HttpResponseStatus.CONTINUE
52+
import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR
53+
import static io.netty.handler.codec.http.HttpUtil.is100ContinueExpected
54+
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1
55+
5856
abstract class Netty41ServerTest extends HttpServerTest<EventLoopGroup> {
5957

6058
static final LoggingHandler LOGGING_HANDLER = new LoggingHandler(SERVER_LOGGER.name, LogLevel.DEBUG)
@@ -159,6 +157,7 @@ abstract class Netty41ServerTest extends HttpServerTest<EventLoopGroup> {
159157
}
160158
}
161159
if (msg instanceof TextWebSocketFrame || msg instanceof BinaryWebSocketFrame) {
160+
// generate a child span. The websocket test expects this way
162161
runUnderTrace("onRead", {})
163162
}
164163
},
@@ -172,15 +171,13 @@ abstract class Netty41ServerTest extends HttpServerTest<EventLoopGroup> {
172171
channelReadComplete: {
173172
it.flush()
174173
},
175-
handlerAdded: {
176-
ChannelHandlerContext ctx -> {
177-
WsEndpoint.onOpen(ctx)
178-
}
179-
},
180174
userEventTriggered: { ChannelHandlerContext ctx, Object evt ->
181-
if (evt == WebSocketServerProtocolHandler.HandshakeComplete) {
175+
if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {
182176
WsEndpoint.onOpen(ctx)
183177
}
178+
},
179+
channelInactive: { ChannelHandlerContext ctx ->
180+
WsEndpoint.onClose()
184181
}
185182
] as SimpleChannelInboundHandler)
186183
}
@@ -210,12 +207,16 @@ abstract class Netty41ServerTest extends HttpServerTest<EventLoopGroup> {
210207

211208
@Override
212209
void serverSendText(String[] messages) {
213-
WsEndpoint.onMessage(messages[0], false)
210+
for (int i = 0; i< messages.size(); i++ ) {
211+
WsEndpoint.activeSession.writeAndFlush(new TextWebSocketFrame(i == messages.size() - 1, 0, messages[i]))
212+
}
214213
}
215214

216215
@Override
217216
void serverSendBinary(byte[][] binaries) {
218-
WsEndpoint.onMessage(binaries[0], false)
217+
for (int i = 0; i< binaries.length; i++ ) {
218+
WsEndpoint.activeSession.writeAndFlush(new BinaryWebSocketFrame(i == binaries.length - 1, 0, Unpooled.wrappedBuffer(binaries[i])))
219+
}
219220
}
220221

221222
@Override
@@ -227,6 +228,11 @@ abstract class Netty41ServerTest extends HttpServerTest<EventLoopGroup> {
227228
void setMaxPayloadSize(int size) {
228229
// not applicable
229230
}
231+
232+
@Override
233+
boolean canSplitLargeWebsocketPayloads() {
234+
false
235+
}
230236
}
231237

232238
@Override
@@ -292,17 +298,4 @@ class WsEndpoint {
292298
static void onClose() {
293299
activeSession = null
294300
}
295-
296-
static void onMessage(String s, boolean last) {
297-
synchronized (WsEndpoint) {
298-
activeSession.writeAndFlush(new TextWebSocketFrame(s))
299-
}
300-
}
301-
302-
static void onMessage(byte[] b, boolean last) {
303-
synchronized (WsEndpoint) {
304-
ByteBuf responseBuf = Unpooled.wrappedBuffer(b)
305-
activeSession.writeAndFlush(new BinaryWebSocketFrame(responseBuf))
306-
}
307-
}
308301
}

‎dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTest.groovy

Lines changed: 2 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -46,17 +46,14 @@ import okhttp3.MultipartBody
4646
import okhttp3.Request
4747
import okhttp3.RequestBody
4848
import okhttp3.Response
49-
import okhttp3.WebSocket
5049
import okhttp3.WebSocketListener
5150
import okio.ByteString
5251
import org.slf4j.Logger
5352
import org.slf4j.LoggerFactory
5453

5554
import javax.annotation.Nonnull
56-
import java.util.concurrent.CountDownLatch
5755
import java.util.concurrent.ExecutorCompletionService
5856
import java.util.concurrent.Executors
59-
import java.util.concurrent.TimeUnit
6057
import java.util.function.BiFunction
6158
import java.util.function.Function
6259
import java.util.function.Supplier
@@ -1931,7 +1928,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
19311928
def request = new Request.Builder().url(HttpUrl.get(WEBSOCKET.resolve(address)))
19321929
.get().build()
19331930

1934-
def ws = websocketConnectAndWait(request)
1931+
client.newWebSocket(request, new WebSocketListener() {})
19351932
wsServer.awaitConnected()
19361933
runUnderTrace("parent", {
19371934
if (messages[0] instanceof String) {
@@ -1979,7 +1976,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
19791976
def request = new Request.Builder().url(HttpUrl.get(WEBSOCKET.resolve(address)))
19801977
.get().build()
19811978

1982-
def ws = websocketConnectAndWait(request)
1979+
def ws = client.newWebSocket(request, new WebSocketListener() {})
19831980
wsServer.awaitConnected()
19841981
wsServer.setMaxPayloadSize(10)
19851982
if (message instanceof String) {
@@ -2024,25 +2021,6 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
20242021
b
20252022
}
20262023

2027-
WebSocket websocketConnectAndWait(Request request) {
2028-
def failure = false
2029-
def latch = new CountDownLatch(1)
2030-
def ws = client.newWebSocket(request, new WebSocketListener() {
2031-
void onOpen(WebSocket webSocket, Response response) {
2032-
latch.countDown()
2033-
}
2034-
void onFailure(WebSocket webSocket, Throwable t, Response response) {
2035-
failure = true
2036-
latch.countDown()
2037-
}
2038-
})
2039-
2040-
if (!latch.await(20, TimeUnit.SECONDS) || failure) {
2041-
return null
2042-
}
2043-
return ws
2044-
}
2045-
20462024
static void websocketSendSpan(TraceAssert trace, DDSpan handshake, String messageType, int messageLength,
20472025
int nbOfChunks = 1, DDSpan parentSpan = null, Map extraTags = [:]) {
20482026
websocketSpan(trace, handshake, "websocket.send", messageType, messageLength, nbOfChunks,

0 commit comments

Comments
 (0)
0