Skip to content

Commit d3bb33d

Browse files
gtukmachevgtukmachev
committed
Add instrumentation tests for chunked/streaming HTTP response handling
New JUnit 5 test class NettyChunkedResponseTest with a real Netty server that writes chunked responses manually (HttpResponse + HttpContent* + LastHttpContent), exercising the code path that HttpObjectAggregator-based tests never reach. Four test cases: - chunkedResponseSpanIncludesFullStreamDuration: span covers full stream time (~1s for 5 chunks x 200ms), not just header-send time (~0ms) - fullResponseStillFinishesSpanImmediately: FullHttpResponse regression - keepAliveSequentialChunkedRequestsGetCorrectSpans: STREAMING_CONTEXT_KEY lifecycle across back-to-back keep-alive requests - connectionDropDuringChunkedResponseFinishesSpan: span finished with error when client disconnects mid-stream
1 parent 74cb793 commit d3bb33d

1 file changed

Lines changed: 372 additions & 0 deletions

File tree

‎dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseTest.java‎

Lines changed: 372 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,372 @@
1+
package datadog.trace.instrumentation.netty41.server;
2+
3+
import static datadog.trace.agent.test.assertions.Matchers.any;
4+
import static datadog.trace.agent.test.assertions.Matchers.is;
5+
import static datadog.trace.agent.test.assertions.SpanMatcher.span;
6+
import static datadog.trace.agent.test.assertions.TagsMatcher.defaultTags;
7+
import static datadog.trace.agent.test.assertions.TagsMatcher.tag;
8+
import static datadog.trace.agent.test.assertions.TraceMatcher.trace;
9+
import static io.netty.handler.codec.http.HttpResponseStatus.OK;
10+
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
11+
import static org.junit.jupiter.api.Assertions.assertEquals;
12+
13+
import datadog.trace.agent.test.AbstractInstrumentationTest;
14+
import io.netty.bootstrap.ServerBootstrap;
15+
import io.netty.buffer.Unpooled;
16+
import io.netty.channel.Channel;
17+
import io.netty.channel.ChannelFutureListener;
18+
import io.netty.channel.ChannelHandlerContext;
19+
import io.netty.channel.ChannelInitializer;
20+
import io.netty.channel.EventLoopGroup;
21+
import io.netty.channel.SimpleChannelInboundHandler;
22+
import io.netty.channel.nio.NioEventLoopGroup;
23+
import io.netty.channel.socket.nio.NioServerSocketChannel;
24+
import io.netty.handler.codec.http.DefaultFullHttpResponse;
25+
import io.netty.handler.codec.http.DefaultHttpContent;
26+
import io.netty.handler.codec.http.DefaultHttpResponse;
27+
import io.netty.handler.codec.http.DefaultLastHttpContent;
28+
import io.netty.handler.codec.http.FullHttpRequest;
29+
import io.netty.handler.codec.http.HttpHeaderNames;
30+
import io.netty.handler.codec.http.HttpHeaderValues;
31+
import io.netty.handler.codec.http.HttpObjectAggregator;
32+
import io.netty.handler.codec.http.HttpResponseStatus;
33+
import io.netty.handler.codec.http.HttpServerCodec;
34+
import java.io.BufferedReader;
35+
import java.io.InputStreamReader;
36+
import java.net.HttpURLConnection;
37+
import java.net.InetSocketAddress;
38+
import java.net.Socket;
39+
import java.net.URL;
40+
import java.nio.charset.StandardCharsets;
41+
import java.time.Duration;
42+
import java.util.regex.Pattern;
43+
import org.junit.jupiter.api.AfterAll;
44+
import org.junit.jupiter.api.BeforeAll;
45+
import org.junit.jupiter.api.Test;
46+
import org.junit.jupiter.api.TestInstance;
47+
48+
/**
49+
* Tests that the Netty HTTP server instrumentation correctly handles chunked (streaming) responses.
50+
*
51+
* <p>The existing Netty41ServerTest uses HttpObjectAggregator which converts all responses to
52+
* FullHttpResponse — so the chunked path (HttpResponse + HttpContent* + LastHttpContent) is never
53+
* exercised. This test fills that gap.
54+
*/
55+
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
56+
public class NettyChunkedResponseTest extends AbstractInstrumentationTest {
57+
58+
private static final long CHUNK_DELAY_MS = 200;
59+
private static final int CHUNK_COUNT = 5;
60+
private static final Pattern NETTY_REQUEST = Pattern.compile("netty\\.request");
61+
private static final Pattern GET_CHUNKED = Pattern.compile("GET /chunked");
62+
private static final Pattern GET_FULL = Pattern.compile("GET /full");
63+
64+
private EventLoopGroup eventLoopGroup;
65+
private int port;
66+
67+
@BeforeAll
68+
void startServer() throws Exception {
69+
eventLoopGroup = new NioEventLoopGroup();
70+
ServerBootstrap bootstrap =
71+
new ServerBootstrap()
72+
.group(eventLoopGroup)
73+
.channel(NioServerSocketChannel.class)
74+
.childHandler(
75+
new ChannelInitializer<Channel>() {
76+
@Override
77+
protected void initChannel(Channel ch) {
78+
ch.pipeline().addLast(new HttpServerCodec());
79+
ch.pipeline().addLast(new HttpObjectAggregator(65536));
80+
ch.pipeline().addLast(new ChunkedTestHandler());
81+
}
82+
});
83+
Channel channel = bootstrap.bind(0).sync().channel();
84+
port = ((InetSocketAddress) channel.localAddress()).getPort();
85+
}
86+
87+
@AfterAll
88+
void stopServer() {
89+
if (eventLoopGroup != null) {
90+
eventLoopGroup.shutdownGracefully();
91+
}
92+
}
93+
94+
/**
95+
* Verifies that the span for a chunked HTTP response covers the full streaming duration, not just
96+
* the time to send headers. Without the fix in HttpServerResponseTracingHandler, the span would
97+
* finish when HttpResponse (headers) is written (~0ms), ignoring the time spent writing
98+
* HttpContent chunks and LastHttpContent.
99+
*/
100+
@Test
101+
void chunkedResponseSpanIncludesFullStreamDuration() throws Exception {
102+
String body = doGet("/chunked");
103+
assertEquals("chunk0chunk1chunk2chunk3chunk4", body);
104+
105+
long expectedMinDurationMs = CHUNK_DELAY_MS * CHUNK_COUNT;
106+
107+
assertTraces(
108+
trace(
109+
span()
110+
.root()
111+
.operationName(NETTY_REQUEST)
112+
.resourceName(GET_CHUNKED)
113+
.durationLongerThan(Duration.ofMillis(expectedMinDurationMs))
114+
.type("web")
115+
.tags(
116+
defaultTags(),
117+
tag("http.status_code", is(200)),
118+
tag("http.method", any()),
119+
tag("http.url", any()),
120+
tag("http.hostname", any()),
121+
tag("http.useragent", any()),
122+
tag("component", any()),
123+
tag("span.kind", any()),
124+
tag("peer.port", any()),
125+
tag("peer.ipv4", any()))));
126+
}
127+
128+
/**
129+
* Regression test: a non-chunked FullHttpResponse must still finish the span immediately. This
130+
* ensures the instanceof ordering fix (FullHttpResponse checked before HttpResponse and
131+
* LastHttpContent) does not break the standard single-message response path.
132+
*/
133+
@Test
134+
void fullResponseStillFinishesSpanImmediately() throws Exception {
135+
String body = doGet("/full");
136+
assertEquals("full-response", body);
137+
138+
assertTraces(
139+
trace(
140+
span()
141+
.root()
142+
.operationName(NETTY_REQUEST)
143+
.resourceName(GET_FULL)
144+
.durationShorterThan(Duration.ofMillis(500))
145+
.type("web")
146+
.tags(
147+
defaultTags(),
148+
tag("http.status_code", is(200)),
149+
tag("http.method", any()),
150+
tag("http.url", any()),
151+
tag("http.hostname", any()),
152+
tag("http.useragent", any()),
153+
tag("component", any()),
154+
tag("span.kind", any()),
155+
tag("peer.port", any()),
156+
tag("peer.ipv4", any()))));
157+
}
158+
159+
/**
160+
* Verifies that two sequential chunked requests each produce a correctly-timed span. This
161+
* exercises the STREAMING_CONTEXT_KEY lifecycle across multiple requests on the same connection:
162+
* each request must set and clear the key independently. Note: HttpURLConnection sends requests
163+
* sequentially (no pipelining), so this does not reproduce the concurrent race condition — it
164+
* validates that the streaming context bookkeeping works correctly for back-to-back requests.
165+
*/
166+
@Test
167+
void keepAliveSequentialChunkedRequestsGetCorrectSpans() throws Exception {
168+
URL url = new URL("http://localhost:" + port + "/chunked");
169+
170+
HttpURLConnection conn1 = (HttpURLConnection) url.openConnection();
171+
conn1.setRequestProperty("Connection", "keep-alive");
172+
String body1 = readResponse(conn1);
173+
assertEquals("chunk0chunk1chunk2chunk3chunk4", body1);
174+
conn1.disconnect();
175+
176+
HttpURLConnection conn2 = (HttpURLConnection) url.openConnection();
177+
conn2.setRequestProperty("Connection", "keep-alive");
178+
String body2 = readResponse(conn2);
179+
assertEquals("chunk0chunk1chunk2chunk3chunk4", body2);
180+
conn2.disconnect();
181+
182+
long expectedMinDurationMs = CHUNK_DELAY_MS * CHUNK_COUNT;
183+
184+
assertTraces(
185+
trace(
186+
span()
187+
.root()
188+
.operationName(NETTY_REQUEST)
189+
.resourceName(GET_CHUNKED)
190+
.durationLongerThan(Duration.ofMillis(expectedMinDurationMs))
191+
.type("web")
192+
.tags(
193+
defaultTags(),
194+
tag("http.status_code", is(200)),
195+
tag("http.method", any()),
196+
tag("http.url", any()),
197+
tag("http.hostname", any()),
198+
tag("http.useragent", any()),
199+
tag("component", any()),
200+
tag("span.kind", any()),
201+
tag("peer.port", any()),
202+
tag("peer.ipv4", any()))),
203+
trace(
204+
span()
205+
.root()
206+
.operationName(NETTY_REQUEST)
207+
.resourceName(GET_CHUNKED)
208+
.durationLongerThan(Duration.ofMillis(expectedMinDurationMs))
209+
.type("web")
210+
.tags(
211+
defaultTags(),
212+
tag("http.status_code", is(200)),
213+
tag("http.method", any()),
214+
tag("http.url", any()),
215+
tag("http.hostname", any()),
216+
tag("http.useragent", any()),
217+
tag("component", any()),
218+
tag("span.kind", any()),
219+
tag("peer.port", any()),
220+
tag("peer.ipv4", any()))));
221+
}
222+
223+
/**
224+
* Verifies that a streaming span is properly finished (not leaked) when the client disconnects
225+
* mid-stream. Without the channelInactive fix in HttpServerRequestTracingHandler, the span stored
226+
* in STREAMING_CONTEXT_KEY would never be finished if LastHttpContent is never written because
227+
* the connection dropped.
228+
*/
229+
@Test
230+
void connectionDropDuringChunkedResponseFinishesSpan() throws Exception {
231+
try (Socket socket = new Socket("localhost", port)) {
232+
socket.setSoTimeout(5000);
233+
socket
234+
.getOutputStream()
235+
.write("GET /slow-chunked HTTP/1.1\r\nHost: localhost\r\n\r\n".getBytes());
236+
socket.getOutputStream().flush();
237+
// Read until we get at least the first chunk — synchronization point before closing
238+
byte[] buf = new byte[512];
239+
socket.getInputStream().read(buf);
240+
}
241+
// Socket closed — channelInactive should fire and finish the streaming span
242+
243+
Thread.sleep(1500);
244+
245+
// The span must be finished (not leaked) and marked as error since the channel
246+
// closed before the response completed. Duration should be much shorter than
247+
// the full 10s streaming time since we disconnected early.
248+
assertTraces(
249+
trace(
250+
span()
251+
.root()
252+
.operationName(NETTY_REQUEST)
253+
.resourceName(Pattern.compile("GET /slow-chunked"))
254+
.type("web")
255+
.error()
256+
.durationShorterThan(Duration.ofMillis(5000))
257+
.tags(
258+
defaultTags(),
259+
tag("http.status_code", is(200)),
260+
tag("http.method", any()),
261+
tag("http.url", any()),
262+
tag("http.hostname", any()),
263+
tag("http.useragent", any()),
264+
tag("component", any()),
265+
tag("span.kind", any()),
266+
tag("peer.port", any()),
267+
tag("peer.ipv4", any()),
268+
tag("error.type", any()),
269+
tag("error.message", any()),
270+
tag("error.stack", any()))));
271+
}
272+
273+
private String doGet(String path) throws Exception {
274+
URL url = new URL("http://localhost:" + port + path);
275+
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
276+
try {
277+
return readResponse(conn);
278+
} finally {
279+
conn.disconnect();
280+
}
281+
}
282+
283+
private String readResponse(HttpURLConnection conn) throws Exception {
284+
StringBuilder sb = new StringBuilder();
285+
try (BufferedReader reader =
286+
new BufferedReader(new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8))) {
287+
String line;
288+
while ((line = reader.readLine()) != null) {
289+
sb.append(line);
290+
}
291+
}
292+
return sb.toString();
293+
}
294+
295+
static class ChunkedTestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
296+
@Override
297+
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
298+
String uri = request.uri();
299+
if ("/chunked".equals(uri)) {
300+
handleChunked(ctx);
301+
} else if ("/slow-chunked".equals(uri)) {
302+
handleSlowChunked(ctx);
303+
} else if ("/full".equals(uri)) {
304+
handleFull(ctx);
305+
} else {
306+
DefaultFullHttpResponse resp =
307+
new DefaultFullHttpResponse(
308+
HTTP_1_1,
309+
HttpResponseStatus.NOT_FOUND,
310+
Unpooled.copiedBuffer("not found", StandardCharsets.UTF_8));
311+
resp.headers().set(HttpHeaderNames.CONTENT_LENGTH, resp.content().readableBytes());
312+
ctx.writeAndFlush(resp).addListener(ChannelFutureListener.CLOSE);
313+
}
314+
}
315+
316+
private void handleChunked(ChannelHandlerContext ctx) {
317+
DefaultHttpResponse headers = new DefaultHttpResponse(HTTP_1_1, OK);
318+
headers.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
319+
ctx.writeAndFlush(headers);
320+
321+
ctx.executor()
322+
.execute(
323+
() -> {
324+
try {
325+
for (int i = 0; i < CHUNK_COUNT; i++) {
326+
Thread.sleep(CHUNK_DELAY_MS);
327+
byte[] data = ("chunk" + i).getBytes(StandardCharsets.UTF_8);
328+
ctx.writeAndFlush(new DefaultHttpContent(Unpooled.wrappedBuffer(data)));
329+
}
330+
ctx.writeAndFlush(new DefaultLastHttpContent());
331+
} catch (InterruptedException e) {
332+
Thread.currentThread().interrupt();
333+
ctx.close();
334+
}
335+
});
336+
}
337+
338+
private void handleSlowChunked(ChannelHandlerContext ctx) {
339+
DefaultHttpResponse headers = new DefaultHttpResponse(HTTP_1_1, OK);
340+
headers.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
341+
ctx.writeAndFlush(headers);
342+
343+
// Long streaming — 20 chunks × 500ms = 10s. The test will close the client socket
344+
// after the first chunk, triggering channelInactive before LastHttpContent is sent.
345+
ctx.executor()
346+
.execute(
347+
() -> {
348+
try {
349+
for (int i = 0; i < 20; i++) {
350+
if (!ctx.channel().isActive()) {
351+
return;
352+
}
353+
Thread.sleep(500);
354+
byte[] data = ("slow" + i).getBytes(StandardCharsets.UTF_8);
355+
ctx.writeAndFlush(new DefaultHttpContent(Unpooled.wrappedBuffer(data)));
356+
}
357+
ctx.writeAndFlush(new DefaultLastHttpContent());
358+
} catch (Exception e) {
359+
// Channel closed — expected
360+
}
361+
});
362+
}
363+
364+
private void handleFull(ChannelHandlerContext ctx) {
365+
byte[] body = "full-response".getBytes(StandardCharsets.UTF_8);
366+
DefaultFullHttpResponse resp =
367+
new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(body));
368+
resp.headers().set(HttpHeaderNames.CONTENT_LENGTH, body.length);
369+
ctx.writeAndFlush(resp);
370+
}
371+
}
372+
}

0 commit comments

Comments
 (0)