Skip to content

Hybrid support for NRNS header for message queues and async kafka producers #2939

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
jbedell-newrelic wants to merge 12 commits into
base: main
Choose a base branch
from
Open

Hybrid support for NRNS header for message queues and async kafka producers #2939

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
95b9157
First commit working version, cleanup still to do
jbedell-newrelic Jun 12, 2026
7084d6b
Code cleanup
jbedell-newrelic Jun 22, 2026
2ea945f
Ported changes to the 1.28 module and more code cleanup
jbedell-newrelic Jun 22, 2026
5cfbb4d
todo cleanup
jbedell-newrelic Jun 22, 2026
f79bb82
more logging
jbedell-newrelic Jun 22, 2026
8c20655
Code review updates
jbedell-newrelic Jun 23, 2026
d2e4ebd
Go ahead and remove the span id from tracestate header as part of this.
jbedell-newrelic Jun 26, 2026
9f70801
Unit test updates
jbedell-newrelic Jun 26, 2026
ea375f8
Another unit test update
jbedell-newrelic Jun 26, 2026
68eb8de
More unit test updates
jbedell-newrelic Jun 26, 2026
a2c6ae6
Updates for async kafka producers
jbedell-newrelic Jul 16, 2026
cf416ae
Port the updates to the 1.59.0 module
jbedell-newrelic Jul 20, 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
16 changes: 15 additions & 1 deletion ...tension-autoconfigure-1.28.0/src/main/java/io/opentelemetry/sdk/trace/ExitTracerSpan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import com.newrelic.api.agent.DatastoreParameters;
import com.newrelic.api.agent.HttpParameters;
import com.newrelic.api.agent.NewRelic;
import com.newrelic.api.agent.Segment;
import com.newrelic.api.agent.Token;
import com.nr.agent.instrumentation.utils.AttributesHelper;
import com.nr.agent.instrumentation.utils.span.AttributeMapper;
Expand Down Expand Up @@ -83,6 +84,7 @@ public class ExitTracerSpan implements ReadWriteSpan {
.collect(Collectors.toSet()));

final ExitTracer tracer;
private final Segment segment;
private final SpanKind spanKind;
private final InstrumentationLibraryInfo instrumentationLibraryInfo;
private final Map<String, Object> attributes;
Expand Down Expand Up @@ -110,7 +112,15 @@ public class ExitTracerSpan implements ReadWriteSpan {
ExitTracerSpan(ExitTracer tracer, InstrumentationLibraryInfo instrumentationLibraryInfo, SpanKind spanKind, String spanName, SpanContext parentSpanContext,
Resource resource, Clock tracerClock, Map<String, Object> attributes, Consumer<ExitTracerSpan> onEnd, List<LinkData> links,
int totalNumberOfLinksAdded, long userStartEpochNanos) {
this(tracer, instrumentationLibraryInfo, spanKind, spanName, parentSpanContext,
resource, tracerClock, attributes, onEnd, links, totalNumberOfLinksAdded, userStartEpochNanos, null);
}

ExitTracerSpan(ExitTracer tracer, InstrumentationLibraryInfo instrumentationLibraryInfo, SpanKind spanKind, String spanName, SpanContext parentSpanContext,
Resource resource, Clock tracerClock, Map<String, Object> attributes, Consumer<ExitTracerSpan> onEnd, List<LinkData> links,
int totalNumberOfLinksAdded, long userStartEpochNanos, Segment segment) {
this.tracer = tracer;
this.segment = segment;
this.spanKind = spanKind;
this.spanName = spanName;
this.parentSpanContext = parentSpanContext;
Expand Down Expand Up @@ -285,7 +295,11 @@ public void end() {
copySpanLinksToTracer(links);
List<EventData> immutableEvents = this.events == null ? Collections.emptyList() : Collections.unmodifiableList(this.events);
copySpanEventsToTracer(immutableEvents);
tracer.finish();
if (segment != null) {
segment.end();
} else {
tracer.finish();
}
endEpochNanos = System.nanoTime();
ended = true;
onEnd.accept(this);
Expand Down
212 changes: 163 additions & 49 deletions ...xtension-autoconfigure-1.28.0/src/main/java/io/opentelemetry/sdk/trace/NRSpanBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
/*
*
* * Copyright 2024 New Relic Corporation. All rights reserved.
* * Copyright 2026 New Relic Corporation. All rights reserved.
* * SPDX-License-Identifier: Apache-2.0
*
*/
Expand All @@ -10,13 +10,18 @@
import com.newrelic.agent.bridge.AgentBridge;
import com.newrelic.agent.bridge.ExitTracer;
import com.newrelic.agent.bridge.Instrumentation;
import com.newrelic.agent.bridge.NoOpSegment;
import com.newrelic.agent.bridge.NoOpTransaction;
import com.newrelic.agent.bridge.Transaction;
import com.newrelic.agent.tracers.TracerFlags;
import com.newrelic.api.agent.ExtendedRequest;
import com.newrelic.api.agent.ExtendedResponse;
import com.newrelic.api.agent.HeaderType;
import com.newrelic.api.agent.Headers;
import com.newrelic.api.agent.NewRelic;
import com.newrelic.api.agent.Segment;
import com.newrelic.api.agent.TracedMethod;
import com.newrelic.api.agent.TransportType;
import com.nr.agent.instrumentation.utils.header.W3CTraceParentHeader;
import com.nr.agent.instrumentation.utils.span.AttributeMapper;
import com.nr.agent.instrumentation.utils.span.AttributeType;
Expand All @@ -33,6 +38,7 @@
import io.opentelemetry.sdk.trace.data.LinkData;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
Expand Down Expand Up @@ -72,6 +78,7 @@ class NRSpanBuilder implements SpanBuilder {
private static final int MAX_LINKS_PER_SPAN = 100;
private static final int MAX_LINK_ATTRIBUTES = 64;
private static final int MAX_LINK_ATTRIBUTE_LENGTH = 255;
private static final Boolean sendMessageQueueNotSampledHeader = NewRelic.getAgent().getConfig().getValue("distributed_tracing.send_message_queue_not_sampled_header", false);

public NRSpanBuilder(Instrumentation instrumentation, String instrumentationScopeName, String instrumentationScopeVersion, TracerSharedState sharedState,
String spanName) {
Expand Down Expand Up @@ -206,26 +213,57 @@ public Span startSpan() {
if (SpanKind.SERVER == spanKind) {
return startServerSpan(parentSpanContext);
}

final boolean dispatcher = SpanKind.CONSUMER.equals(spanKind);
if (dispatcher) {
AgentBridge.getAgent().getTransaction(true);
Transaction tx = AgentBridge.getAgent().getTransaction(true);
acceptMessageQueueNotSampledHeaderIfNecessary(tx);
}
final ExitTracer tracer = instrumentation.createTracer(spanName, getTracerFlags(dispatcher));
if (tracer == null) {
return NO_OP_SPAN;

Segment segment = null;
ExitTracer tracer;
if (SpanKind.PRODUCER == spanKind) {
insertMessageQueueNotSampledHeaderIfNecessary();
segment = NewRelic.getAgent().getTransaction().startSegment(spanName);
if (segment == null || segment == NoOpSegment.INSTANCE) {
return NO_OP_SPAN;
}
tracer = ((com.newrelic.agent.Segment)segment).getTracer();
if (tracer == null) {
// no active transaction
segment.end();
return NO_OP_SPAN;
}
} else {
tracer = instrumentation.createTracer(spanName, getTracerFlags(dispatcher));
if (tracer == null) {
return NO_OP_SPAN;
}
}

if (SpanKind.INTERNAL != spanKind) {
tracer.addCustomAttribute("span.kind", spanKind.name());
}
List<LinkData> immutableLinks = this.links == null ? Collections.emptyList() : Collections.unmodifiableList(this.links);
// TODO REVIEW - we're not picking up the global resources
return onStart(new ExitTracerSpan(tracer, instrumentationLibraryInfo, spanKind, spanName, parentSpanContext, sharedState.getResource(), sharedState.getClock(), attributes,
endHandler, immutableLinks, totalNumberOfLinksAdded, startEpochNanos));
endHandler, immutableLinks, totalNumberOfLinksAdded, startEpochNanos, segment));
}

private Span startServerSpan(SpanContext parentSpanContext) {
Transaction transaction = AgentBridge.getAgent().getTransaction(true);
final ExtendedRequest request = new ExtendedRequest() {
final ExtendedRequest request = generateExtendedRequestForServerSpan();
final ExtendedResponse response = generateExtendedResponseForServerSpan();

transaction.requestInitialized(request, response);
TracedMethod tracedMethod = transaction.getTracedMethod();
List<LinkData> immutableLinks = this.links == null ? Collections.emptyList() : Collections.unmodifiableList(this.links);
return onStart(new ExitTracerSpan((ExitTracer) tracedMethod, instrumentationLibraryInfo, spanKind, spanName,
parentSpanContext, sharedState.getResource(), sharedState.getClock(), attributes, endHandler, immutableLinks, totalNumberOfLinksAdded, startEpochNanos));
}

private ExtendedRequest generateExtendedRequestForServerSpan() {
return new ExtendedRequest() {

@Override
public String getRequestURI() {
Expand Down Expand Up @@ -265,54 +303,23 @@ public HeaderType getHeaderType() {

@Override
public String getHeader(String name) {
if ("User-Agent".equalsIgnoreCase(name)) {
return (String) attributes.get(attributeMapper.findProperOtelKey(SpanKind.SERVER, AttributeType.Host, attributes.keySet()));
}
// TODO is it possible to get the newrelic DT header from OTel??? It doesn't seem so.
if (NEWRELIC.equalsIgnoreCase(name)) {
return null;
}
return null;
return getHeaderFromAttributes(name);
}

@Override
public List<String> getHeaders(String name) {
if (name.isEmpty()) {
return Collections.emptyList();
}
List<String> headers = new ArrayList<>();

if (W3C_TRACESTATE.equalsIgnoreCase(name)) {
Map<String, String> traceState = parentSpanContext.getTraceState().asMap();
StringBuilder tracestateStringBuilder = new StringBuilder();
// Build full tracestate header incase there are multiple vendors
for (Map.Entry<String, String> entry : traceState.entrySet()) {
if (tracestateStringBuilder.length() == 0) {
tracestateStringBuilder.append(entry.toString());
} else {
tracestateStringBuilder.append(",").append(entry.toString());
}
}
headers.add(tracestateStringBuilder.toString());
return headers;
}
if (W3C_TRACEPARENT.equalsIgnoreCase(name)) {
String traceParent = W3CTraceParentHeader.create(parentSpanContext);
if (!traceParent.isEmpty()) {
headers.add(traceParent);
return headers;
}
}
return headers;
return getHeadersFromAttributes(name);
}

@Override
public String getMethod() {
return (String) attributes.get(attributeMapper.findProperOtelKey(SpanKind.SERVER, AttributeType.Method, attributes.keySet()));
}
};
}

final ExtendedResponse response = new ExtendedResponse() {
private ExtendedResponse generateExtendedResponseForServerSpan() {
return new ExtendedResponse() {

@Override
public int getStatus() throws Exception {
Expand Down Expand Up @@ -345,15 +352,122 @@ public long getContentLength() {
return 0;
}
};
transaction.requestInitialized(request, response);
TracedMethod tracedMethod = transaction.getTracedMethod();
List<LinkData> immutableLinks = this.links == null ? Collections.emptyList() : Collections.unmodifiableList(this.links);
return onStart(new ExitTracerSpan((ExitTracer) tracedMethod, instrumentationLibraryInfo, spanKind, spanName,
parentSpanContext, sharedState.getResource(), sharedState.getClock(), attributes, endHandler, immutableLinks, totalNumberOfLinksAdded, startEpochNanos));
}

private void acceptMessageQueueNotSampledHeaderIfNecessary(Transaction tx) {
if (sendMessageQueueNotSampledHeader) tx.acceptDistributedTraceHeaders(TransportType.Queue, generateDtHeaders());
}

private void insertMessageQueueNotSampledHeaderIfNecessary() {
Transaction tx = AgentBridge.getAgent().getTransaction(false);
if (tx == null) return;

// ask the agent to insert the header, if necessary
tx.insertDistributedTraceHeaders(generateDtHeaders());
// if it did, then set the corresponding attribute on the span
if (attributes.containsKey("nrns")) {
attributes.put("new_relic_not_sampled", "");
}
}

private Headers generateDtHeaders() {
return new Headers() {
@Override
public HeaderType getHeaderType() {
return HeaderType.MESSAGE;
}

@Override
public String getHeader(String name) {
String result = getHeaderFromAttributes(name);
return result;
}

@Override
public Collection<String> getHeaders(String name) {
Collection<String> result = getHeadersFromAttributes(name);
return result;
}

@Override
public void setHeader(String name, String value) {
attributes.put(name, value);
}

@Override
public void addHeader(String name, String value) {
attributes.put(name, value);
}

@Override
public Collection<String> getHeaderNames() {
return null;
}

@Override
public boolean containsHeader(String name) {
return attributes.containsKey(name);
}
};
}

private String getHeaderFromAttributes(String name) {
if ("User-Agent".equalsIgnoreCase(name)) {
return (String) attributes.get(attributeMapper.findProperOtelKey(SpanKind.SERVER, AttributeType.Host, attributes.keySet()));
}
// yes, the 2 if statements are purposefully cross-matched
// if we are looking for nrns, then we'll want to know if it already has new_relic_not_sampled
// and vice versa
if ("nrns".equalsIgnoreCase(name)) {
return attributes.containsKey("new_relic_not_sampled") ? "" : null;
}
if ("new_relic_not_sampled".equalsIgnoreCase(name)) {
return attributes.containsKey("nrns") ? "" : null;
}
// it is not currently possible to get the newrelic DT header from OTel
if (NEWRELIC.equalsIgnoreCase(name)) {
return null;
}
return null;
}

private List<String> getHeadersFromAttributes(String name) {
if (name.isEmpty()) {
return Collections.emptyList();
}
List<String> headers = new ArrayList<>();

if ("nrns".equalsIgnoreCase(name) &&
attributes.containsKey("new_relic_not_sampled") &&
sendMessageQueueNotSampledHeader) {
headers.add(""); // intentionally blank
return headers;
}
if (W3C_TRACESTATE.equalsIgnoreCase(name)) {
Map<String, String> traceState = parentSpanContext.getTraceState().asMap();
StringBuilder tracestateStringBuilder = new StringBuilder();
// Build full tracestate header incase there are multiple vendors
for (Map.Entry<String, String> entry : traceState.entrySet()) {
if (tracestateStringBuilder.length() == 0) {
tracestateStringBuilder.append(entry.toString());
} else {
tracestateStringBuilder.append(",").append(entry.toString());
}
}
headers.add(tracestateStringBuilder.toString());
return headers;
}
if (W3C_TRACEPARENT.equalsIgnoreCase(name)) {
String traceParent = W3CTraceParentHeader.create(parentSpanContext);
if (!traceParent.isEmpty()) {
headers.add(traceParent);
return headers;
}
}
return headers;
}

Span onStart(ReadWriteSpan span) {
// FIXME
Context parent = Context.current();
if (sharedState.getActiveSpanProcessor().isStartRequired()) {
sharedState.getActiveSpanProcessor().onStart(parent, span);
Expand Down
16 changes: 15 additions & 1 deletion ...tension-autoconfigure-1.59.0/src/main/java/io/opentelemetry/sdk/trace/ExitTracerSpan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import com.newrelic.api.agent.DatastoreParameters;
import com.newrelic.api.agent.HttpParameters;
import com.newrelic.api.agent.NewRelic;
import com.newrelic.api.agent.Segment;
import com.newrelic.api.agent.Token;
import com.nr.agent.instrumentation.utils.AttributesHelper;
import com.nr.agent.instrumentation.utils.span.AttributeMapper;
Expand Down Expand Up @@ -85,6 +86,7 @@ public class ExitTracerSpan implements ReadWriteSpan {
.collect(Collectors.toSet()));

final ExitTracer tracer;
private final Segment segment;
private final SpanKind spanKind;
private final InstrumentationScopeInfo instrumentationScopeInfo;
private final Map<String, Object> attributes;
Expand Down Expand Up @@ -112,7 +114,15 @@ public class ExitTracerSpan implements ReadWriteSpan {
ExitTracerSpan(ExitTracer tracer, InstrumentationScopeInfo instrumentationScopeInfo, SpanKind spanKind, String spanName, SpanContext parentSpanContext,
Resource resource, Clock tracerClock, Map<String, Object> attributes, Consumer<ExitTracerSpan> onEnd, List<LinkData> links,
int totalNumberOfLinksAdded, long userStartEpochNanos) {
this(tracer, instrumentationScopeInfo, spanKind, spanName, parentSpanContext,
resource, tracerClock, attributes, onEnd, links, totalNumberOfLinksAdded, userStartEpochNanos, null);
}

ExitTracerSpan(ExitTracer tracer, InstrumentationScopeInfo instrumentationScopeInfo, SpanKind spanKind, String spanName, SpanContext parentSpanContext,
Resource resource, Clock tracerClock, Map<String, Object> attributes, Consumer<ExitTracerSpan> onEnd, List<LinkData> links,
int totalNumberOfLinksAdded, long userStartEpochNanos, Segment segment) {
this.tracer = tracer;
this.segment = segment;
this.spanKind = spanKind;
this.spanName = spanName;
this.parentSpanContext = parentSpanContext;
Expand Down Expand Up @@ -287,7 +297,11 @@ public void end() {
copySpanLinksToTracer(links);
List<EventData> immutableEvents = this.events == null ? Collections.emptyList() : Collections.unmodifiableList(this.events);
copySpanEventsToTracer(immutableEvents);
tracer.finish();
if (segment != null) {
segment.end();
} else {
tracer.finish();
}
endEpochNanos = System.nanoTime();
ended = true;
onEnd.accept(this);
Expand Down
Loading
Loading