Skip to content

Fix Netty 4.1 HTTP/1.1 pipelined response tracing #11937

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

Open
ygree wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Fix Netty 4.1 HTTP/1.1 pipelined response tracing #11937

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
bc97709
test(netty-4.1): Add pipelining test
ygree Jul 13, 2026
7f7fdf4
fix(netty-4.1): pipelined response context tracking
ygree Jul 13, 2026
57f9d65
Disable Netty server tracing on context queue overflow
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
1 change: 1 addition & 0 deletions ...main/java/datadog/trace/instrumentation/netty41/ChannelFutureListenerInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ public ElementMatcher<TypeDescription> hierarchyMatcher() {
public String[] helperClassNames() {
return new String[] {
packageName + ".AttributeKeys",
packageName + ".ServerRequestContext",
// client helpers
packageName + ".client.NettyHttpClientDecorator",
packageName + ".client.NettyResponseInjectAdapter",
Expand Down
6 changes: 3 additions & 3 deletions ...java/datadog/trace/instrumentation/netty41/NettyChannelHandlerContextInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@
import static net.bytebuddy.matcher.ElementMatchers.isPublic;

import com.google.auto.service.AutoService;
import datadog.context.Context;
import datadog.trace.agent.tooling.Instrumenter;
import datadog.trace.agent.tooling.InstrumenterModule;
import datadog.trace.bootstrap.instrumentation.api.AgentScope;
Expand Down Expand Up @@ -48,6 +47,7 @@ public ElementMatcher<TypeDescription> hierarchyMatcher() {
public String[] helperClassNames() {
return new String[] {
packageName + ".AttributeKeys",
packageName + ".ServerRequestContext",
packageName + ".client.NettyHttpClientDecorator",
packageName + ".server.ResponseExtractAdapter",
packageName + ".server.NettyHttpServerDecorator",
Expand All @@ -71,8 +71,8 @@ public void methodAdvice(MethodTransformer transformer) {
public static class FireAdvice {
@Advice.OnMethodEnter(suppress = Throwable.class)
public static AgentScope scopeSpan(@Advice.This final ChannelHandlerContext ctx) {
final Context storedContext = ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).get();
final AgentSpan channelSpan = spanFromContext(storedContext);
final AgentSpan channelSpan =
spanFromContext(ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).get());
if (channelSpan == null || channelSpan == activeSpan()) {
// don't modify the scope
return noopScope();
Expand Down
1 change: 1 addition & 0 deletions .../main/java/datadog/trace/instrumentation/netty41/NettyChannelPipelineInstrumentation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,7 @@ public ElementMatcher<TypeDescription> hierarchyMatcher() {
public String[] helperClassNames() {
return new String[] {
packageName + ".AttributeKeys",
packageName + ".ServerRequestContext",
// client helpers
packageName + ".client.NettyHttpClientDecorator",
packageName + ".client.NettyResponseInjectAdapter",
Expand Down
26 changes: 11 additions & 15 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
@@ -1,16 +1,14 @@
package datadog.trace.instrumentation.netty41.server;

import static datadog.trace.instrumentation.netty41.AttributeKeys.ANALYZED_RESPONSE_KEY;
import static datadog.trace.instrumentation.netty41.AttributeKeys.BLOCKED_RESPONSE_KEY;
import static datadog.trace.instrumentation.netty41.AttributeKeys.CONTEXT_ATTRIBUTE_KEY;
import static datadog.trace.instrumentation.netty41.AttributeKeys.PARENT_CONTEXT_ATTRIBUTE_KEY;
import static datadog.trace.instrumentation.netty41.AttributeKeys.REQUEST_HEADERS_ATTRIBUTE_KEY;
import static datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator.DECORATE;

import datadog.context.Context;
import datadog.context.ContextScope;
import datadog.trace.api.gateway.Flow;
import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
import datadog.trace.instrumentation.netty41.ServerRequestContext;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
Expand Down Expand Up @@ -38,6 +36,12 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
}

final HttpRequest request = (HttpRequest) msg;
if (!ServerRequestContext.canTrackRequest(channel)) {
channel.attr(PARENT_CONTEXT_ATTRIBUTE_KEY).remove();
ctx.fireChannelRead(msg);
return;
}

final HttpHeaders headers = request.headers();
final Context storedParentContext = channel.attr(PARENT_CONTEXT_ATTRIBUTE_KEY).getAndRemove();
final Context parentContext =
Expand All @@ -49,11 +53,8 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
DECORATE.afterStart(span);
DECORATE.onRequest(span, channel, request, parentContext);

channel.attr(ANALYZED_RESPONSE_KEY).set(null);
channel.attr(BLOCKED_RESPONSE_KEY).set(null);

channel.attr(CONTEXT_ATTRIBUTE_KEY).set(context);
channel.attr(REQUEST_HEADERS_ATTRIBUTE_KEY).set(request.headers());
final ServerRequestContext serverContext =
ServerRequestContext.add(channel, context, request.headers());

Flow.Action.RequestBlockingAction rba = span.getRequestBlockingAction();
if (rba != null) {
Expand All @@ -76,7 +77,7 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
DECORATE.onError(span, throwable);
DECORATE.beforeFinish(ignored.context());
span.finish(); // Finish the span manually since finishSpanOnClose was false
ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).remove();
ServerRequestContext.remove(ctx.channel(), serverContext);
throw throwable;
}
}
Expand All @@ -88,12 +89,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
super.channelInactive(ctx);
} finally {
try {
final Context storedContext = ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).getAndRemove();
final AgentSpan span = AgentSpan.fromContext(storedContext);
if (span != null && span.phasedFinish()) {
// at this point we can just publish this span to avoid loosing the rest of the trace
span.publish();
}
ServerRequestContext.closeAll(ctx.channel());
} catch (final Throwable ignored) {
}
}
Expand Down
16 changes: 11 additions & 5 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
@@ -1,13 +1,13 @@
package datadog.trace.instrumentation.netty41.server;

import static datadog.trace.instrumentation.netty41.AttributeKeys.CONTEXT_ATTRIBUTE_KEY;
import static datadog.trace.instrumentation.netty41.AttributeKeys.WEBSOCKET_SENDER_HANDLER_CONTEXT;
import static datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator.DECORATE;

import datadog.context.Context;
import datadog.context.ContextScope;
import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
import datadog.trace.bootstrap.instrumentation.websocket.HandlerContext;
import datadog.trace.instrumentation.netty41.ServerRequestContext;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
Expand All @@ -22,10 +22,16 @@ public class HttpServerResponseTracingHandler extends ChannelOutboundHandlerAdap

@Override
public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise prm) {
final Context storedContext = ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).get();
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);

if (span == null || !(msg instanceof HttpResponse)) {
if (span == null) {
ctx.write(msg, prm);
return;
}
Expand All @@ -39,7 +45,7 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann
DECORATE.onError(span, throwable);
span.setHttpStatusCode(500);
span.finish(); // Finish the span manually since finishSpanOnClose was false
ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).remove();
ServerRequestContext.remove(ctx.channel(), serverContext);
throw throwable;
}
final boolean isWebsocketUpgrade =
Expand All @@ -55,7 +61,7 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann
DECORATE.onResponse(span, response);
DECORATE.beforeFinish(scope.context());
span.finish(); // Finish the span manually since finishSpanOnClose was false
ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).remove();
ServerRequestContext.remove(ctx.channel(), serverContext);
}
}
}
Expand Down
34 changes: 8 additions & 26 deletions ...src/main/java/datadog/trace/instrumentation/netty41/server/MaybeBlockResponseHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,5 @@
package datadog.trace.instrumentation.netty41.server;

import static datadog.trace.instrumentation.netty41.AttributeKeys.ANALYZED_RESPONSE_KEY;
import static datadog.trace.instrumentation.netty41.AttributeKeys.BLOCKED_RESPONSE_KEY;
import static datadog.trace.instrumentation.netty41.AttributeKeys.CONTEXT_ATTRIBUTE_KEY;
import static datadog.trace.instrumentation.netty41.AttributeKeys.REQUEST_HEADERS_ATTRIBUTE_KEY;
import static datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator.DECORATE;
import static io.netty.handler.codec.http.HttpHeaders.setContentLength;

Expand All @@ -13,6 +9,7 @@
import datadog.trace.api.gateway.RequestContext;
import datadog.trace.bootstrap.blocking.BlockingActionHelper;
import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
import datadog.trace.instrumentation.netty41.ServerRequestContext;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
Expand All @@ -36,36 +33,21 @@ public class MaybeBlockResponseHandler extends ChannelOutboundHandlerAdapter {

private MaybeBlockResponseHandler() {}

private static boolean isAnalyzedResponse(Channel ch) {
return ch.attr(ANALYZED_RESPONSE_KEY).get() != null;
}

private static void markAnalyzedResponse(Channel ch) {
ch.attr(ANALYZED_RESPONSE_KEY).set(Boolean.TRUE);
}

private static boolean isBlockedResponse(Channel ch) {
return ch.attr(BLOCKED_RESPONSE_KEY).get() != null;
}

private static void markBlockedResponse(Channel ch) {
ch.attr(BLOCKED_RESPONSE_KEY).set(Boolean.TRUE);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) throws Exception {
Channel channel = ctx.channel();

Context storedContext = channel.attr(CONTEXT_ATTRIBUTE_KEY).get();
ServerRequestContext serverContext = ServerRequestContext.nextResponse(channel);
Context storedContext = serverContext == null ? null : serverContext.tracingContext();
AgentSpan span = AgentSpan.fromContext(storedContext);
RequestContext requestContext;
if (span == null || (requestContext = span.getRequestContext()) == null) {
super.write(ctx, msg, prm);
return;
}

if (isAnalyzedResponse(channel)) {
if (isBlockedResponse(channel)) {
if (serverContext.isResponseAnalyzed()) {
if (serverContext.isResponseBlocked()) {
Comment on lines +49 to +50

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 16, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Keep dropping writes after a response block

When AppSec blocks a streaming response, this state now lives only on the head ServerRequestContext; the replacement blocking FullHttpResponse is written through HttpServerResponseTracingHandler, which immediately removes that context while the channel close is still asynchronous. Any subsequent original response chunks written before the close then see no blocked context (or the next pipelined request's context) and are forwarded instead of being released, regressing the previous channel-level blocked flag behavior for non-FullHttpResponse responses.

Useful? React with 👍 / 👎.

// block further writes
log.debug("Write suppressed, msg {} dropped", msg);
ReferenceCountUtil.release(msg);
Expand All @@ -88,14 +70,14 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) thr
Flow<Void> flow =
DECORATE.callIGCallbackResponseAndHeaders(
span, origResponse, origResponse.getStatus().code(), ResponseExtractAdapter.GETTER);
markAnalyzedResponse(channel);
serverContext.markResponseAnalyzed();
Flow.Action action = flow.getAction();
if (!(action instanceof Flow.Action.RequestBlockingAction)) {
super.write(ctx, msg, prm);
return;
}

markBlockedResponse(channel);
serverContext.markResponseBlocked();
Flow.Action.RequestBlockingAction rba = (Flow.Action.RequestBlockingAction) action;
int httpCode = BlockingActionHelper.getHttpCode(rba.getStatusCode());
HttpResponseStatus httpResponseStatus = HttpResponseStatus.valueOf(httpCode);
Expand All @@ -112,7 +94,7 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) thr

BlockingContentType bct = rba.getBlockingContentType();
if (bct != BlockingContentType.NONE) {
HttpHeaders reqHeaders = ctx.attr(REQUEST_HEADERS_ATTRIBUTE_KEY).get();
HttpHeaders reqHeaders = serverContext.requestHeaders();
String acceptHeader = reqHeaders != null ? reqHeaders.get("accept") : null;
BlockingActionHelper.TemplateType type =
BlockingActionHelper.determineTemplateType(bct, acceptHeader);
Expand Down
28 changes: 28 additions & 0 deletions ...tty-4.1/src/test/java/datadog/trace/instrumentation/netty41/ServerRequestContextTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
package datadog.trace.instrumentation.netty41;

import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;

import datadog.context.Context;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.util.DefaultAttributeMap;
import org.junit.jupiter.api.Test;

class ServerRequestContextTest {
private static final int PIPELINING_LIMIT = 1000;

@Test
void disablesTrackingWhenPipeliningLimitIsExceeded() {
DefaultAttributeMap attributes = new DefaultAttributeMap();

for (int i = 0; i < PIPELINING_LIMIT; i++) {
assertNotNull(ServerRequestContext.add(attributes, Context.root(), new DefaultHttpHeaders()));
}

assertFalse(ServerRequestContext.canTrackRequest(attributes));
assertNull(ServerRequestContext.nextResponse(attributes));
assertNull(attributes.attr(AttributeKeys.CONTEXT_ATTRIBUTE_KEY).get());
assertNull(ServerRequestContext.add(attributes, Context.root(), new DefaultHttpHeaders()));
}
}
Loading
Loading