Skip to content

Fix Netty 4.1 server span completion for streaming responses #11941

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
ygree wants to merge 3 commits into
base: ygree/fix-netty41-http11-pipelining-context-queue
Choose a base branch
from
+785 −116
Draft

Fix Netty 4.1 server span completion for streaming responses #11941

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b3f85fb
Add Netty chunked response span timing tests
ygree Jul 13, 2026
f65d4a4
Fix Netty 4.1 server span completion for streaming responses
ygree Jul 14, 2026
de23316
Fix Netty server span lifecycle for streaming responses
ygree Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 37 additions & 2 deletions ...in/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,13 @@
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpRequest;
import java.util.Deque;

@ChannelHandler.Sharable
public class HttpServerRequestTracingHandler extends ChannelInboundHandlerAdapter {
public static HttpServerRequestTracingHandler INSTANCE = new HttpServerRequestTracingHandler();
private static final String INCOMPLETE_RESPONSE_MESSAGE =
"Channel closed before response completed";

@Override
public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
Expand Down Expand Up @@ -54,7 +57,7 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
DECORATE.onRequest(span, channel, request, parentContext);

final ServerRequestContext serverContext =
ServerRequestContext.add(channel, context, request.headers());
ServerRequestContext.add(channel, context, request);

Flow.Action.RequestBlockingAction rba = span.getRequestBlockingAction();
if (rba != null) {
Expand Down Expand Up @@ -89,9 +92,41 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
super.channelInactive(ctx);
} finally {
try {
ServerRequestContext.closeAll(ctx.channel());
final Deque<ServerRequestContext> storedContexts =
ServerRequestContext.removeAll(ctx.channel());
if (storedContexts != null) {
ServerRequestContext storedContext;
while ((storedContext = storedContexts.pollFirst()) != null) {
if (storedContext.isResponseStarted()) {
finishSpanOnIncompleteResponse(storedContext);
} else {
publishSpanOnChannelClose(storedContext.tracingContext());
}
}
}
} catch (final Throwable ignored) {
}
}
}

private static void finishSpanOnIncompleteResponse(final ServerRequestContext serverContext) {
final Context storedContext = serverContext.tracingContext();
final AgentSpan span = AgentSpan.fromContext(storedContext);
if (span != null) {
DECORATE.onError(span, new IllegalStateException(INCOMPLETE_RESPONSE_MESSAGE));
if (!serverContext.isBeforeFinishCalled()) {
serverContext.markBeforeFinishCalled();
DECORATE.beforeFinish(storedContext);
}
span.finish();
}
}

private static void publishSpanOnChannelClose(final Context storedContext) {
final AgentSpan span = AgentSpan.fromContext(storedContext);
if (span != null && span.phasedFinish()) {
// At this point we can just publish this span to avoid losing the rest of the trace.
span.publish();
}
}
}
116 changes: 96 additions & 20 deletions ...n/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,20 +13,19 @@
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaderValues;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.util.concurrent.Future;

@ChannelHandler.Sharable
public class HttpServerResponseTracingHandler extends ChannelOutboundHandlerAdapter {
public static HttpServerResponseTracingHandler INSTANCE = new HttpServerResponseTracingHandler();

@Override
public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise prm) {
if (!(msg instanceof HttpResponse)) {
ctx.write(msg, prm);
return;
}

final ServerRequestContext serverContext = ServerRequestContext.nextResponse(ctx.channel());
final Context storedContext = serverContext == null ? null : serverContext.tracingContext();
final AgentSpan span = AgentSpan.fromContext(storedContext);
Expand All @@ -36,33 +35,110 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann
return;
}

try (final ContextScope scope = storedContext.attach()) {
final HttpResponse response = (HttpResponse) msg;
if (!(msg instanceof HttpResponse) && !serverContext.isResponseStarted()) {
ctx.write(msg, prm);
return;
}

try (final ContextScope ignored = storedContext.attach()) {
final boolean terminalResponse = isTerminalResponse(ctx, span, serverContext, msg);
final ChannelPromise writePromise = terminalResponse && prm.isVoid() ? ctx.newPromise() : prm;
try {
ctx.write(msg, prm);
if (terminalResponse) {
ServerRequestContext.remove(ctx.channel(), serverContext);
writePromise.addListener(
future -> finishSpan(serverContext, storedContext, span, future));
}
ctx.write(msg, writePromise);
if (serverContext.isResponseStarted()) {
beforeFinish(serverContext, storedContext);
}
} catch (final Throwable throwable) {
DECORATE.onError(span, throwable);
span.setHttpStatusCode(500);
span.finish(); // Finish the span manually since finishSpanOnClose was false
ServerRequestContext.remove(ctx.channel(), serverContext);
if (!terminalResponse) {
ServerRequestContext.remove(ctx.channel(), serverContext);
}
finishSpan(serverContext, storedContext, span);
throw throwable;
}
final boolean isWebsocketUpgrade =
response.status() == HttpResponseStatus.SWITCHING_PROTOCOLS
&& "websocket".equals(response.headers().get(HttpHeaderNames.UPGRADE));
}
}

private static boolean isTerminalResponse(
final ChannelHandlerContext ctx,
final AgentSpan span,
final ServerRequestContext serverContext,
final Object msg) {
if (msg instanceof HttpResponse) {
final HttpResponse response = (HttpResponse) msg;

final boolean isWebsocketUpgrade = isWebsocketUpgrade(response);
if (isInformationalResponse(response) && !isWebsocketUpgrade) {
return false;
}

if (isWebsocketUpgrade) {
ctx.channel()
.attr(WEBSOCKET_SENDER_HANDLER_CONTEXT)
.set(new HandlerContext.Sender(span, ctx.channel().id().asShortText()));
}
if (response.status() != HttpResponseStatus.CONTINUE
&& (response.status() != HttpResponseStatus.SWITCHING_PROTOCOLS || isWebsocketUpgrade)) {
DECORATE.onResponse(span, response);
DECORATE.beforeFinish(scope.context());
span.finish(); // Finish the span manually since finishSpanOnClose was false
ServerRequestContext.remove(ctx.channel(), serverContext);
}
DECORATE.onResponse(span, response);
serverContext.markResponseStarted();
return msg instanceof LastHttpContent
|| isBodylessResponse(serverContext, response)
|| isWebsocketUpgrade;
}
return serverContext.isResponseStarted() && msg instanceof LastHttpContent;
}

private static boolean isInformationalResponse(final HttpResponse response) {
final int statusCode = response.status().code();
return statusCode >= 100 && statusCode < 200;
}

private static boolean isWebsocketUpgrade(final HttpResponse response) {
return response.status() == HttpResponseStatus.SWITCHING_PROTOCOLS
&& response
.headers()
.containsValue(HttpHeaderNames.UPGRADE, HttpHeaderValues.WEBSOCKET, true);
}

private static boolean isBodylessResponse(
final ServerRequestContext serverContext, final HttpResponse response) {
final int statusCode = response.status().code();
return serverContext.isHeadRequest()
|| statusCode == 204
|| statusCode == 205
|| statusCode == 304
|| (serverContext.isConnectRequest() && statusCode >= 200 && statusCode < 300)
|| (HttpUtil.getContentLength(response, -1) == 0
&& !HttpUtil.isTransferEncodingChunked(response));
}

private static void finishSpan(
final ServerRequestContext serverContext,
final Context storedContext,
final AgentSpan span,
final Future<?> future) {
if (!future.isSuccess()) {
DECORATE.onError(span, future.cause());
span.setHttpStatusCode(500);
}
finishSpan(serverContext, storedContext, span);
}

private static void finishSpan(
final ServerRequestContext serverContext, final Context storedContext, final AgentSpan span) {
beforeFinish(serverContext, storedContext);
span.finish(); // Finish the span manually since finishSpanOnClose was false
}

private static void beforeFinish(
final ServerRequestContext serverContext, final Context storedContext) {
if (!serverContext.isBeforeFinishCalled()) {
serverContext.markBeforeFinishCalled();
DECORATE.beforeFinish(storedContext);
}
}
}
Loading